diff --git a/CHANGES.txt b/CHANGES.txt index 5bbd8cba36fc..5c665a3fe5c5 100644 --- a/CHANGES.txt +++ b/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) diff --git a/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java b/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java index f0e31c4e590d..f94481f24b11 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java +++ b/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) { diff --git a/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java b/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java index e25da5115751..850cbaa00732 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java +++ b/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(), diff --git a/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairTest.java b/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairTest.java index 5a47ce398e6a..a37a6d7357d1 100644 --- a/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairTest.java +++ b/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() {