Skip to content

CASSANDRA-21663 Failed force repair should clear force repair flag #5125

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
tommystendahl wants to merge 1 commit into
base: cassandra-5.0
Choose a base branch
from
Open

CASSANDRA-21663 Failed force repair should clear force repair flag #5125

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGES.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
5.0.10
* Failed force repair should clear force repair flag (CASSANDRA-21663)
* Let the commit log allocator observe the shutdown state (CASSANDRA-21616)
* Unwrap LongType properly when calculating min/max terms in V1SSTableIndex (CASSANDRA-21635)
* Force repair should ignore min_repair_interval (CASSANDRA-21552)
Expand Down
130 changes: 77 additions & 53 deletions src/java/org/apache/cassandra/repair/autorepair/AutoRepair.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<Keyspace> keyspaces = new ArrayList<>();
Keyspace.all().forEach(keyspaces::add);
// Filter out keyspaces and tables to repair and group into a map by keyspace.
Map<String, List<String>> 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<Keyspace> keyspaces = new ArrayList<>();
Keyspace.all().forEach(keyspaces::add);
// Filter out keyspaces and tables to repair and group into a map by keyspace.
Map<String, List<String>> keyspacesAndTablesToRepair = new LinkedHashMap<>();
for (Keyspace keyspace : keyspaces)
{
continue;
if (!AutoRepairUtils.shouldConsiderKeyspace(keyspace))
{
continue;
}
List<String> tablesToBeRepairedList = retrieveTablesToBeRepaired(keyspace, config, repairType, repairState, collectedRepairStats);
keyspacesAndTablesToRepair.put(keyspace.getName(), tablesToBeRepairedList);
}
List<String> 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<PrioritizedRepairPlan> 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<PrioritizedRepairPlan> repairPlans = PrioritizedRepairPlan.build(keyspacesAndTablesToRepair, repairType, shuffleFunc, primaryRangeOnly);
repairState.updateRepairScheduleStatistics(repairPlans);

// calculate the repair assignments for each priority:keyspace.
Iterator<KeyspaceRepairAssignments> repairAssignmentsIterator = config.getTokenRangeSplitterInstance(repairType).getRepairAssignments(primaryRangeOnly, repairPlans);
// calculate the repair assignments for each priority:keyspace.
Iterator<KeyspaceRepairAssignments> repairAssignmentsIterator = config.getTokenRangeSplitterInstance(repairType).getRepairAssignments(primaryRangeOnly, repairPlans);

int keyspaceRepairAssignmentsAlreadyRepaired = 0;
while (repairAssignmentsIterator.hasNext())
{
KeyspaceRepairAssignments repairAssignments = repairAssignmentsIterator.next();
List<RepairAssignment> assignments = repairAssignments.getRepairAssignments();
if (assignments.isEmpty())
int keyspaceRepairAssignmentsAlreadyRepaired = 0;
while (repairAssignmentsIterator.hasNext())
{
KeyspaceRepairAssignments repairAssignments = repairAssignmentsIterator.next();
List<RepairAssignment> 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
{
Expand Down
36 changes: 34 additions & 2 deletions src/java/org/apache/cassandra/repair/autorepair/AutoRepairUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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 = ?"
Expand All @@ -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);
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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.
* <p>
* 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.
*
Expand Down
44 changes: 44 additions & 0 deletions test/unit/org/apache/cassandra/repair/autorepair/AutoRepairTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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));
}
}
Loading