diff --git a/CHANGES.txt b/CHANGES.txt index ab306d961a3..b029925305b 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -1,4 +1,5 @@ 5.0.10 + * Failed force repair should clear force repair flag (CASSANDRA-21663) * Fix concurrent SAI vector inserts failing once jvector's per-graph pool limit is exceeded (CASSANDRA-21644) * Let the commit log allocator observe the shutdown state (CASSANDRA-21616) * Unwrap LongType properly when calculating min/max terms in V1SSTableIndex (CASSANDRA-21635) diff --git a/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java b/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java index 05f9c46f75b..2c4aa5f503f 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java +++ b/src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java @@ -205,70 +205,94 @@ public void repair(AutoRepairConfig.RepairType repairType) RepairTurn turn = AutoRepairUtils.myTurnToRunRepair(repairType, myId); if (turn == MY_TURN || turn == MY_TURN_DUE_TO_PRIORITY || turn == MY_TURN_FORCE_REPAIR) { - repairState.recordTurn(turn); - repairState.setBytesAlreadyRepaired(0L); - repairState.setKeyspaceRepairPlansAlreadyRepaired(0); - // For normal auto repair, we will use primary range only repairs (Repair with -pr option). - // For some cases, we may set the auto_repair_primary_token_range_only flag to false then we will do repair - // without -pr. We may also do force repair for certain node that we want to repair all the data on one node - // When doing force repair, we want to repair without -pr. - boolean primaryRangeOnly = config.getRepairPrimaryTokenRangeOnly(repairType) - && turn != MY_TURN_FORCE_REPAIR; - - long startTimeInMillis = timeFunc.get(); - logger.info("My host id: {}, my turn to run repair...repair primary-ranges only? {}", myId, - config.getRepairPrimaryTokenRangeOnly(repairType)); - AutoRepairUtils.updateStartAutoRepairHistory(repairType, myId, timeFunc.get(), turn); - - repairState.setRepairKeyspaceCount(0); - repairState.setRepairInProgress(true); - repairState.setTotalTablesConsideredForRepair(0); - repairState.setTotalMVTablesConsideredForRepair(0); - - CollectedRepairStats collectedRepairStats = new CollectedRepairStats(); - - List keyspaces = new ArrayList<>(); - Keyspace.all().forEach(keyspaces::add); - // Filter out keyspaces and tables to repair and group into a map by keyspace. - Map> keyspacesAndTablesToRepair = new LinkedHashMap<>(); - for (Keyspace keyspace : keyspaces) + // When this run was triggered by a force repair, consume the force-repair flag in the + // finally below, whether the repair succeeds or throws. Otherwise a failed force repair + // leaves force_repair=true and, because it bypasses min_repair_interval, the node would + // re-run repair on every subsequent cycle until it happens to succeed. A normal repair + // (MY_TURN / MY_TURN_DUE_TO_PRIORITY) must never clear the flag, so a force repair that + // was requested while a normal repair is running is still honored afterwards. + boolean forceRepairTurn = turn == MY_TURN_FORCE_REPAIR; + try { - if (!AutoRepairUtils.shouldConsiderKeyspace(keyspace)) + repairState.recordTurn(turn); + repairState.setBytesAlreadyRepaired(0L); + repairState.setKeyspaceRepairPlansAlreadyRepaired(0); + // For normal auto repair, we will use primary range only repairs (Repair with -pr option). + // For some cases, we may set the auto_repair_primary_token_range_only flag to false then we will do repair + // without -pr. We may also do force repair for certain node that we want to repair all the data on one node + // When doing force repair, we want to repair without -pr. + boolean primaryRangeOnly = config.getRepairPrimaryTokenRangeOnly(repairType) + && turn != MY_TURN_FORCE_REPAIR; + + long startTimeInMillis = timeFunc.get(); + logger.info("My host id: {}, my turn to run repair...repair primary-ranges only? {}", myId, + config.getRepairPrimaryTokenRangeOnly(repairType)); + AutoRepairUtils.updateStartAutoRepairHistory(repairType, myId, timeFunc.get(), turn); + + repairState.setRepairKeyspaceCount(0); + repairState.setRepairInProgress(true); + repairState.setTotalTablesConsideredForRepair(0); + repairState.setTotalMVTablesConsideredForRepair(0); + + CollectedRepairStats collectedRepairStats = new CollectedRepairStats(); + + List keyspaces = new ArrayList<>(); + Keyspace.all().forEach(keyspaces::add); + // Filter out keyspaces and tables to repair and group into a map by keyspace. + Map> keyspacesAndTablesToRepair = new LinkedHashMap<>(); + for (Keyspace keyspace : keyspaces) { - continue; + if (!AutoRepairUtils.shouldConsiderKeyspace(keyspace)) + { + continue; + } + List tablesToBeRepairedList = retrieveTablesToBeRepaired(keyspace, config, repairType, repairState, collectedRepairStats); + keyspacesAndTablesToRepair.put(keyspace.getName(), tablesToBeRepairedList); } - List tablesToBeRepairedList = retrieveTablesToBeRepaired(keyspace, config, repairType, repairState, collectedRepairStats); - keyspacesAndTablesToRepair.put(keyspace.getName(), tablesToBeRepairedList); - } - // Separate out the keyspaces and tables to repair based on their priority, with each repair plan representing a uniquely occuring priority. - List repairPlans = PrioritizedRepairPlan.build(keyspacesAndTablesToRepair, repairType, shuffleFunc, primaryRangeOnly); - repairState.updateRepairScheduleStatistics(repairPlans); + // Separate out the keyspaces and tables to repair based on their priority, with each repair plan representing a uniquely occuring priority. + List repairPlans = PrioritizedRepairPlan.build(keyspacesAndTablesToRepair, repairType, shuffleFunc, primaryRangeOnly); + repairState.updateRepairScheduleStatistics(repairPlans); - // calculate the repair assignments for each priority:keyspace. - Iterator repairAssignmentsIterator = config.getTokenRangeSplitterInstance(repairType).getRepairAssignments(primaryRangeOnly, repairPlans); + // calculate the repair assignments for each priority:keyspace. + Iterator repairAssignmentsIterator = config.getTokenRangeSplitterInstance(repairType).getRepairAssignments(primaryRangeOnly, repairPlans); - int keyspaceRepairAssignmentsAlreadyRepaired = 0; - while (repairAssignmentsIterator.hasNext()) - { - KeyspaceRepairAssignments repairAssignments = repairAssignmentsIterator.next(); - List assignments = repairAssignments.getRepairAssignments(); - if (assignments.isEmpty()) + int keyspaceRepairAssignmentsAlreadyRepaired = 0; + while (repairAssignmentsIterator.hasNext()) { + KeyspaceRepairAssignments repairAssignments = repairAssignmentsIterator.next(); + List assignments = repairAssignments.getRepairAssignments(); + if (assignments.isEmpty()) + { + keyspaceRepairAssignmentsAlreadyRepaired++; + logger.info("Skipping repairs for priorityBucket={} for keyspace={} since it yielded no assignments", repairAssignments.getPriority(), repairAssignments.getKeyspaceName()); + continue; + } + + logger.info("Submitting repairs for priorityBucket={} for keyspace={} with assignmentCount={} and keyspaceRepairAssignmentsAlreadyRepaired={}/{}", + repairAssignments.getPriority(), repairAssignments.getKeyspaceName(), repairAssignments.getRepairAssignments().size(), + keyspaceRepairAssignmentsAlreadyRepaired, repairState.getTotalKeyspaceRepairPlansToRepair()); + repairKeyspace(repairType, primaryRangeOnly, repairAssignments.getKeyspaceName(), repairAssignments.getRepairAssignments(), collectedRepairStats); keyspaceRepairAssignmentsAlreadyRepaired++; - logger.info("Skipping repairs for priorityBucket={} for keyspace={} since it yielded no assignments", repairAssignments.getPriority(), repairAssignments.getKeyspaceName()); - continue; + repairState.setKeyspaceRepairPlansAlreadyRepaired(keyspaceRepairAssignmentsAlreadyRepaired); } - logger.info("Submitting repairs for priorityBucket={} for keyspace={} with assignmentCount={} and keyspaceRepairAssignmentsAlreadyRepaired={}/{}", - repairAssignments.getPriority(), repairAssignments.getKeyspaceName(), repairAssignments.getRepairAssignments().size(), - keyspaceRepairAssignmentsAlreadyRepaired, repairState.getTotalKeyspaceRepairPlansToRepair()); - repairKeyspace(repairType, primaryRangeOnly, repairAssignments.getKeyspaceName(), repairAssignments.getRepairAssignments(), collectedRepairStats); - keyspaceRepairAssignmentsAlreadyRepaired++; - repairState.setKeyspaceRepairPlansAlreadyRepaired(keyspaceRepairAssignmentsAlreadyRepaired); + cleanupAndUpdateStats(turn, repairType, repairState, myId, startTimeInMillis, collectedRepairStats); + } + finally + { + // Only consume the flag when this run was itself triggered by a force repair. + // A normal repair must never clear it. clearForceRepair sets force_repair=false only; + // unlike updateFinishAutoRepairHistory it deliberately does NOT advance repair_finish_ts + // (the timestamp of the last SUCCESSFUL repair, which drives min_repair_interval). On + // failure this releases the flag so the node stops bypassing min_repair_interval and + // re-running repair every cycle, while the "last successful repair" time stays truthful + // instead of being bumped to now. + if (forceRepairTurn) + { + AutoRepairUtils.clearForceRepair(repairType, myId); + } } - - cleanupAndUpdateStats(turn, repairType, repairState, myId, startTimeInMillis, collectedRepairStats); } else { diff --git a/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java b/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java index 1c788fe6600..b8efc4999ae 100644 --- a/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java +++ b/src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java @@ -155,10 +155,13 @@ public class AutoRepairUtils , SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY, COL_REPAIR_START_TS, COL_REPAIR_TYPE, COL_HOST_ID); + // NOTE: this deliberately updates only repair_finish_ts and does NOT clear force_repair. The + // force-repair flag is consumed exclusively by clearForceRepair, and only when the run was itself + // triggered by a force repair, so a normal repair never clears a pending force-repair request. final static String RECORD_FINISH_REPAIR_HISTORY = String.format( - "UPDATE %s.%s SET %s= ?, %s=false WHERE %s = ? AND %s = ?" + "UPDATE %s.%s SET %s= ? WHERE %s = ? AND %s = ?" , SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY, COL_REPAIR_FINISH_TS, - COL_FORCE_REPAIR, COL_REPAIR_TYPE, COL_HOST_ID); + COL_REPAIR_TYPE, COL_HOST_ID); final static String CLEAR_DELETE_HOSTS = String.format( "UPDATE %s.%s SET %s= {} WHERE %s = ? AND %s = ?" @@ -170,6 +173,11 @@ public class AutoRepairUtils , SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY, COL_FORCE_REPAIR, COL_REPAIR_TYPE, COL_HOST_ID); + final static String CLEAR_FORCE_REPAIR = String.format( + "UPDATE %s.%s SET %s=false WHERE %s = ? AND %s = ?" + , SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY, COL_FORCE_REPAIR, + COL_REPAIR_TYPE, COL_HOST_ID); + final static String SELECT_LAST_REPAIR_TIME_FOR_NODE = String.format( "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); @@ -191,6 +199,7 @@ public class AutoRepairUtils static ModificationStatement addHostIDToDeleteHostsStatement; static ModificationStatement clearDeleteHostsStatement; static ModificationStatement setForceRepairStatement; + static ModificationStatement clearForceRepairStatement; static ConsistencyLevel internalQueryCL; public enum RepairTurn @@ -225,6 +234,8 @@ public static void setup() .forInternalCalls()); setForceRepairStatement = (ModificationStatement) QueryProcessor.getStatement(SET_FORCE_REPAIR, ClientState .forInternalCalls()); + clearForceRepairStatement = (ModificationStatement) QueryProcessor.getStatement(CLEAR_FORCE_REPAIR, ClientState + .forInternalCalls()); clearDeleteHostsStatement = (ModificationStatement) QueryProcessor.getStatement(CLEAR_DELETE_HOSTS, ClientState .forInternalCalls()); delStatementRepairHistory = (ModificationStatement) QueryProcessor.getStatement(DEL_AUTO_REPAIR_HISTORY, ClientState @@ -411,6 +422,27 @@ public static void setForceRepair(RepairType repairType, UUID hostId) logger.info("Set force repair repair type: {}, node: {}", repairType, hostId); } + /** + * Clear the force repair flag for the given node. + *

