CASSANDRA-21552 Force repair should ignore min_repair_interval

CHANGES.txt

@@ -72,6 +72,7 @@ Merged from 6.0:
  * Introduce minimum_threshold for data resurrection startup check (CASSANDRA-21293)
  * Synchronously publish changes to local gossip state following metadata updates (CASSANDRA-21239)
 Merged from 5.0:
+ * Force repair should ignore min_repair_interval (CASSANDRA-21552)
  * Propagate trickle_fsync settings to compressed SSTable writers (CASSANDRA-21487)
  * Allow setCompressedReadAheadBufferSizeInKb(0) to disable read-ahead buffer (CASSANDRA-21522)
  * Fix ThreadLocalReadAheadBuffer#fill() to throw a CorruptBlockException if chunk metadata and file size are out of sync (CASSANDRA-21519)

src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java

@@ -188,8 +188,9 @@ public void repair(AutoRepairConfig.RepairType repairType)
             //consistency level to use for local query
             UUID myId = StorageService.instance.getHostIdForEndpoint(FBUtilities.getBroadcastAddressAndPort());
 
-            // If it's too soon to run repair, don't bother checking if it's our turn.
-            if (tooSoonToRunRepair(repairType, repairState, config, myId))
+            // Skip this repair cycle if the minimum interval since the last repair has not elapsed,
+            // unless force repair is set for this node, which bypasses the interval check
+            if (shouldSkipRepairDueToInterval(repairType, repairState, config, myId))
             {
                 return;
             }
@@ -404,6 +405,31 @@ else if (retryCount < config.getRepairMaxRetries(repairType))
         }
     }
 
+    /**
+     * Determines whether the repair should be skipped due to the minimum repair interval.
+     * Force repair bypasses the interval check to ensure immediate repair trigger.
+     *
+     * @return true if repair should be skipped, false if it should proceed
+     */
+    @VisibleForTesting
+    boolean shouldSkipRepairDueToInterval(AutoRepairConfig.RepairType repairType, AutoRepairState repairState, AutoRepairConfig config, UUID myId)
+    {
+        if (AutoRepairUtils.isForceRepairSetForNode(repairType, myId))
+        {
+            logger.info("Force repair is set for this node, bypassing min_repair_interval check");
+            return false;
+        }
+        return tooSoonToRunRepair(repairType, repairState, config, myId);
+    }
+
+    /**
+     * Determines whether it is too soon to run a repair based on the minimum repair interval configured
+     * for the given repair type. If no last repair time is recorded in the state, it fetches the most
+     * recent repair time for this node from the database.
+     *
+     * @return true if the elapsed time since the last repair is less than the configured
+     *         minimum interval
+     */
     @VisibleForTesting
     boolean tooSoonToRunRepair(AutoRepairConfig.RepairType repairType, AutoRepairState repairState, AutoRepairConfig config, UUID myId)
     {

src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java

@@ -181,12 +181,17 @@ public class AutoRepairUtils
     "SELECT %s FROM %s.%s WHERE %s = ? AND %s = ?", COL_REPAIR_START_TS, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME,
     SystemDistributedKeyspace.AUTO_REPAIR_HISTORY, COL_REPAIR_TYPE, COL_HOST_ID);
 
+    final static String SELECT_FORCE_REPAIR_FOR_NODE = String.format(
+    "SELECT %s FROM %s.%s WHERE %s = ? AND %s = ?", COL_FORCE_REPAIR, SchemaConstants.DISTRIBUTED_KEYSPACE_NAME,
+    SystemDistributedKeyspace.AUTO_REPAIR_HISTORY, COL_REPAIR_TYPE, COL_HOST_ID);
+
     static ModificationStatement delStatementRepairHistory;
     static SelectStatement selectStatementRepairHistory;
     static ModificationStatement delStatementPriorityStatus;
     static SelectStatement selectStatementRepairPriority;
     static SelectStatement selectLastRepairTimeForNode;
     static SelectStatement selectLastRepairStartTimeForNode;
+    static SelectStatement selectForceRepairForNode;
     static ModificationStatement addPriorityHost;
     static ModificationStatement insertNewRepairHistoryStatement;
     static ModificationStatement recordStartRepairHistoryStatement;
@@ -214,6 +219,8 @@ public static void setup()
                                                                                                                       .forInternalCalls());
         selectLastRepairStartTimeForNode = (SelectStatement) QueryProcessor.getStatement(SELECT_LAST_REPAIR_START_TIME_FOR_NODE, ClientState
                                                                                                                                  .forInternalCalls());
