CASSANDRA-21552 Force repair should ignore min_repair_interval

CHANGES.txt

@@ -1,4 +1,5 @@
 5.0.10
+ * Force repair should ignore min_repair_interval (CASSANDRA-21552)
  * Propagate trickle_fsync settings to compressed SSTable writers (CASSANDRA-21487)
  * Allow DatabaseDescriptor.setCompressedReadAheadBufferSizeInKb(0) to disable read-ahead buffer (CASSANDRA-21522)
  * Return CorruptSSTableException if chunk metadata and file size are out of sync (CASSANDRA-21519)

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

@@ -195,8 +195,9 @@ public void repair(AutoRepairConfig.RepairType repairType)
             //consistency level to use for local query
             UUID myId = Gossiper.instance.getHostId(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;
             }
@@ -410,6 +411,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
+     */
     private boolean tooSoonToRunRepair(AutoRepairConfig.RepairType repairType, AutoRepairState repairState, AutoRepairConfig config, UUID myId)
     {
         if (repairState.getLastRepairTime() == 0)

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

@@ -174,11 +174,16 @@ public class AutoRepairUtils
     "SELECT %s FROM %s.%s WHERE %s = ? AND %s = ?", COL_REPAIR_FINISH_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 selectForceRepairForNode;
     static ModificationStatement addPriorityHost;
     static ModificationStatement insertNewRepairHistoryStatement;
     static ModificationStatement recordStartRepairHistoryStatement;
@@ -204,6 +209,8 @@ public static void setup()
                                                                                                               .forInternalCalls());
         selectLastRepairTimeForNode = (SelectStatement) QueryProcessor.getStatement(SELECT_LAST_REPAIR_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
@@ -404,6 +411,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 getLastRepairTimeForNode(RepairType repairType, UUID hostId)
     {
         ResultMessage.Rows rows = selectLastRepairTimeForNode.execute(QueryState.forInternalCalls(),

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

@@ -20,6 +20,7 @@
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.UUID;
 
 import org.junit.Before;
 import org.junit.BeforeClass;
@@ -29,16 +30,22 @@
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.config.DurationSpec;
 import org.apache.cassandra.cql3.CQLTester;
+import org.apache.cassandra.cql3.QueryProcessor;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.schema.KeyspaceMetadata;
 import org.apache.cassandra.schema.KeyspaceParams;
 import org.apache.cassandra.schema.ReplicationParams;
+import org.apache.cassandra.schema.SchemaConstants;
 import org.apache.cassandra.repair.autorepair.AutoRepairConfig.RepairType;
 import org.apache.cassandra.schema.SchemaTestUtil;
+import org.apache.cassandra.schema.SystemDistributedKeyspace;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.FBUtilities;
 
 import static org.apache.cassandra.Util.setAutoRepairEnabled;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 
 /**
@@ -159,4 +166,136 @@ 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.getLastRepairTimeForNode(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.getLastRepairTimeForNode(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());
+    }
 }