+ * This is called once a repair run for the node completes, whether it succeeded or failed, so that + * a force repair is consumed exactly once. Unlike {@link #updateFinishAutoRepairHistory}, it does not + * advance {@code repair_finish_ts}: a failed forced repair must not be recorded as a successful one, + * but it must also not leave {@code force_repair=true}, which (since force repair bypasses + * min_repair_interval) would make the node re-run repair on every subsequent cycle. + * + * @param repairType the repair type + * @param hostId the host id whose force repair flag should be cleared + */ + public static void clearForceRepair(RepairType repairType, UUID hostId) + { + clearForceRepairStatement.execute(QueryState.forInternalCalls(), + QueryOptions.forInternalCalls(internalQueryCL, + Lists.newArrayList(ByteBufferUtil.bytes(repairType.toString()), + ByteBufferUtil.bytes(hostId))), + Dispatcher.RequestTime.forImmediateExecution()); + } + /** * Check if force repair is set for the given node. * diff --git a/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairTest.java b/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairTest.java index a9ea3c61d89..9c5b9a2b7c9 100644 --- a/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairTest.java +++ b/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairTest.java @@ -300,7 +300,51 @@ public void testForceRepairBypassesMinRepairIntervalEndToEnd() finishTimeAfter > finishTimeBefore); assertTrue(AutoRepair.instance.shouldSkipRepairDueToInterval(repairType, repairState, config, myId)); + // The force repair is one-shot: after a run, force_repair must be cleared so the node + // resumes honoring min_repair_interval instead of force-repairing every cycle. + assertFalse("force_repair must be cleared after a forced repair run", + AutoRepairUtils.isForceRepairSetForNode(repairType, myId)); + // Restore original value DatabaseDescriptor.getAutoRepairConfig().setRepairTaskMinDuration(repairTaskMinDuration.toString()); } + + /** + * clearForceRepair (invoked from the finally around a repair run) must consume the flag on both + * success and failure, without advancing repair_finish_ts. This is what prevents a failed force + * repair from wedging force_repair=true and bypassing min_repair_interval on every subsequent cycle. + */ + @Test + public void testClearForceRepairClearsFlagWithoutAdvancingFinishTs() + { + 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)); + + // A node with force_repair=true and a known repair_finish_ts. + 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)); + + assertTrue(AutoRepairUtils.isForceRepairSetForNode(repairType, myId)); + long finishBefore = AutoRepairUtils.getLastRepairTimeForNode(repairType, myId); + + // Simulate the end-of-run consumption of the flag (as the finally in AutoRepair.repair does). + AutoRepairUtils.clearForceRepair(repairType, myId); + + // The force_repair flag is cleared ... + assertFalse(AutoRepairUtils.isForceRepairSetForNode(repairType, myId)); + // ... but repair_finish_ts is left untouched. That column records when a repair last finished + // SUCCESSFULLY and is what min_repair_interval measures against. Advancing it here would make a + // failed forced repair look like a completed one and wrongly throttle the next real repair, so + // clearForceRepair writes only force_repair and the timestamp keeps its previous value. + assertEquals("clearForceRepair must not advance repair_finish_ts", + finishBefore, AutoRepairUtils.getLastRepairTimeForNode(repairType, myId)); + } } diff --git a/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairUtilsTest.java b/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairUtilsTest.java index b8ff8b8b140..d59faaa513f 100644 --- a/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairUtilsTest.java +++ b/test/unit/org/apache/cassandra/repair/autorepair/AutoRepairUtilsTest.java @@ -455,6 +455,37 @@ public void testUpdateFinishAutoRepairHistory() assertEquals(123, result.one().getLong(COL_REPAIR_FINISH_TS, 0)); } + /** + * A normal repair finish records repair_finish_ts but must NOT clear a pending force_repair flag. + * The flag is consumed only by clearForceRepair, and only for runs triggered by a force repair, so + * a force repair requested while a normal repair is running is still honored on the next cycle. + */ + @Test + public void testUpdateFinishAutoRepairHistoryPreservesForceRepair() + { + QueryProcessor.executeInternal(String.format( + "INSERT INTO %s.%s (repair_type, host_id, force_repair) VALUES ('%s', %s, true)", + SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY, + repairType.toString(), hostId)); + + assertTrue(AutoRepairUtils.isForceRepairSetForNode(repairType, hostId)); + + AutoRepairUtils.updateFinishAutoRepairHistory(repairType, hostId, 123); + + // repair_finish_ts is advanced ... + UntypedResultSet result = QueryProcessor.executeInternal(String.format( + "SELECT repair_finish_ts FROM %s.%s WHERE repair_type = '%s' AND host_id = %s", + SchemaConstants.DISTRIBUTED_KEYSPACE_NAME, SystemDistributedKeyspace.AUTO_REPAIR_HISTORY, + repairType.toString(), hostId)); + assertNotNull(result); + assertEquals(1, result.size()); + assertEquals(123, result.one().getLong(COL_REPAIR_FINISH_TS, 0)); + + // ... but the pending force_repair flag is left untouched. + assertTrue("normal repair finish must not clear a pending force_repair flag", + AutoRepairUtils.isForceRepairSetForNode(repairType, hostId)); + } + @Test public void testAddHostIdToDeleteHosts() {