+        selectForceRepairForNode = (SelectStatement) QueryProcessor.getStatement(SELECT_FORCE_REPAIR_FOR_NODE, ClientState
+                                                                                                               .forInternalCalls());
         delStatementPriorityStatus = (ModificationStatement) QueryProcessor.getStatement(DEL_REPAIR_PRIORITY, ClientState
                                                                                                               .forInternalCalls());
         addPriorityHost = (ModificationStatement) QueryProcessor.getStatement(ADD_PRIORITY_HOST, ClientState
@@ -485,6 +492,29 @@ public static void setForceRepair(RepairType repairType, UUID hostId)
         logger.info("Set force repair repair type: {}, node: {}", repairType, hostId);
     }
 
+    /**
+     * Check if force repair is set for the given node.
+     *
+     * @param repairType the repair type to check
+     * @param hostId the host id to check
+     * @return true if force repair is set for this node
+     */
+    public static boolean isForceRepairSetForNode(RepairType repairType, UUID hostId)
+    {
+        ResultMessage.Rows rows = selectForceRepairForNode.execute(QueryState.forInternalCalls(),
+                                                                   QueryOptions.forInternalCalls(internalQueryCL,
+                                                                                                 Lists.newArrayList(
+                                                                                                 ByteBufferUtil.bytes(repairType.toString()),
+                                                                                                 ByteBufferUtil.bytes(hostId))),
+                                                                   Dispatcher.RequestTime.forImmediateExecution());
+        UntypedResultSet result = UntypedResultSet.create(rows.result);
+        if (result.isEmpty())
+            return false;
+
+        UntypedResultSet.Row one = result.one();
+        return one.has(COL_FORCE_REPAIR) && one.getBoolean(COL_FORCE_REPAIR);
+    }
+
     public static long getLastRepairFinishTimeForNode(RepairType repairType, UUID hostId)
     {
         ResultMessage.Rows rows = selectLastRepairTimeForNode.execute(QueryState.forInternalCalls(),

test/unit/org/apache/cassandra/repair/autorepair/AutoRepairTest.java

@@ -171,6 +171,138 @@ else if (ks.getName().equals(ksname2))
         }
     }
 
