This is an automated email from the ASF dual-hosted git repository.
iamaleksey pushed a commit to branch cep-45-mutation-tracking
in repository https://gitbox.apache.org/repos/asf/cassandra.git
The following commit(s) were added to refs/heads/cep-45-mutation-tracking by
this push:
new 950f930a8a CEP-45: Fixes and improvements to mutation tracking
migration
950f930a8a is described below
commit 950f930a8a952db5a16acc27aacdcc6268b21c46
Author: Aleksey Yeshchenko <[email protected]>
AuthorDate: Thu Aug 27 14:56:38 2026 +0100
CEP-45: Fixes and improvements to mutation tracking migration
patch by Aleksey Yeschenko; reviewed by Blake Eggleston for
CASSANDSA-21653
---
CHANGES.txt | 1 +
.../db/virtual/MutationTrackingTables.java | 95 +++++++++++++++++++++-
.../org/apache/cassandra/repair/RepairJob.java | 2 +-
.../migration/KeyspaceMigrationInfo.java | 14 +++-
.../MutationTrackingMigrationRepairResult.java | 16 +++-
.../migration/MutationTrackingMigrationState.java | 4 +-
.../migration/MutationTrackingRepairHandler.java | 27 ++++--
.../AdvanceMutationTrackingMigration.java | 24 ++++--
.../migration/KeyspaceMigrationInfoTest.java | 6 +-
.../AdvanceMutationTrackingMigrationTest.java | 15 +---
10 files changed, 167 insertions(+), 37 deletions(-)
diff --git a/CHANGES.txt b/CHANGES.txt
index 0738fb414d..c79abe7f88 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
cep-45-mutation-tracking
+ * CEP-45: Fixes and improvements to mutation tracking migration
* Add support for counters to mutation tracking (CASSANDRA-20959)
* CEP-45: Query forwarding (CASSANDRA-20309)
* Fix mutation tracking startup (CASSANDRA-20540)
diff --git
a/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java
b/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java
index 406693d23e..e0498028ee 100644
--- a/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java
+++ b/src/java/org/apache/cassandra/db/virtual/MutationTrackingTables.java
@@ -22,15 +22,21 @@ import java.util.Collection;
import java.util.Collections;
import java.util.List;
import java.util.Map;
+import java.util.stream.Collectors;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.db.DecoratedKey;
import org.apache.cassandra.db.Mutation;
import org.apache.cassandra.db.marshal.BooleanType;
import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.db.marshal.ListType;
import org.apache.cassandra.db.marshal.LongType;
import org.apache.cassandra.db.marshal.UTF8Type;
+import org.apache.cassandra.db.marshal.UUIDType;
import org.apache.cassandra.dht.LocalPartitioner;
+import org.apache.cassandra.dht.NormalizedRanges;
+import org.apache.cassandra.dht.Range;
+import org.apache.cassandra.dht.Token;
import org.apache.cassandra.journal.ActiveSegment;
import org.apache.cassandra.journal.Segment;
import org.apache.cassandra.replication.CoordinatorLog;
@@ -39,12 +45,16 @@ import org.apache.cassandra.replication.MutationJournal;
import org.apache.cassandra.replication.MutationTrackingService;
import org.apache.cassandra.replication.Shard;
import org.apache.cassandra.replication.ShortMutationId;
+import org.apache.cassandra.schema.TableId;
import org.apache.cassandra.schema.TableMetadata;
+import
org.apache.cassandra.service.replication.migration.KeyspaceMigrationInfo;
+import org.apache.cassandra.tcm.ClusterMetadata;
public class MutationTrackingTables
{
public static final String MUTATION_JOURNAL = "mutation_journal";
public static final String MUTATION_TRACKING_SHARDS =
"mutation_tracking_shards";
+ public static final String MUTATION_TRACKING_MIGRATION_STATE =
"mutation_tracking_migration_state";
private MutationTrackingTables() {}
@@ -53,7 +63,9 @@ public class MutationTrackingTables
if (!DatabaseDescriptor.getMutationTrackingEnabled())
return Collections.emptyList();
- return List.of(new MutationJournalTable(keyspace), new
MutationTrackingShardsTable(keyspace));
+ return List.of(new MutationJournalTable(keyspace),
+ new MutationTrackingShardsTable(keyspace),
+ new MutationTrackingMigrationStateTable(keyspace));
}
public static final class MutationJournalTable extends AbstractVirtualTable
@@ -183,4 +195,85 @@ public class MutationTrackingTables
return result;
}
}
+
+ /**
+ * Mutation tracking migration progress (held in {@link ClusterMetadata}).
+ */
+ public static class MutationTrackingMigrationStateTable extends
AbstractVirtualTable
+ {
+ private static final String KEYSPACE_NAME = "keyspace_name";
+ private static final String TABLE_NAME = "table_name";
+ private static final String TABLE_ID = "table_id";
+ private static final String STARTED_AT_EPOCH = "started_at_epoch";
+ private static final String PENDING_RANGES = "pending_ranges";
+ private static final String MIGRATED_RANGES = "migrated_ranges";
+
+ private static final ListType<String> STRING_LIST_TYPE =
ListType.getInstance(UTF8Type.instance, false);
+
+ MutationTrackingMigrationStateTable(String keyspace)
+ {
+ super(TableMetadata.builder(keyspace,
MUTATION_TRACKING_MIGRATION_STATE)
+ .comment("ranges still to be repaired for
in-progress mutation tracking migrations")
+ .kind(TableMetadata.Kind.VIRTUAL)
+ .partitioner(new
LocalPartitioner(UTF8Type.instance))
+ .addPartitionKeyColumn(KEYSPACE_NAME,
UTF8Type.instance)
+ .addClusteringColumn(TABLE_NAME,
UTF8Type.instance)
+ .addRegularColumn(TABLE_ID, UUIDType.instance)
+ .addRegularColumn(STARTED_AT_EPOCH,
LongType.instance)
+ .addRegularColumn(PENDING_RANGES,
STRING_LIST_TYPE)
+ .addRegularColumn(MIGRATED_RANGES,
STRING_LIST_TYPE)
+ .build());
+ }
+
+ @Override
+ public DataSet data()
+ {
+ SimpleDataSet result = new SimpleDataSet(metadata());
+ ClusterMetadata metadata = ClusterMetadata.current();
+
+ for (KeyspaceMigrationInfo info :
metadata.mutationTrackingMigrationState.keyspaceInfo.values())
+ addTableRows(metadata, info, result);
+
+ return result;
+ }
+
+ @Override
+ public DataSet data(DecoratedKey key)
+ {
+ String keyspaceName = UTF8Type.instance.compose(key.getKey());
+ SimpleDataSet result = new SimpleDataSet(metadata());
+ ClusterMetadata metadata = ClusterMetadata.current();
+
+ KeyspaceMigrationInfo info =
metadata.mutationTrackingMigrationState.getKeyspaceInfo(keyspaceName);
+ if (info != null)
+ addTableRows(metadata, info, result);
+
+ return result;
+ }
+
+ private static void addTableRows(ClusterMetadata metadata,
KeyspaceMigrationInfo info, SimpleDataSet result)
+ {
+ NormalizedRanges<Token> fullRing =
KeyspaceMigrationInfo.fullRing();
+ for (Map.Entry<TableId, NormalizedRanges<Token>> entry :
info.pendingRangesPerTable.entrySet())
+ {
+ TableId tid = entry.getKey();
+ NormalizedRanges<Token> pendingRanges = entry.getValue();
+
+ TableMetadata tm = metadata.schema.getTableMetadata(tid);
+ if (tm == null)
+ continue;
+
+ result.row(info.keyspace, tm.name)
+ .column(TABLE_ID, tid.asUUID())
+ .column(STARTED_AT_EPOCH, info.startedAtEpoch.getEpoch())
+ .column(PENDING_RANGES, rangesToStrings(pendingRanges))
+ .column(MIGRATED_RANGES,
rangesToStrings(fullRing.subtract(pendingRanges)));
+ }
+ }
+
+ private static List<String> rangesToStrings(NormalizedRanges<Token>
ranges)
+ {
+ return
ranges.stream().map(Range::toString).collect(Collectors.toList());
+ }
+ }
}
diff --git a/src/java/org/apache/cassandra/repair/RepairJob.java
b/src/java/org/apache/cassandra/repair/RepairJob.java
index db7079efc1..35f764c489 100644
--- a/src/java/org/apache/cassandra/repair/RepairJob.java
+++ b/src/java/org/apache/cassandra/repair/RepairJob.java
@@ -295,7 +295,7 @@ public class RepairJob extends AsyncFuture<RepairResult>
implements Runnable
cfs.metric.repairsCompleted.inc();
logger.info("Completing repair with excludedDeadNodes {}",
session.excludedDeadNodes);
ConsensusMigrationRepairResult cmrs =
ConsensusMigrationRepairResult.fromRepair(repairStartingEpoch,
getUnchecked(accordRepair), session.repairData, doPaxosRepair, doAccordRepair,
session.excludedDeadNodes, session.isIncremental);
- MutationTrackingMigrationRepairResult mtmrs =
MutationTrackingMigrationRepairResult.fromRepair(repairStartingEpoch,
session.excludedDeadNodes, session.previewKind.isPreview());
+ MutationTrackingMigrationRepairResult mtmrs =
MutationTrackingMigrationRepairResult.fromRepair(repairStartingEpoch,
session.repairData, session.allReplicas, session.pullRepair,
session.excludedDeadNodes, session.previewKind.isPreview());
trySuccess(new RepairResult(desc, stats, cmrs, mtmrs));
}
diff --git
a/src/java/org/apache/cassandra/service/replication/migration/KeyspaceMigrationInfo.java
b/src/java/org/apache/cassandra/service/replication/migration/KeyspaceMigrationInfo.java
index 65b08d0d78..a2021226cb 100644
---
a/src/java/org/apache/cassandra/service/replication/migration/KeyspaceMigrationInfo.java
+++
b/src/java/org/apache/cassandra/service/replication/migration/KeyspaceMigrationInfo.java
@@ -174,18 +174,24 @@ public class KeyspaceMigrationInfo
@Nonnull TableId
tableId,
@Nonnull
Collection<Range<Token>> repairedRanges)
{
- if (repairStartedEpoch.isBefore(startedAtEpoch))
- return this;
+ // TODO (expected): do something about this? nuke or serialize the
correct epoch alongised the transformation?
+ // this was dead code; repairStartedEpoch as passed was always
next transformation's epoch,
+ // and it was always > startedAtEpoch, guarding against nothing;
+ // there is an epoch eligibility check in
MutationTrackingRepairHandler in onSuccess(), but it is
+ // insufficient in face of potential race conditions (AY)
+ // if (repairStartedEpoch.isBefore(startedAtEpoch))
+ // return this;
NormalizedRanges<Token> currentPendingForTable =
pendingRangesPerTable.get(tableId);
if (currentPendingForTable == null)
- {
return this;
- }
NormalizedRanges<Token> normalizedRepaired =
NormalizedRanges.normalizedRanges(repairedRanges);
NormalizedRanges<Token> remainingForTable =
currentPendingForTable.subtract(normalizedRepaired);
+ if (remainingForTable.equals(currentPendingForTable))
+ return this;
+
ImmutableMap.Builder<TableId, NormalizedRanges<Token>> builder =
ImmutableMap.builder();
for (Map.Entry<TableId, NormalizedRanges<Token>> entry :
pendingRangesPerTable.entrySet())
{
diff --git
a/src/java/org/apache/cassandra/service/replication/migration/MutationTrackingMigrationRepairResult.java
b/src/java/org/apache/cassandra/service/replication/migration/MutationTrackingMigrationRepairResult.java
index e0ec934382..fb5009f4fb 100644
---
a/src/java/org/apache/cassandra/service/replication/migration/MutationTrackingMigrationRepairResult.java
+++
b/src/java/org/apache/cassandra/service/replication/migration/MutationTrackingMigrationRepairResult.java
@@ -33,6 +33,12 @@ public class MutationTrackingMigrationRepairResult
new MutationTrackingMigrationRepairResult(Epoch.EMPTY, false, "dead
nodes were excluded from the repair");
private static final MutationTrackingMigrationRepairResult PREVIEW =
new MutationTrackingMigrationRepairResult(Epoch.EMPTY, false, "the
repair was a preview");
+ private static final MutationTrackingMigrationRepairResult NO_DATA_REPAIR =
+ new MutationTrackingMigrationRepairResult(Epoch.EMPTY, false, "the
repair did not repair data (paxos-only or accord-only repair)");
+ private static final MutationTrackingMigrationRepairResult
NOT_ALL_REPLICAS =
+ new MutationTrackingMigrationRepairResult(Epoch.EMPTY, false, "the
repair did not include all replicas (-local, -dc, or -hosts repair)");
+ private static final MutationTrackingMigrationRepairResult PULL_REPAIR =
+ new MutationTrackingMigrationRepairResult(Epoch.EMPTY, false, "the
repair only streamed data one way (-pull repair)");
public final Epoch minEpoch;
public final boolean eligible;
@@ -48,10 +54,18 @@ public class MutationTrackingMigrationRepairResult
this.ineligibleReason = ineligibleReason;
}
- public static MutationTrackingMigrationRepairResult fromRepair(Epoch
minEpoch, boolean deadNodesExcluded, boolean isPreview)
+ public static MutationTrackingMigrationRepairResult fromRepair(Epoch
minEpoch,
+ boolean
dataRepaired,
+ boolean
allReplicas,
+ boolean
pullRepair,
+ boolean
deadNodesExcluded,
+ boolean
isPreview)
{
if (deadNodesExcluded) return DEAD_NODES_EXCLUDED;
if (isPreview) return PREVIEW;
+ if (!dataRepaired) return NO_DATA_REPAIR;
+ if (!allReplicas) return NOT_ALL_REPLICAS;
+ if (pullRepair) return PULL_REPAIR;
return new MutationTrackingMigrationRepairResult(minEpoch, true, null);
}
}
diff --git
a/src/java/org/apache/cassandra/service/replication/migration/MutationTrackingMigrationState.java
b/src/java/org/apache/cassandra/service/replication/migration/MutationTrackingMigrationState.java
index 1c500c6b10..68ac0ea2a1 100644
---
a/src/java/org/apache/cassandra/service/replication/migration/MutationTrackingMigrationState.java
+++
b/src/java/org/apache/cassandra/service/replication/migration/MutationTrackingMigrationState.java
@@ -177,8 +177,10 @@ public class MutationTrackingMigrationState implements
MetadataValue<MutationTra
if (info == null)
return this;
- // Subtract repaired ranges from table's pending set
+ // subtract repaired ranges from table's pending set; noop is
nothing's changed
KeyspaceMigrationInfo updated = info.withRangesRepairedForTable(epoch,
tableId, repairedRanges);
+ if (info == updated)
+ return this;
// if all tables fully repaired, remove keyspace (migration complete)
if (updated.isComplete())
diff --git
a/src/java/org/apache/cassandra/service/replication/migration/MutationTrackingRepairHandler.java
b/src/java/org/apache/cassandra/service/replication/migration/MutationTrackingRepairHandler.java
index b55c362a30..ee8e43879c 100644
---
a/src/java/org/apache/cassandra/service/replication/migration/MutationTrackingRepairHandler.java
+++
b/src/java/org/apache/cassandra/service/replication/migration/MutationTrackingRepairHandler.java
@@ -63,8 +63,8 @@ public class MutationTrackingRepairHandler
if (migrationInfo == null)
{
- logger.info("Repair session {} (parent session {})
completed for {}.{} but the keyspace is not migrating, not advancing mutation
tracking migration",
- desc.sessionId, desc.parentSessionId,
keyspace, tableName);
+ logger.debug("Repair session {} (parent session {})
completed for {}.{} but the keyspace is not migrating, not advancing mutation
tracking migration",
+ desc.sessionId, desc.parentSessionId,
keyspace, tableName);
return;
}
@@ -78,10 +78,13 @@ public class MutationTrackingRepairHandler
return;
}
- if
(migrationInfo.getPendingRangesForTable(tableMetadata.id).isEmpty())
+ NormalizedRanges<Token> pendingRanges =
migrationInfo.getPendingRangesForTable(tableMetadata.id);
+ NormalizedRanges<Token> repairedPendingRanges =
pendingRanges.intersection(NormalizedRanges.normalizedRanges(repairedRanges));
+
+ if (repairedPendingRanges.isEmpty())
{
- logger.info("Repair session {} (parent session {})
completed for {}.{} but the table has no ranges left to migrate, not advancing
mutation tracking migration",
- desc.sessionId, desc.parentSessionId,
keyspace, tableName);
+ logger.info("Repair session {} (parent session {})
completed for {}.{} but none of the repaired ranges {} are still pending
migration (pending: {}), not advancing mutation tracking migration",
+ desc.sessionId, desc.parentSessionId,
keyspace, tableName, repairedRanges, pendingRanges);
return;
}
@@ -104,7 +107,17 @@ public class MutationTrackingRepairHandler
}
ClusterMetadata committed =
ClusterMetadataService.instance().commit(
- new AdvanceMutationTrackingMigration(keyspace,
tableMetadata.id, repairedRanges));
+ new AdvanceMutationTrackingMigration(keyspace,
tableMetadata.id, repairedPendingRanges),
+ ignore -> ignore,
+ (code, message) ->
+ {
+ logger.info("Repair session {} (parent session {})
did not advance mutation tracking migration of {}.{}: {} ({})",
+ desc.sessionId, desc.parentSessionId,
keyspace, tableName, message, code);
+ return null;
+ });
+
+ if (committed == null)
+ return;
// Report from the metadata commit returned, not
current(), which races with other epochs
KeyspaceMigrationInfo advanced =
committed.mutationTrackingMigrationState.getKeyspaceInfo(keyspace);
@@ -119,7 +132,7 @@ public class MutationTrackingRepairHandler
"contributed {} range(s) {}; {} range(s)
remain to be repaired {}; {} range(s) already repaired {}; " +
"{} table(s) in the keyspace still migrating",
desc.sessionId, desc.parentSessionId,
keyspace, tableName, committed.epoch,
- repairedRanges.size(), repairedRanges,
+ repairedPendingRanges.size(),
repairedPendingRanges,
pending.size(), pending,
repaired.size(), repaired,
keyspaceComplete ? 0 :
advanced.pendingRangesPerTable.size());
diff --git
a/src/java/org/apache/cassandra/tcm/transformations/AdvanceMutationTrackingMigration.java
b/src/java/org/apache/cassandra/tcm/transformations/AdvanceMutationTrackingMigration.java
index 9245a11349..89e635669d 100644
---
a/src/java/org/apache/cassandra/tcm/transformations/AdvanceMutationTrackingMigration.java
+++
b/src/java/org/apache/cassandra/tcm/transformations/AdvanceMutationTrackingMigration.java
@@ -23,9 +23,6 @@ import java.util.Collection;
import javax.annotation.Nonnull;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
import org.apache.cassandra.db.TypeSizes;
import org.apache.cassandra.dht.Range;
import org.apache.cassandra.dht.Token;
@@ -57,7 +54,6 @@ import static
org.apache.cassandra.utils.CollectionSerializers.serializedCollect
*/
public class AdvanceMutationTrackingMigration implements Transformation
{
- private static final Logger logger =
LoggerFactory.getLogger(AdvanceMutationTrackingMigration.class);
public static final Serializer serializer = new Serializer();
@Nonnull
@@ -94,10 +90,7 @@ public class AdvanceMutationTrackingMigration implements
Transformation
KeyspaceMigrationInfo ksInfo =
prev.mutationTrackingMigrationState.getKeyspaceInfo(keyspace);
if (ksInfo == null)
- {
- logger.warn("Attempted to advance mutation tracking migration for
keyspace {} table {} which is not migrating", keyspace, tableId);
return new Rejected(INVALID, String.format("Keyspace %s is not
migrating", keyspace));
- }
Transformer transformer = prev.transformer();
@@ -105,14 +98,27 @@ public class AdvanceMutationTrackingMigration implements
Transformation
MutationTrackingMigrationState newState =
prev.mutationTrackingMigrationState
.withRangesRepairedForTable(keyspace, tableId, repairedRanges,
transformer.epoch());
- logger.info("Advanced mutation tracking migration for keyspace {},
table {}: {} ranges repaired",
- keyspace, tableId, repairedRanges.size());
+ if (newState == prev.mutationTrackingMigrationState)
+ {
+ return new Rejected(INVALID, String.format("Keyspace %s table %s
has no pending ranges intersecting %s",
+ keyspace, tableId,
repairedRanges));
+ }
return Transformation.success(
transformer.with(newState),
LockedRanges.AffectedRanges.EMPTY);
}
+ @Override
+ public String toString()
+ {
+ return "AdvanceMutationTrackingMigration{" +
+ "keyspace='" + keyspace + '\'' +
+ ", tableId=" + tableId +
+ ", repairedRanges=" + repairedRanges +
+ '}';
+ }
+
public static class Serializer implements
AsymmetricMetadataSerializer<Transformation, AdvanceMutationTrackingMigration>
{
@Override
diff --git
a/test/unit/org/apache/cassandra/service/replication/migration/KeyspaceMigrationInfoTest.java
b/test/unit/org/apache/cassandra/service/replication/migration/KeyspaceMigrationInfoTest.java
index 48a4170b75..3592209115 100644
---
a/test/unit/org/apache/cassandra/service/replication/migration/KeyspaceMigrationInfoTest.java
+++
b/test/unit/org/apache/cassandra/service/replication/migration/KeyspaceMigrationInfoTest.java
@@ -102,7 +102,7 @@ public class KeyspaceMigrationInfoTest
// info should be unchanged if a repair started before migration
started
KeyspaceMigrationInfo noop = info.withRangesRepairedForTable(epoch1,
testTableId, Collections.singleton(ranges.get(0)));
- assertSame(info, noop);
+ // assertSame(info, noop); // see the comment in
withRangesRepairedForTable()
// Subtract first range
KeyspaceMigrationInfo updated =
info.withRangesRepairedForTable(epoch2, testTableId,
Collections.singleton(ranges.get(0)));
@@ -117,6 +117,10 @@ public class KeyspaceMigrationInfoTest
assertFalse(updated2.pendingRangesPerTable.get(testTableId).intersects(ranges.get(0).right));
assertFalse(updated2.pendingRangesPerTable.get(testTableId).intersects(ranges.get(1).right));
assertFalse(updated2.isComplete());
+
+ // Subtracting non-intersecting ranges should return `this`
+ assertSame(updated2, updated2.withRangesRepairedForTable(epoch2,
testTableId, Collections.singleton(ranges.get(0))));
+ assertSame(updated2, updated2.withRangesRepairedForTable(epoch2,
testTableId, Collections.singleton(ranges.get(1))));
}
@Test
diff --git
a/test/unit/org/apache/cassandra/tcm/transformations/AdvanceMutationTrackingMigrationTest.java
b/test/unit/org/apache/cassandra/tcm/transformations/AdvanceMutationTrackingMigrationTest.java
index da33dc9958..948395f40a 100644
---
a/test/unit/org/apache/cassandra/tcm/transformations/AdvanceMutationTrackingMigrationTest.java
+++
b/test/unit/org/apache/cassandra/tcm/transformations/AdvanceMutationTrackingMigrationTest.java
@@ -193,18 +193,9 @@ public class AdvanceMutationTrackingMigrationTest
Transformation.Result result = transformation.execute(prev);
- // confirm noop
- assertTrue(result.isSuccess());
- ClusterMetadata updated = result.success().metadata;
-
- KeyspaceMigrationInfo expected = createExpectedInfo(
- "test_ks",
- testTableId,
- Collections.singleton(fullRing()),
- epoch1
- );
-
- assertEquals(expected,
updated.mutationTrackingMigrationState.getKeyspaceInfo("test_ks"));
+ // confirm rejection
+ assertTrue(result.isRejected());
+ assertTrue(result.rejected().reason.contains("no pending ranges
intersecting"));
}
@Test
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