+    @Test
+    public void testForceRepairBypassesMinRepairInterval()
+    {
+        RepairType repairType = RepairType.FULL;
+        UUID myId = StorageService.instance.getHostIdForEndpoint(FBUtilities.getBroadcastAddressAndPort());
+        long now = System.currentTimeMillis();
+
+        // Truncate history table to start fresh
+        QueryProcessor.executeInternal(String.format(
+            "TRUNCATE %s.%s",
+            SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY));
+
+        // Seed auto_repair_history directly with a recently completed repair and force_repair=true
+        QueryProcessor.executeInternal(String.format(
+            "INSERT INTO %s.%s (repair_type, host_id, repair_start_ts, repair_finish_ts, force_repair) VALUES (?, ?, ?, ?, true)",
+            SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY),
+            repairType.toString(), myId, new java.util.Date(now - 1000), new java.util.Date(now));
+
+        // Verify force repair is detected
+        assertTrue(AutoRepairUtils.isForceRepairSetForNode(repairType, myId));
+
+        AutoRepairConfig config = DatabaseDescriptor.getAutoRepairConfig();
+        AutoRepairState repairState = RepairType.getAutoRepairState(repairType, config);
+
+        // Even though min_repair_interval hasn't passed, shouldSkipRepairDueToInterval returns false
+        // because force repair is set
+        assertFalse(AutoRepair.instance.shouldSkipRepairDueToInterval(repairType, repairState, config, myId));
+    }
+
+    @Test
+    public void testShouldSkipRepairDueToIntervalWithoutForceRepair()
+    {
+        RepairType repairType = RepairType.FULL;
+        UUID myId = StorageService.instance.getHostIdForEndpoint(FBUtilities.getBroadcastAddressAndPort());
+        long now = System.currentTimeMillis();
+
+        // Truncate history table to start fresh
+        QueryProcessor.executeInternal(String.format(
+            "TRUNCATE %s.%s",
+            SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY));
+
+        // Seed auto_repair_history directly with a recently completed repair and force_repair=false
+        QueryProcessor.executeInternal(String.format(
+            "INSERT INTO %s.%s (repair_type, host_id, repair_start_ts, repair_finish_ts, force_repair) VALUES (?, ?, ?, ?, false)",
+            SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY),
+            repairType.toString(), myId, new java.util.Date(now - 1000), new java.util.Date(now));
+
+        // Verify force repair is NOT set
+        assertFalse(AutoRepairUtils.isForceRepairSetForNode(repairType, myId));
+
+        AutoRepairConfig config = DatabaseDescriptor.getAutoRepairConfig();
+        AutoRepairState repairState = RepairType.getAutoRepairState(repairType, config);
+
+        // Without force repair, should skip because min_repair_interval hasn't passed
+        assertTrue(AutoRepair.instance.shouldSkipRepairDueToInterval(repairType, repairState, config, myId));
+    }
+
+    @Test
+    public void testIsForceRepairSetForNodeReturnsFalseWhenNotSet()
+    {
+        RepairType repairType = RepairType.FULL;
+        UUID myId = StorageService.instance.getHostIdForEndpoint(FBUtilities.getBroadcastAddressAndPort());
+        long now = System.currentTimeMillis();
+
+        // Truncate history table to start fresh
+        QueryProcessor.executeInternal(String.format(
+            "TRUNCATE %s.%s",
+            SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY));
+
+        // Seed auto_repair_history directly without setting force_repair
+        QueryProcessor.executeInternal(String.format(
+            "INSERT INTO %s.%s (repair_type, host_id, repair_start_ts, repair_finish_ts) VALUES (?, ?, ?, ?)",
+            SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY),
+            repairType.toString(), myId, new java.util.Date(now - 1000), new java.util.Date(now));
+
+        // Verify force repair is not set
+        assertFalse(AutoRepairUtils.isForceRepairSetForNode(repairType, myId));
+    }
+
+    @Test
+    public void testIsForceRepairSetForNodeReturnsFalseWhenNoHistory()
+    {
+        RepairType repairType = RepairType.FULL;
+        UUID myId = StorageService.instance.getHostIdForEndpoint(FBUtilities.getBroadcastAddressAndPort());
+
+        // Truncate history table to start fresh
+        QueryProcessor.executeInternal(String.format(
+            "TRUNCATE %s.%s",
+            SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY));
+
+        // Verify force repair is not set when no history exists
+        assertFalse(AutoRepairUtils.isForceRepairSetForNode(repairType, myId));
+    }
+
+    @Test
+    public void testForceRepairBypassesMinRepairIntervalEndToEnd()
+    {
+        RepairType repairType = RepairType.FULL;
+        UUID myId = StorageService.instance.getHostIdForEndpoint(FBUtilities.getBroadcastAddressAndPort());
+        long now = System.currentTimeMillis();
+
+        DurationSpec.LongSecondsBound repairTaskMinDuration = DatabaseDescriptor.getAutoRepairConfig().getRepairTaskMinDuration();
+        // Ensure repair tasks don't artificially sleep
+        DatabaseDescriptor.getAutoRepairConfig().setRepairTaskMinDuration("0s");
+
+        // Truncate history table to start fresh
+        QueryProcessor.executeInternal(String.format(
+            "TRUNCATE %s.%s",
+            SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY));
+
+        // Insert a recently completed repair so tooSoonToRunRepair would normally block
+        QueryProcessor.executeInternal(String.format(
+            "INSERT INTO %s.%s (repair_type, host_id, repair_start_ts, repair_finish_ts, force_repair) VALUES (?, ?, ?, ?, true)",
+            SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY),
+            repairType.toString(), myId, new java.util.Date(now - 1000), new java.util.Date(now));
+
+        // Record the finish time before repair runs
+        long finishTimeBefore = AutoRepairUtils.getLastRepairFinishTimeForNode(repairType, myId);
+
+        // Invoke the full repair path; with force repair set, the interval check is bypassed
+        AutoRepair.instance.repair(repairType);
+
+        // Verify that repair_finish_ts has advanced, proving repair actually ran
+        long finishTimeAfter = AutoRepairUtils.getLastRepairFinishTimeForNode(repairType, myId);
+        assertTrue("repair_finish_ts should advance after force repair runs, but was "
+                   + finishTimeBefore + " -> " + finishTimeAfter,
+                   finishTimeAfter > finishTimeBefore);
+
+        // Restore original value
+        DatabaseDescriptor.getAutoRepairConfig().setRepairTaskMinDuration(repairTaskMinDuration.toString());
+    }
+
     @Test
     public void testTooSoonToRunRepairAllowsResumeOfInProgressRepair()
     {