From a79d34a2c064b78ce4c79f1af670a982915692fa Mon Sep 17 00:00:00 2001 From: Ariel Weisberg Date: Thu, 2 Apr 2026 14:43:21 -0400 Subject: [PATCH] CEP-45: BRR does tracked writes during migration to mutation tracking MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit During migration from untracked to tracked replication, blocking read repair (BRR) continues to run because mutation tracking lacks sufficient initialized state to provide monotonic reads for pending ranges. However, BRR was previously bypassing mutation tracking entirely — read repair mutations were sent as untracked writes via READ_REPAIR_REQ, which meant they were invisible to the mutation journal. This is unsafe because the migration repair that completes the transition assumes all writes after it started are tracked, so that tracked reads can safely begin afterward. Route BRR writes through TrackedWriteRequest.perform() when MigrationRouter.shouldUseTrackedForWrites() indicates the token is migrating. This gives each read repair mutation a proper MutationId and records it in the mutation journal, ensuring the migration invariant holds. Key changes: - Remove the isReadRepair flag from Mutation and all special-case bypasses that let read repair skip mutation tracking validation and routing - Add repairViaTrackedWrite() to BlockingReadRepair which sends each per-replica mutation as an independent tracked write with retry logic for migration races (RetryOnDifferentSystemException/CoordinatorBehindException) - Introduce WriteCallback interface (replacing Runnable) on write response handlers so tracked write completion can be observed without races - Make tracked→untracked instant in AlterSchema by skipping migration state entirely — tracked writes are already quorum writes so untracked reads are strictly less demanding. Reverting an in-progress untracked→tracked migration cleans up the now-unnecessary state. - Add RepairedBlockingViaTrackedWrite metric to observe BRR-via-tracked path - Add pauseRegularPriorityForTesting() to ActiveLogReconciler to allow tests to suppress background reconciliation while keeping high-priority (tracked read) reconciliation active - Add MutationTrackingReadRepairTest covering BRR behavior across all migration phases with both point reads and range scans --- .../db/AbstractMutationVerbHandler.java | 7 - .../db/CassandraKeyspaceWriteHandler.java | 3 +- .../db/CounterMutationVerbHandler.java | 2 +- .../org/apache/cassandra/db/Keyspace.java | 4 +- .../org/apache/cassandra/db/Mutation.java | 23 +- .../cassandra/db/ReadRepairVerbHandler.java | 1 - .../locator/AbstractReplicationStrategy.java | 5 +- .../cassandra/metrics/ReadRepairMetrics.java | 6 + .../replication/ActiveLogReconciler.java | 26 +- .../cassandra/replication/ForwardedWrite.java | 47 +- .../replication/MutationTrackingService.java | 25 +- .../replication/TrackedWriteRequest.java | 24 +- .../service/AbstractWriteResponseHandler.java | 7 +- .../DatacenterSyncWriteResponseHandler.java | 3 +- .../DatacenterWriteResponseHandler.java | 3 +- .../cassandra/service/StorageProxy.java | 7 +- .../service/WriteResponseHandler.java | 3 +- .../service/reads/AbstractReadExecutor.java | 5 +- .../reads/repair/AbstractReadRepair.java | 22 +- .../reads/repair/BlockingReadRepair.java | 188 ++++- .../MutationTrackingMigrationState.java | 7 + .../tcm/transformations/AlterSchema.java | 43 +- .../test/MutationTrackingMigrationTest.java | 55 +- ...tionTrackingIncrementalRepairTaskTest.java | 5 +- .../MutationTrackingReadRepairTest.java | 738 ++++++++++++++++++ .../repair/MutationTrackingRepairTest.java | 90 ++- .../MutationTrackingPendingReadTest.java | 1 - .../migration/MigrationRouterTest.java | 98 +-- .../AlterSchemaMutationTrackingTest.java | 43 +- 29 files changed, 1233 insertions(+), 258 deletions(-) create mode 100644 test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingReadRepairTest.java diff --git a/src/java/org/apache/cassandra/db/AbstractMutationVerbHandler.java b/src/java/org/apache/cassandra/db/AbstractMutationVerbHandler.java index 2fec6d2677..93bd4cbcc4 100644 --- a/src/java/org/apache/cassandra/db/AbstractMutationVerbHandler.java +++ b/src/java/org/apache/cassandra/db/AbstractMutationVerbHandler.java @@ -202,13 +202,6 @@ public abstract class AbstractMutationVerbHandler implement */ protected ClusterMetadata checkReplicationMigration(ClusterMetadata metadata, Message message, InetAddressAndPort respondTo) { - // Read repair mutations always bypass mutation tracking and use the untracked - // write path, so skip the replication migration routing check. The isReadRepair - // flag on the mutation hasn't been set yet at this point — it's set later in - // applyMutation() — so we check the handler type instead. - if (this instanceof ReadRepairVerbHandler) - return metadata; - IMutation mutation = message.payload; MutationRouting expected = mutation.id().isNone() ? MutationRouting.UNTRACKED : MutationRouting.TRACKED; if (expected == MigrationRouter.getMutationRouting(metadata, mutation)) diff --git a/src/java/org/apache/cassandra/db/CassandraKeyspaceWriteHandler.java b/src/java/org/apache/cassandra/db/CassandraKeyspaceWriteHandler.java index 0648233ba3..9b68acc708 100644 --- a/src/java/org/apache/cassandra/db/CassandraKeyspaceWriteHandler.java +++ b/src/java/org/apache/cassandra/db/CassandraKeyspaceWriteHandler.java @@ -47,8 +47,7 @@ public class CassandraKeyspaceWriteHandler implements KeyspaceWriteHandler { group = Keyspace.writeOrder.start(); - if (!mutation.isReadRepair()) - MigrationRouter.validateUntrackedMutation(mutation); + MigrationRouter.validateUntrackedMutation(mutation); // write the mutation to the commitlog and memtables CommitLogPosition position = null; if (makeDurable) diff --git a/src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java b/src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java index c24262fa9c..9623320e01 100644 --- a/src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java +++ b/src/java/org/apache/cassandra/db/CounterMutationVerbHandler.java @@ -70,7 +70,7 @@ public class CounterMutationVerbHandler extends AbstractMutationVerbHandler MessagingService.instance().send(message.emptyResponse(), respondToAddress), + handler -> MessagingService.instance().send(message.emptyResponse(), respondToAddress), Dispatcher.RequestTime.forImmediateExecution()); } } diff --git a/src/java/org/apache/cassandra/db/Keyspace.java b/src/java/org/apache/cassandra/db/Keyspace.java index eddf192f4b..bb5fc12316 100644 --- a/src/java/org/apache/cassandra/db/Keyspace.java +++ b/src/java/org/apache/cassandra/db/Keyspace.java @@ -439,7 +439,7 @@ public class Keyspace boolean updateIndexes, boolean isDroppable) { - if (MigrationRouter.isFullyTracked(mutation) && !mutation.isReadRepair()) + if (MigrationRouter.isFullyTracked(mutation)) applyInternalTracked(mutation, null); else applyInternal(mutation, makeDurable, updateIndexes, isDroppable, false, null); @@ -462,7 +462,7 @@ public class Keyspace boolean isDeferrable, Promise future) { - Preconditions.checkState((!getMetadata().useMutationTracking() || mutation.isReadRepair()) && mutation.id().isNone()); + Preconditions.checkState(!MigrationRouter.isFullyTracked(mutation) && mutation.id().isNone()); if (TEST_FAIL_WRITES && getMetadata().name.equals(TEST_FAIL_WRITES_KS)) throw new RuntimeException("Testing write failures"); diff --git a/src/java/org/apache/cassandra/db/Mutation.java b/src/java/org/apache/cassandra/db/Mutation.java index b165905dc4..9c4bb8a1e3 100644 --- a/src/java/org/apache/cassandra/db/Mutation.java +++ b/src/java/org/apache/cassandra/db/Mutation.java @@ -110,21 +110,6 @@ public class Mutation implements IMutation, Supplier, Commitable // because it is being applied by one or in a context where transaction conflicts don't occur private PotentialTxnConflicts potentialTxnConflicts; - // Transient: not serialized on the wire. Set by ReadRepairVerbHandler on the - // receiving side so downstream code (Keyspace.apply, write handlers) can route - // read repair mutations through the untracked write path during migration. - private transient boolean isReadRepair; - - public void setReadRepair(boolean readRepair) - { - this.isReadRepair = readRepair; - } - - public boolean isReadRepair() - { - return isReadRepair; - } - public Mutation(MutationId id, PartitionUpdate update) { this(id, update.metadata().keyspace, update.partitionKey(), ImmutableMap.of(update.metadata().id, update), approxTime.now(), update.metadata().params.cdc, PotentialTxnConflicts.DISALLOW); @@ -165,9 +150,7 @@ public class Mutation implements IMutation, Supplier, Commitable @Override public Mutation withMutationId(MutationId mutationId) { - Mutation m = new Mutation(mutationId, keyspaceName, key, modifications, approxCreatedAtNanos, cdcEnabled, potentialTxnConflicts); - m.isReadRepair = this.isReadRepair; - return m; + return new Mutation(mutationId, keyspaceName, key, modifications, approxCreatedAtNanos, cdcEnabled, potentialTxnConflicts); } private static boolean cdcEnabled(Iterable modifications) @@ -201,9 +184,7 @@ public class Mutation implements IMutation, Supplier, Commitable Map updates = builder.build(); checkState(!updates.isEmpty(), "Updates should not be empty"); - Mutation result = new Mutation(id, keyspaceName, key, builder.build(), approxCreatedAtNanos, potentialTxnConflicts); - result.isReadRepair = this.isReadRepair; - return result; + return new Mutation(id, keyspaceName, key, builder.build(), approxCreatedAtNanos, potentialTxnConflicts); } public @Nullable Mutation without(TableId tableId) diff --git a/src/java/org/apache/cassandra/db/ReadRepairVerbHandler.java b/src/java/org/apache/cassandra/db/ReadRepairVerbHandler.java index fc2287a967..d40359c144 100644 --- a/src/java/org/apache/cassandra/db/ReadRepairVerbHandler.java +++ b/src/java/org/apache/cassandra/db/ReadRepairVerbHandler.java @@ -27,7 +27,6 @@ public class ReadRepairVerbHandler extends AbstractMutationVerbHandler public void applyMutation(Mutation mutation) { - mutation.setReadRepair(true); mutation.apply(); } diff --git a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java index c839b751ec..a55202aad2 100644 --- a/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java +++ b/src/java/org/apache/cassandra/locator/AbstractReplicationStrategy.java @@ -25,6 +25,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; +import java.util.function.Consumer; import java.util.function.Supplier; import com.google.common.base.Preconditions; @@ -93,7 +94,7 @@ public abstract class AbstractReplicationStrategy public abstract DataPlacement calculateDataPlacement(Epoch epoch, List> ranges, ClusterMetadata metadata); public AbstractWriteResponseHandler getWriteResponseHandler(ReplicaPlan.ForWrite replicaPlan, - Runnable callback, + Consumer> callback, WriteType writeType, Supplier hintOnFailure, Dispatcher.RequestTime requestTime) @@ -103,7 +104,7 @@ public abstract class AbstractReplicationStrategy } public AbstractWriteResponseHandler getWriteResponseHandler(ReplicaPlan.ForWrite replicaPlan, - Runnable callback, + Consumer> callback, WriteType writeType, Supplier hintOnFailure, Dispatcher.RequestTime requestTime, diff --git a/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java b/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java index 28d2a91f82..d035b7bf26 100644 --- a/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java +++ b/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java @@ -37,6 +37,12 @@ public class ReadRepairMetrics */ public static final Meter repairedBlockingViaAccord = Metrics.meter(factory.createMetricName("RepairedBlockingViaAccord")); + /** + * Blocking read repair was sent as a tracked write. This happens when the token is migrating to tracked replication + * but migration hasn't completed + */ + public static final Meter repairedBlockingViaTrackedWrite = Metrics.meter(factory.createMetricName("RepairedBlockingViaTrackedWrite")); + /** * This should be zero if you are trying to run Accord in a 100% correct way and interoperating with non-transactional writes. * diff --git a/src/java/org/apache/cassandra/replication/ActiveLogReconciler.java b/src/java/org/apache/cassandra/replication/ActiveLogReconciler.java index 24083a0f8d..ed605519b1 100644 --- a/src/java/org/apache/cassandra/replication/ActiveLogReconciler.java +++ b/src/java/org/apache/cassandra/replication/ActiveLogReconciler.java @@ -120,8 +120,11 @@ public final class ActiveLogReconciler implements Shutdownable Task task; while ((task = highPriorityTasks.poll()) != null) task.send(); - while ((task = regularPriorityTasks.poll()) != null) - task.send(); + if (!isRegularPriorityPaused) + { + while ((task = regularPriorityTasks.poll()) != null) + task.send(); + } haveWork.acquire(1); } @@ -261,6 +264,7 @@ public final class ActiveLogReconciler implements Shutdownable private volatile boolean isShutdown = false; private volatile boolean isPaused = false; + private volatile boolean isRegularPriorityPaused = false; @Override public boolean isTerminated() @@ -307,4 +311,22 @@ public final class ActiveLogReconciler implements Shutdownable { isPaused = false; } + + /** + * Pause only regular-priority (background write retry) task delivery. + * High-priority tasks (needed by tracked read reconciliation) continue to be processed. + * This allows tests to prevent the reconciler from proactively fixing inconsistencies + * while still allowing tracked reads to pull missing mutations. + */ + @VisibleForTesting + void pauseRegularPriorityForTesting() + { + isRegularPriorityPaused = true; + } + + @VisibleForTesting + void resumeRegularPriorityForTesting() + { + isRegularPriorityPaused = false; + } } diff --git a/src/java/org/apache/cassandra/replication/ForwardedWrite.java b/src/java/org/apache/cassandra/replication/ForwardedWrite.java index d69129a906..498f8857fd 100644 --- a/src/java/org/apache/cassandra/replication/ForwardedWrite.java +++ b/src/java/org/apache/cassandra/replication/ForwardedWrite.java @@ -24,6 +24,7 @@ import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Set; +import java.util.function.Consumer; import com.google.common.base.Preconditions; import com.google.common.collect.Sets; @@ -297,6 +298,13 @@ public class ForwardedWrite } public static AbstractWriteResponseHandler forwardMutation(Mutation mutation, ReplicaPlan.ForWrite plan, AbstractReplicationStrategy strategy, Dispatcher.RequestTime requestTime) + { + return forwardMutationInternal(mutation, plan, strategy, requestTime, null); + } + + private static AbstractWriteResponseHandler forwardMutationInternal( + Mutation mutation, ReplicaPlan.ForWrite plan, AbstractReplicationStrategy strategy, + Dispatcher.RequestTime requestTime, Consumer> onComplete) { // find leader NodeProximity proximity = DatabaseDescriptor.getNodeProximity(); @@ -318,7 +326,7 @@ public class ForwardedWrite // create callback and forward to leader logger.trace("Selected {} as leader for mutation with key {}", leader.endpoint(), mutation.key()); - AbstractWriteResponseHandler handler = strategy.getWriteResponseHandler(plan, null, WriteType.SIMPLE, null, requestTime); + AbstractWriteResponseHandler handler = strategy.getWriteResponseHandler(plan, onComplete, WriteType.SIMPLE, null, requestTime); // Add callbacks for replicas to respond directly to coordinator Message toLeader = Message.outWithRequestTime(Verb.FORWARD_WRITE_REQ, new MutationRequest(mutation, plan), requestTime); @@ -348,6 +356,14 @@ public class ForwardedWrite ReplicaPlan.ForWrite plan, AbstractReplicationStrategy strategy, Dispatcher.RequestTime requestTime) + { + return forwardCounterMutationInternal(counterMutation, plan, strategy, requestTime, null); + } + + private static AbstractWriteResponseHandler forwardCounterMutationInternal( + CounterMutation counterMutation, ReplicaPlan.ForWrite plan, + AbstractReplicationStrategy strategy, Dispatcher.RequestTime requestTime, + Consumer> onComplete) { Preconditions.checkArgument(counterMutation.id().isNone(), "CounterMutation should not have an ID when forwarding"); @@ -373,7 +389,7 @@ public class ForwardedWrite logger.trace("Forwarding tracked counter mutation to leader replica {}", leader); // Create response handler for all replicas - AbstractWriteResponseHandler handler = strategy.getWriteResponseHandler(plan, null, WriteType.COUNTER, null, requestTime); + AbstractWriteResponseHandler handler = strategy.getWriteResponseHandler(plan, onComplete, WriteType.COUNTER, null, requestTime); // Add callbacks for all live replicas to respond directly to coordinator Message forwardMessage = Message.outWithRequestTime(Verb.COUNTER_MUTATION_REQ, counterMutation, requestTime); @@ -417,6 +433,33 @@ public class ForwardedWrite return forwardMutation((Mutation) mutation, plan, strategy, requestTime); } + /** + * Forward a mutation to a replica leader for processing. + * Dispatches to the appropriate method based on mutation type. + * + *

Like {@link #forward(IMutation, ReplicaPlan.ForWrite, AbstractReplicationStrategy, Dispatcher.RequestTime)}, + * but wires a completion callback on the handler before any messages are sent, avoiding races where the handler + * is signaled before the caller can observe it. + * + * @param mutation the mutation to forward (can be Mutation or CounterMutation) + * @param plan the replica plan + * @param strategy the replication strategy + * @param requestTime the request time + * @param onComplete callback invoked when the write response handler completes + * @return the write response handler + */ + static AbstractWriteResponseHandler forward(IMutation mutation, + ReplicaPlan.ForWrite plan, + AbstractReplicationStrategy strategy, + Dispatcher.RequestTime requestTime, + Consumer> onComplete) + { + if (mutation instanceof CounterMutation) + return forwardCounterMutationInternal((CounterMutation) mutation, plan, strategy, requestTime, onComplete); + else + return forwardMutationInternal((Mutation) mutation, plan, strategy, requestTime, onComplete); + } + /** * Apply a forwarded tracked counter mutation on the leader replica. * Called by CounterMutationVerbHandler when receiving a forwarded counter write. diff --git a/src/java/org/apache/cassandra/replication/MutationTrackingService.java b/src/java/org/apache/cassandra/replication/MutationTrackingService.java index 2511a6ebe3..5d4aec80ca 100644 --- a/src/java/org/apache/cassandra/replication/MutationTrackingService.java +++ b/src/java/org/apache/cassandra/replication/MutationTrackingService.java @@ -1073,6 +1073,7 @@ public class MutationTrackingService return NONE; } + // TODO (CASSANDRA-20955): hasExisting should not be possible here once shard cleanup is implemented if (prevKsm == null) return nextKsm.useMutationTracking() ? (hasExisting ? UpdateDecision.REPLICA_GROUP : UpdateDecision.CREATE) : UpdateDecision.NONE; @@ -1093,8 +1094,12 @@ public class MutationTrackingService if (!prevKsm.useMutationTracking() && nextKsm.useMutationTracking()) { - Preconditions.checkState(!hasExisting, "Existing shard found for keyspace, but prev ksn has mutation tracking disabled"); - return UpdateDecision.MIGRATE_TO; + // TODO (CASSANDRA-20955): hasExisting should not be possible here once shard cleanup is implemented. + // Shards from a prior tracked phase can survive a round-trip migration + // (tracked→untracked→tracked) because shard cleanup is not yet implemented. + // Update the replica group instead of failing. +// Preconditions.checkState(!hasExisting, "Existing shard found for keyspace, but prev ksn has mutation tracking disabled"); + return hasExisting ? UpdateDecision.REPLICA_GROUP : UpdateDecision.MIGRATE_TO; } if (!calculateParticipantsForRange(nextKsm, next).equals(calculateParticipantsForRange(prevKsm, prev))) @@ -1491,6 +1496,22 @@ public class MutationTrackingService activeReconciler.resumeForTesting(); } + /** + * Pause only regular-priority (background write retry) delivery in the active reconciler. + * High-priority tasks (needed by tracked read reconciliation) continue to be processed. + */ + @VisibleForTesting + public void pauseActiveReconcilerRegularPriority() + { + activeReconciler.pauseRegularPriorityForTesting(); + } + + @VisibleForTesting + public void resumeActiveReconcilerRegularPriority() + { + activeReconciler.resumeRegularPriorityForTesting(); + } + @VisibleForTesting public static class TestAccess { diff --git a/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java b/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java index ed6975525f..9a5ab64157 100644 --- a/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java +++ b/src/java/org/apache/cassandra/replication/TrackedWriteRequest.java @@ -23,6 +23,7 @@ import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.concurrent.ThreadLocalRandom; +import java.util.function.Consumer; import com.google.common.base.Preconditions; @@ -166,6 +167,23 @@ public class TrackedWriteRequest */ public static AbstractWriteResponseHandler perform( IMutation mutation, ConsistencyLevel consistencyLevel, Dispatcher.RequestTime requestTime) + { + return perform(mutation, consistencyLevel, requestTime, null); + } + + /** + * Like {@link #perform}, but wires a completion callback on the handler before any messages are sent, + * avoiding the race where the handler is signaled before the caller can wire callbacks. + * + * @param mutation the mutation to be applied + * @param consistencyLevel the consistency level for the write operation + * @param requestTime object holding times when request got enqueued and started execution + * @param onComplete callback invoked when the handler is signaled; receives the handler + */ + public static AbstractWriteResponseHandler perform( + IMutation mutation, ConsistencyLevel consistencyLevel, + Dispatcher.RequestTime requestTime, + Consumer> onComplete) { Tracing.trace("Determining replicas for mutation"); @@ -181,7 +199,7 @@ public class TrackedWriteRequest { logger.trace("Remote tracked request {} {}", mutation, plan); writeMetrics.remoteRequests.mark(); - return ForwardedWrite.forward(mutation, plan, rs, requestTime); + return ForwardedWrite.forward(mutation, plan, rs, requestTime, onComplete); } logger.trace("Local tracked request {} {}", mutation, plan); @@ -199,12 +217,12 @@ public class TrackedWriteRequest final TrackedWriteResponseHandler handler; if (mutation instanceof CounterMutation) { - handler = TrackedWriteResponseHandler.wrap(rs.getWriteResponseHandler(plan, null, WriteType.COUNTER, null, requestTime), id); + handler = TrackedWriteResponseHandler.wrap(rs.getWriteResponseHandler(plan, onComplete, WriteType.COUNTER, null, requestTime), id); applyCounterMutationLocally((CounterMutation) mutation, plan, handler); } else { - handler = TrackedWriteResponseHandler.wrap(rs.getWriteResponseHandler(plan, null, WriteType.SIMPLE, null, requestTime), id); + handler = TrackedWriteResponseHandler.wrap(rs.getWriteResponseHandler(plan, onComplete, WriteType.SIMPLE, null, requestTime), id); applyLocallyAndSendToReplicas((Mutation) mutation, plan, handler); } return handler; diff --git a/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java b/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java index 434f25055b..61a9e867c7 100644 --- a/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/AbstractWriteResponseHandler.java @@ -24,6 +24,7 @@ import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import java.util.concurrent.atomic.AtomicReferenceFieldUpdater; +import java.util.function.Consumer; import java.util.function.Function; import java.util.function.Supplier; @@ -83,7 +84,7 @@ public abstract class AbstractWriteResponseHandler implements RequestCallback private final Condition condition = newOneTimeCondition(); protected final ReplicaPlan.ForWrite replicaPlan; - protected final Runnable callback; + protected final Consumer> callback; protected final WriteType writeType; private static final AtomicIntegerFieldUpdater failuresUpdater = AtomicIntegerFieldUpdater.newUpdater(AbstractWriteResponseHandler.class, "failures"); @@ -117,7 +118,7 @@ public abstract class AbstractWriteResponseHandler implements RequestCallback * @param hintOnFailure * @param requestTime */ - protected AbstractWriteResponseHandler(ForWrite replicaPlan, Runnable callback, WriteType writeType, + protected AbstractWriteResponseHandler(ForWrite replicaPlan, Consumer> callback, WriteType writeType, Supplier hintOnFailure, Dispatcher.RequestTime requestTime) { this.replicaPlan = replicaPlan; @@ -358,7 +359,7 @@ public abstract class AbstractWriteResponseHandler implements RequestCallback condition.signalAll(); if (callback != null) - callback.run(); + callback.accept(this); } @Override diff --git a/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java b/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java index b897298dbb..97fd1750a1 100644 --- a/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/DatacenterSyncWriteResponseHandler.java @@ -20,6 +20,7 @@ package org.apache.cassandra.service; import java.util.HashMap; import java.util.Map; import java.util.concurrent.atomic.AtomicInteger; +import java.util.function.Consumer; import java.util.function.Supplier; import org.apache.cassandra.config.DatabaseDescriptor; @@ -47,7 +48,7 @@ public class DatacenterSyncWriteResponseHandler extends AbstractWriteResponse private final AtomicInteger acks = new AtomicInteger(0); public DatacenterSyncWriteResponseHandler(ReplicaPlan.ForWrite replicaPlan, - Runnable callback, + Consumer> callback, WriteType writeType, Supplier hintOnFailure, Dispatcher.RequestTime requestTime) diff --git a/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java b/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java index dc9f38a99e..cd64cc0146 100644 --- a/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/DatacenterWriteResponseHandler.java @@ -17,6 +17,7 @@ */ package org.apache.cassandra.service; +import java.util.function.Consumer; import java.util.function.Predicate; import java.util.function.Supplier; @@ -36,7 +37,7 @@ public class DatacenterWriteResponseHandler extends WriteResponseHandler private final Predicate waitingFor = InOurDc.endpoints(); public DatacenterWriteResponseHandler(ReplicaPlan.ForWrite replicaPlan, - Runnable callback, + Consumer> callback, WriteType writeType, Supplier hintOnFailure, Dispatcher.RequestTime requestTime) diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java index eaf511fd9a..4a29d87165 100644 --- a/src/java/org/apache/cassandra/service/StorageProxy.java +++ b/src/java/org/apache/cassandra/service/StorageProxy.java @@ -38,6 +38,7 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; import java.util.function.Function; import java.util.function.IntPredicate; import java.util.stream.Collectors; @@ -2194,7 +2195,7 @@ public class StorageProxy implements StorageProxyMBean ConsistencyLevel consistencyLevel, String localDataCenter, WritePerformer performer, - Runnable callback, + Consumer> callback, WriteType writeType, Dispatcher.RequestTime requestTime) { @@ -2243,7 +2244,7 @@ public class StorageProxy implements StorageProxyMBean Dispatcher.RequestTime requestTime) { AbstractReplicationStrategy replicationStrategy = replicaPlan.replicationStrategy(); - AbstractWriteResponseHandler writeHandler = replicationStrategy.getWriteResponseHandler(replicaPlan, () -> { + AbstractWriteResponseHandler writeHandler = replicationStrategy.getWriteResponseHandler(replicaPlan, handler -> { long delay = Math.max(0, currentTimeMillis() - baseComplete.get()); viewWriteMetrics.viewWriteLatency.update(delay, MILLISECONDS); }, writeType, mutation, requestTime); @@ -2582,7 +2583,7 @@ public class StorageProxy implements StorageProxyMBean // Must be called on a replica of the mutation. This replica becomes the // leader of this mutation. - public static AbstractWriteResponseHandler applyCounterMutationOnLeader(CounterMutation cm, String localDataCenter, Runnable callback, Dispatcher.RequestTime requestTime) + public static AbstractWriteResponseHandler applyCounterMutationOnLeader(CounterMutation cm, String localDataCenter, Consumer> callback, Dispatcher.RequestTime requestTime) throws UnavailableException, OverloadedException { return performWrite(cm, cm.consistency(), localDataCenter, counterWritePerformer, callback, WriteType.COUNTER, requestTime); diff --git a/src/java/org/apache/cassandra/service/WriteResponseHandler.java b/src/java/org/apache/cassandra/service/WriteResponseHandler.java index d99aa3989f..41eb28ff21 100644 --- a/src/java/org/apache/cassandra/service/WriteResponseHandler.java +++ b/src/java/org/apache/cassandra/service/WriteResponseHandler.java @@ -19,6 +19,7 @@ package org.apache.cassandra.service; import java.util.Map; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; +import java.util.function.Consumer; import java.util.function.Supplier; import org.slf4j.Logger; @@ -47,7 +48,7 @@ public class WriteResponseHandler extends AbstractWriteResponseHandler = AtomicIntegerFieldUpdater.newUpdater(WriteResponseHandler.class, "responses"); public WriteResponseHandler(ReplicaPlan.ForWrite replicaPlan, - Runnable callback, + Consumer> callback, WriteType writeType, Supplier hintOnFailure, Dispatcher.RequestTime requestTime) diff --git a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java index df694b0477..3c7294ca6d 100644 --- a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java +++ b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java @@ -44,6 +44,7 @@ import org.apache.cassandra.net.Message; import org.apache.cassandra.net.MessagingService; import org.apache.cassandra.service.StorageProxy.LocalReadRunnable; import org.apache.cassandra.service.reads.repair.ReadRepair; +import org.apache.cassandra.service.replication.migration.MigrationRouter; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tracing.TraceState; import org.apache.cassandra.tracing.Tracing; @@ -190,7 +191,9 @@ public abstract class AbstractReadExecutor implements ReadExecutor ReadCoordinator coordinator, Dispatcher.RequestTime requestTime) throws UnavailableException { - Preconditions.checkArgument(!command.metadata().replicationType().isTracked()); + // During migration, a tracked table may have pending ranges that use untracked reads. + // MigrationRouter.shouldUseTracked() returns false for those tokens, routing them here. + Preconditions.checkArgument(!MigrationRouter.shouldUseTracked(command)); Keyspace keyspace = Keyspace.open(command.metadata().keyspace); ColumnFamilyStore cfs = keyspace.getColumnFamilyStore(command.metadata().id); SpeculativeRetryPolicy retry = cfs.metadata().params.speculativeRetry; diff --git a/src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java b/src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java index c8f11df0cb..a84259ee9e 100644 --- a/src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java +++ b/src/java/org/apache/cassandra/service/reads/repair/AbstractReadRepair.java @@ -61,7 +61,8 @@ public abstract class AbstractReadRepair, P extends Repli protected final ReplicaPlan.Shared replicaPlan; protected final ColumnFamilyStore cfs; - private volatile DigestRepair digestRepair = null; + private DigestRepair digestRepair = null; + private boolean repairMeterMarked = false; private static class DigestRepair, P extends ReplicaPlan.ForRead> { @@ -119,10 +120,27 @@ public abstract class AbstractReadRepair, P extends Repli abstract Meter getRepairMeter(); + /** + * Mark the repair meter at most once per read command. Single-partition reads mark + * via {@code startRepair()} on digest mismatch. Range reads never call {@code startRepair()} + * — they go through {@code DataResolver.resolveWithReadRepair()} which calls + * {@code repairPartition()} directly for each inconsistent partition. Without this + * guard, range reads would either miss the metric entirely (if only marked in + * {@code startRepair()}) or over-count (if marked per partition in {@code repairPartition()}). + */ + protected void markRepairMeter() + { + if (!repairMeterMarked) + { + repairMeterMarked = true; + getRepairMeter().mark(); + } + } + // digestResolver isn't used here because we resend read requests to all participants public void startRepair(DigestResolver digestResolver, Consumer resultConsumer) { - getRepairMeter().mark(); + markRepairMeter(); /* * When repaired data tracking is enabled, a digest will be created from data reads from repaired SSTables. diff --git a/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java b/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java index 3535f6ed50..c34cefc5c1 100644 --- a/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java +++ b/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java @@ -18,6 +18,8 @@ package org.apache.cassandra.service.reads.repair; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import java.util.Queue; import java.util.concurrent.ConcurrentLinkedQueue; @@ -42,12 +44,20 @@ import org.apache.cassandra.db.Mutation; import org.apache.cassandra.db.ReadCommand; import org.apache.cassandra.db.ReadCommand.PotentialTxnConflicts; import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators; +import org.apache.cassandra.exceptions.CoordinatorBehindException; import org.apache.cassandra.exceptions.ReadTimeoutException; +import org.apache.cassandra.exceptions.RequestFailure; import org.apache.cassandra.locator.Endpoints; +import org.apache.cassandra.locator.InetAddressAndPort; import org.apache.cassandra.locator.Replica; import org.apache.cassandra.locator.ReplicaPlan; import org.apache.cassandra.locator.ReplicaPlan.ForWrite; import org.apache.cassandra.metrics.ReadRepairMetrics; +import org.apache.cassandra.net.Message; +import org.apache.cassandra.net.MessagingService; +import org.apache.cassandra.net.NoPayload; +import org.apache.cassandra.net.RequestCallback; +import org.apache.cassandra.replication.TrackedWriteRequest; import org.apache.cassandra.service.accord.AccordService; import org.apache.cassandra.service.accord.api.PartitionKey; import org.apache.cassandra.service.accord.serializers.TableMetadatas; @@ -60,15 +70,19 @@ import org.apache.cassandra.service.consensus.TransactionalMode; import org.apache.cassandra.service.consensus.migration.ConsensusMigrationMutationHelper; import org.apache.cassandra.service.consensus.migration.TransactionalMigrationFromMode; import org.apache.cassandra.service.reads.ReadCoordinator; +import org.apache.cassandra.service.replication.migration.MigrationRouter; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tracing.Tracing; import org.apache.cassandra.transport.Dispatcher; +import org.apache.cassandra.utils.Throwables; +import org.apache.cassandra.utils.concurrent.AsyncPromise; import org.apache.cassandra.utils.concurrent.Future; import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException; import static com.google.common.base.Preconditions.checkState; import static java.util.concurrent.TimeUnit.MICROSECONDS; import static java.util.concurrent.TimeUnit.NANOSECONDS; +import static org.apache.cassandra.net.Verb.READ_REPAIR_REQ; import static org.apache.cassandra.utils.Clock.Global.nanoTime; /** @@ -199,6 +213,7 @@ public class BlockingReadRepair, P extends ReplicaPlan.Fo @Override public void repairPartition(DecoratedKey dk, Map mutations, ReplicaPlan.ForWrite writePlan, ReadRepairSource rrSource) { + markRepairMeter(); // non-Accord reads only ever touch one table and key so all mutations need to be applied either transactionally // or non-transactionally (not a mix). There is no retry loop here because read repair is relatively rare so it racing // with changes to migrating ranges should also be pretty rare so it isn't worth the added complexity. If you were @@ -209,12 +224,183 @@ public class BlockingReadRepair, P extends ReplicaPlan.Fo // then we take the non-transactional path and the mutations are intercepted in ReadCoordinator.sendRepairMutation // which will ensure the repair mutation runs in the command store thread after preceding transactions are done ClusterMetadata cm = ClusterMetadata.current(); - if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord)) + if (MigrationRouter.shouldUseTrackedForWrites(cm, command.metadata().keyspace, command.metadata().id, dk.getToken())) + repairViaTrackedWrite(dk, mutations, writePlan); + else if (coordinator.isEventuallyConsistent() && ConsensusMigrationMutationHelper.tokenShouldBeWrittenThroughAccord(cm, command.metadata().id, dk.getToken(), TransactionalMode::readRepairsThroughAccord, TransactionalMigrationFromMode::readRepairsThroughAccord)) repairViaAccordTransaction(dk, mutations, writePlan); else repairViaReadCoordinator(dk, mutations, writePlan, rrSource); } + /* + * Send each per-replica mutation as a separate tracked write via TrackedWriteRequest.perform(). + * We do NOT merge mutations because merging can create oversized mutations that make partitions unreadable. + * Each tracked write gets a proper MutationId and is recorded in the mutation journal. + * + * If a tracked write fails with RetryOnDifferentSystemException or CoordinatorBehindException (migration raced), + * we refetch ClusterMetadata and retry through the correct path (tracked or untracked). + */ + private void repairViaTrackedWrite(DecoratedKey dk, Map mutations, ForWrite writePlan) + { + ReadRepairMetrics.repairedBlockingViaTrackedWrite.mark(); + ConsistencyLevel cl = writePlan.consistencyLevel(); + + List> promises = new ArrayList<>(mutations.size()); + + for (Map.Entry entry : mutations.entrySet()) + { + Replica replica = entry.getKey(); + Mutation mutation = entry.getValue(); + AsyncPromise promise = new AsyncPromise<>(); + promises.add(promise); + startTrackedWriteAttempt(dk, replica, mutation, cl, promise, true); + } + + // Compute blockFor using the same logic as BlockingPartitionRepair: writeQuorum adjusted + // for contacts that don't have mutations to send + int adjustedBlockFor = writePlan.writeQuorum(); + for (Replica participant : writePlan.contacts()) + { + if (!mutations.containsKey(participant)) + adjustedBlockFor--; + } + final int blockFor = adjustedBlockFor; + + repairs.add(new PendingPartitionRepair() + { + @Override + public boolean awaitRepairs(long remaining, TimeUnit timeUnit) throws InterruptedException + { + long deadlineNanos = nanoTime() + timeUnit.toNanos(remaining); + Throwable error = null; + for (AsyncPromise promise : promises) + { + long remainingNanos = deadlineNanos - nanoTime(); + if (remainingNanos <= 0) + return false; + try + { + promise.get(remainingNanos, NANOSECONDS); + } + catch (TimeoutException e) + { + return false; + } + catch (ExecutionException e) + { + error = Throwables.merge(error, e.getCause()); + } + } + Throwables.maybeFail(error); + return true; + } + + @Override + public ForWrite repairPlan() + { + return writePlan; + } + + @Override + public int blockFor() + { + return blockFor; + } + + @Override + public int waitingOn() + { + int count = 0; + for (AsyncPromise promise : promises) + { + if (!promise.isDone()) + count++; + } + return count; + } + }); + } + + /** + * Start a tracked write attempt for a single read repair mutation. + * Wires a callback on the handler that resolves the promise on success, retries on retriable errors, + * or fails the promise on non-retriable errors. + * + * @param allowRetry if true, retriable errors (migration races) will refetch metadata and retry once + * through the correct path; if false, any error fails the promise directly + */ + private void startTrackedWriteAttempt(DecoratedKey dk, Replica replica, Mutation mutation, + ConsistencyLevel cl, AsyncPromise promise, boolean allowRetry) + { + TrackedWriteRequest.perform(mutation, cl, requestTime, handler -> { + try + { + handler.get(); // non-blocking — condition already signaled when callback fires + promise.trySuccess(null); + } + catch (CoordinatorBehindException e) + { + if (allowRetry) + { + logger.debug("Tracked write read repair failed with retriable error, retrying: {}", e.getMessage()); + retryRepairMutation(dk, replica, mutation, cl, promise); + } + else + { + logger.debug("Tracked write read repair retry failed again, not retrying further: {}", e.getMessage()); + promise.tryFailure(e); + } + } + catch (Throwable t) + { + promise.tryFailure(t); + } + }); + } + + /** + * Retry a single read repair mutation after a migration race. Checks updated ClusterMetadata and + * sends the mutation through the correct path (tracked or untracked), resolving the promise + * when complete. + */ + private void retryRepairMutation(DecoratedKey dk, Replica replica, Mutation mutation, + ConsistencyLevel cl, AsyncPromise promise) + { + ClusterMetadata cm = ClusterMetadata.current(); + + if (MigrationRouter.shouldUseTrackedForWrites(cm, command.metadata().keyspace, command.metadata().id, dk.getToken())) + { + throw new IllegalStateException("Retrying BRR for tracked write repair, but it looks like it should still be tracked. " + + "Should be untracked if it needs retry."); + } + else + { + // No longer tracked — send classic READ_REPAIR_REQ and wait for the response + MessagingService.instance().sendWithCallback(Message.outWithRequestTime(READ_REPAIR_REQ, mutation, requestTime), + replica.endpoint(), + new RequestCallback() + { + @Override + public void onResponse(Message msg) + { + promise.trySuccess(null); + } + + @Override + public void onFailure(InetAddressAndPort from, RequestFailure failure) + { + promise.tryFailure(new RuntimeException("Read repair retry failed on " + from + ": " + failure.reason)); + } + + @Override + public boolean invokeOnFailure() + { + return true; + } + }); + } + } + /* * Create a new Accord transaction to apply this blocking read repair ensuring that any data being written * consists of already committed Accord writes just by virtue of creating a new transaction which must occur 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 06a0cb693f..1c500c6b10 100644 --- a/src/java/org/apache/cassandra/service/replication/migration/MutationTrackingMigrationState.java +++ b/src/java/org/apache/cassandra/service/replication/migration/MutationTrackingMigrationState.java @@ -283,6 +283,13 @@ public class MutationTrackingMigrationState implements MetadataValueuntracked is instant and should never create migration state. + checkState(ksm.params.replicationType.isTracked(), + "Migration state exists for untracked keyspace %s; " + + "tracked-to-untracked migration should be instant with no migration state", + keyspace); + // Validate all tables in migration exist in schema for (TableId tableId : info.pendingRangesPerTable.keySet()) { diff --git a/src/java/org/apache/cassandra/tcm/transformations/AlterSchema.java b/src/java/org/apache/cassandra/tcm/transformations/AlterSchema.java index 87ecdf793a..318432e7e0 100644 --- a/src/java/org/apache/cassandra/tcm/transformations/AlterSchema.java +++ b/src/java/org/apache/cassandra/tcm/transformations/AlterSchema.java @@ -61,6 +61,7 @@ import org.apache.cassandra.schema.Tables; import org.apache.cassandra.schema.ViewMetadata; import org.apache.cassandra.schema.Views; import org.apache.cassandra.service.consensus.migration.ConsensusMigrationState; +import org.apache.cassandra.service.replication.migration.KeyspaceMigrationInfo; import org.apache.cassandra.service.replication.migration.MutationTrackingMigrationState; import org.apache.cassandra.tcm.ClusterMetadata; import org.apache.cassandra.tcm.ClusterMetadata.Transformer; @@ -383,15 +384,43 @@ public class AlterSchema implements Transformation // Check if replication type changed if (beforeType != afterType) { - // Auto-start migration for this keyspace - logger.info("Auto-starting mutation tracking migration for keyspace {} (replication_type={})", - diff.after.name, afterType); + if (afterType.isTracked()) + { + // untracked -> tracked: requires migration with repair because tracked reads + // do single-replica data reads and the chosen replica must reflect all prior + // writes for read monotonicity. The writes made while the keyspace was untracked + // are not in the mutation journal, so mutation tracking cannot reconcile them. + // Repair must ensure all replicas are consistent with those pre-tracking writes + // before tracked single-replica reads can safely begin. + logger.info("Auto-starting mutation tracking migration for keyspace {} (replication_type={})", + diff.after.name, afterType); - Collection tableIds = diff.after.tables.stream() - .map(table -> table.id) - .collect(Collectors.toList()); + Collection tableIds = diff.after.tables.stream() + .map(table -> table.id) + .collect(Collectors.toList()); - migrationState = migrationState.withKeyspaceMigrating(diff.after.name, tableIds, nextEpoch); + migrationState = migrationState.withKeyspaceMigrating(diff.after.name, tableIds, nextEpoch); + } + else + { + // tracked -> untracked: instant, no migration needed. Tracked writes are + // already quorum writes and the target state (untracked) just uses regular + // quorum reads, which is strictly less demanding. + KeyspaceMigrationInfo existing = migrationState.getKeyspaceInfo(diff.after.name); + if (existing != null) + { + // An in-progress untracked->tracked migration is being reversed. + // Remove it since tracked->untracked is instant. + logger.info("Removing in-progress migration for keyspace {} (switching to untracked is instant)", + diff.after.name); + migrationState = migrationState.dropKeyspaces(nextEpoch, Collections.singleton(diff.after.name)); + } + else + { + logger.info("Switching keyspace {} to untracked (instant, no migration needed)", + diff.after.name); + } + } } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/MutationTrackingMigrationTest.java b/test/distributed/org/apache/cassandra/distributed/test/MutationTrackingMigrationTest.java index 9b1b6f3c02..c0df2f6779 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/MutationTrackingMigrationTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/MutationTrackingMigrationTest.java @@ -292,49 +292,28 @@ public class MutationTrackingMigrationTest extends TestBaseImpl ConsistencyLevel.QUORUM); assertEquals(100, initialResults.length); - // start migration to untracked replication + // switch to untracked replication — tracked→untracked is instant, no migration needed coordinator.execute(format("ALTER KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3} AND replication_type='untracked'", testKeyspace), ConsistencyLevel.ALL); waitForEpochOf(SHARED_CLUSTER, 1); - verifyKeyspaceState(testKeyspace, ExpectedKeyspaceState.MIGRATING_TO_UNTRACKED); - - long journalEntriesBeforeMigrationWrites = countJournalEntries(); - - // Write more data during migration - for (int i = 100; i < 200; i++) - { - coordinator.execute(format("INSERT INTO %s.%s (pk, value) VALUES (%d, 'migration_%d')", - testKeyspace, TEST_TABLE, i, i), - ConsistencyLevel.QUORUM); - } - - // writes should also be tracked during migration - long journalEntriesAfterMigrationWrites = countJournalEntries(); - assertTrue("Migration writes should still create journal entries (tracked mechanism still active)", - journalEntriesAfterMigrationWrites > journalEntriesBeforeMigrationWrites); - - // complete migration - SHARED_CLUSTER.get(1).nodetoolResult("repair", testKeyspace, TEST_TABLE).asserts().success(); - - waitForEpochOf(SHARED_CLUSTER, 1); - + // Should go directly to UNTRACKED — no migration state verifyKeyspaceState(testKeyspace, ExpectedKeyspaceState.UNTRACKED); long journalEntriesBeforeUntracked = countJournalEntries(); - for (int i = 200; i < 210; i++) + for (int i = 100; i < 210; i++) { coordinator.execute(format("INSERT INTO %s.%s (pk, value) VALUES (%d, 'untracked_%d')", testKeyspace, TEST_TABLE, i, i), ConsistencyLevel.QUORUM); } - // but they should not be tracked after migration + // writes should not be tracked after instant switch to untracked long journalEntriesAfterUntracked = countJournalEntries(); - assertEquals("Post-migration untracked writes should NOT create journal entries", + assertEquals("Post-switch untracked writes should NOT create journal entries", journalEntriesBeforeUntracked, journalEntriesAfterUntracked); Object[][] finalResults = coordinator.execute(format("SELECT * FROM %s.%s", testKeyspace, TEST_TABLE), @@ -344,10 +323,6 @@ public class MutationTrackingMigrationTest extends TestBaseImpl Object[][] initialRecord = coordinator.execute(format("SELECT value FROM %s.%s WHERE pk = 50", testKeyspace, TEST_TABLE), ConsistencyLevel.QUORUM); assertEquals("initial_50", initialRecord[0][0]); - - Object[][] migrationRecord = coordinator.execute(format("SELECT value FROM %s.%s WHERE pk = 150", testKeyspace, TEST_TABLE), - ConsistencyLevel.QUORUM); - assertEquals("migration_150", migrationRecord[0][0]); } @Test @@ -393,20 +368,14 @@ public class MutationTrackingMigrationTest extends TestBaseImpl waitForEpochOf(SHARED_CLUSTER, 1); verifyKeyspaceState(testKeyspace, ExpectedKeyspaceState.MIGRATING_TO_TRACKED); - // Reverse the migration by changing back to untracked + // Reverse the migration by changing back to untracked — tracked→untracked is instant coordinator.execute(format("ALTER KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3} AND replication_type='untracked'", testKeyspace), ConsistencyLevel.ALL); waitForEpochOf(SHARED_CLUSTER, 1); - verifyKeyspaceState(testKeyspace, ExpectedKeyspaceState.MIGRATING_TO_UNTRACKED); - - // Complete the reversed migration - SHARED_CLUSTER.get(1).nodetoolResult("repair", "-pr", testKeyspace, TEST_TABLE).asserts().success(); - - waitForEpochOf(SHARED_CLUSTER, 1); - + // Should go directly to UNTRACKED — no migration state verifyKeyspaceState(testKeyspace, ExpectedKeyspaceState.UNTRACKED); Object[][] results = coordinator.execute(format("SELECT * FROM %s.%s", testKeyspace, TEST_TABLE), @@ -455,20 +424,14 @@ public class MutationTrackingMigrationTest extends TestBaseImpl coordinator.execute(format("INSERT INTO %s.%s (pk, value) VALUES (1, 'new_table_data')", testKeyspace, newTable), ConsistencyLevel.QUORUM); - // Reverse the migration + // Reverse the migration — tracked→untracked is instant coordinator.execute(format("ALTER KEYSPACE %s WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3} AND replication_type='untracked'", testKeyspace), ConsistencyLevel.ALL); waitForEpochOf(SHARED_CLUSTER, 1); - verifyKeyspaceState(testKeyspace, ExpectedKeyspaceState.MIGRATING_TO_UNTRACKED); - - // Complete migration (both tables should be in migration) - SHARED_CLUSTER.get(1).nodetoolResult("repair", testKeyspace).asserts().success(); - - waitForEpochOf(SHARED_CLUSTER, 1); - + // Should go directly to UNTRACKED — no migration state verifyKeyspaceState(testKeyspace, ExpectedKeyspaceState.UNTRACKED); Object[][] results = coordinator.execute(format("SELECT value FROM %s.%s WHERE pk = 1", testKeyspace, newTable), diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingIncrementalRepairTaskTest.java b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingIncrementalRepairTaskTest.java index 44e8424bf9..84606bd36b 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingIncrementalRepairTaskTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingIncrementalRepairTaskTest.java @@ -192,17 +192,18 @@ public class MutationTrackingIncrementalRepairTaskTest extends TestBaseImpl "{'class': 'SimpleStrategy', 'replication_factor': 3} " + "AND replication_type='untracked'"); - // During reverse migration, both should still apply + // During reverse migration, tracked->untracked is instant; no migration state Boolean shouldUseAfter = CLUSTER.get(1).callOnInstance(() -> { ClusterMetadata metadata = ClusterMetadata.current(); return MutationTrackingIncrementalRepairTask.shouldUseMutationTrackingRepair(metadata, ksName); }); assertFalse("Keyspace migrating from tracked should not use mutation tracking repair", shouldUseAfter); + // tracked->untracked is instant, no migration state created Boolean migrationAfter = CLUSTER.get(1).callOnInstance(() -> { ClusterMetadata metadata = ClusterMetadata.current(); return MutationTrackingIncrementalRepairTask.isMutationTrackingMigrationInProgress(metadata, ksName); }); - assertTrue("Keyspace migrating from tracked should have migration in progress", migrationAfter); + assertFalse("Keyspace migrating from tracked should NOT have migration in progress (instant)", migrationAfter); } } diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingReadRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingReadRepairTest.java new file mode 100644 index 0000000000..7134a9d5cd --- /dev/null +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingReadRepairTest.java @@ -0,0 +1,738 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.cassandra.distributed.test.repair; + +import java.io.IOException; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.Callable; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; + +import com.google.common.util.concurrent.ThreadFactoryBuilder; + +import org.junit.After; +import org.junit.AfterClass; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Ignore; +import org.junit.Test; + +import org.apache.cassandra.distributed.Cluster; +import org.apache.cassandra.distributed.api.ConsistencyLevel; +import org.apache.cassandra.distributed.api.Feature; +import org.apache.cassandra.distributed.api.IInvokableInstance; +import org.apache.cassandra.distributed.api.NodeToolResult; +import org.apache.cassandra.distributed.test.TestBaseImpl; +import org.apache.cassandra.gms.EndpointState; +import org.apache.cassandra.gms.FailureDetector; +import org.apache.cassandra.gms.Gossiper; +import org.apache.cassandra.locator.InetAddressAndPort; +import org.apache.cassandra.metrics.ReadRepairMetrics; +import org.apache.cassandra.repair.MutationTrackingIncrementalRepairTask; +import org.apache.cassandra.replication.MutationTrackingService; +import org.apache.cassandra.schema.ReplicationType; +import org.apache.cassandra.tcm.ClusterMetadata; +import org.apache.cassandra.tcm.Epoch; +import org.apache.cassandra.utils.FBUtilities; + +import static org.apache.cassandra.distributed.shared.ClusterUtils.decode; +import static org.apache.cassandra.distributed.shared.ClusterUtils.encode; +import static org.apache.cassandra.distributed.shared.ClusterUtils.getNextEpoch; +import static org.apache.cassandra.distributed.shared.ClusterUtils.pauseBeforeEnacting; +import static org.apache.cassandra.distributed.shared.ClusterUtils.unpauseEnactment; +import static org.apache.cassandra.schema.ReplicationType.tracked; +import static org.apache.cassandra.schema.ReplicationType.untracked; +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +/** + * Tests that blocking read repair works correctly before, during, and after migration + * between untracked and tracked replication types (in both directions). + * + *

+ * When a keyspace uses untracked replication, reads detect inconsistencies and use + * blocking read repair (BRR) via {@code READ_REPAIR_REQ} messages (the classic path). + * When a keyspace uses tracked replication, reads use reconciliation (mutation summaries) + * instead of BRR. During migration, reads for pending ranges use the untracked path + * with BRR, but BRR writes go through the tracked write path so they get proper + * MutationIds and are recorded in the mutation journal. + *

+ * Each test creates a unique keyspace to avoid interference. Keyspaces are not dropped + * between tests to avoid background broadcast races (see MutationTrackingRepairTest). + */ +public class MutationTrackingReadRepairTest extends TestBaseImpl +{ + private static final int NUM_NODES = 3; + private static final List ALL_NODES = List.of(1, 2, 3); + private static final String REPLICATION = "{'class': 'SimpleStrategy', 'replication_factor': 3}"; + + private static Cluster CLUSTER; + private static ExecutorService executor; + private static final AtomicInteger ksCounter = new AtomicInteger(); + + private String ksName; + + @BeforeClass + public static void setupCluster() throws IOException + { + executor = Executors.newCachedThreadPool(new ThreadFactoryBuilder().setDaemon(true).build()); + CLUSTER = Cluster.build() + .withNodes(NUM_NODES) + .withConfig(cfg -> cfg.set("hinted_handoff_enabled", false) + .set("mutation_tracking_sync_timeout", "10s") + .set("request_timeout", "10000ms") + .set("repair.retries.max_attempts", 10) + .set("repair.retries.base_sleep_time", "100ms") + .set("repair.retries.max_sleep_time", "500ms") + .with(Feature.GOSSIP)) + .start(); + } + + @AfterClass + public static void teardownCluster() + { + executor.shutdownNow(); + if (CLUSTER != null) + CLUSTER.close(); + } + + @Before + public void setUp() + { + ksName = "mt_rr_" + ksCounter.incrementAndGet(); + // Pause regular-priority (background write retry) delivery on all nodes so the + // reconciler cannot proactively fix inconsistencies before BRR gets a chance to + // fire. High-priority tasks (needed by tracked read reconciliation) still run, + // so tracked reads work normally. + CLUSTER.forEach(() -> MutationTrackingService.instance().pauseActiveReconcilerRegularPriority()); + } + + @After + public void tearDown() + { + CLUSTER.filters().reset(); + CLUSTER.forEach(() -> MutationTrackingService.instance().resumeActiveReconciler()); + CLUSTER.forEach(() -> MutationTrackingService.instance().resumeActiveReconcilerRegularPriority()); + // Unpause any paused epoch enactment + for (int i = 1; i <= CLUSTER.size(); i++) + unpauseEnactment(CLUSTER.get(i)); + // Re-mark all nodes as alive in case a test isolated one + CLUSTER.forEach(() -> + Gossiper.runInGossipStageBlocking(() -> { + for (var entry : Gossiper.instance.endpointStateMap.entrySet()) + { + InetAddressAndPort ep = entry.getKey(); + EndpointState state = entry.getValue(); + if (!ep.equals(FBUtilities.getBroadcastAddressAndPort()) && !state.isAlive()) + { + FailureDetector.instance.report(ep); + Gossiper.instance.realMarkAlive(ep, state); + } + } + })); + } + + private void createKeyspace(ReplicationType replicationType) + { + CLUSTER.schemaChange("CREATE KEYSPACE " + ksName + " WITH replication = " + + REPLICATION + " AND replication_type='" + replicationType + "'"); + CLUSTER.schemaChange("CREATE TABLE " + ksName + ".tbl (k int PRIMARY KEY, v int)"); + } + + private void alterKeyspace(ReplicationType replicationType) + { + CLUSTER.schemaChange("ALTER KEYSPACE " + ksName + " WITH replication = " + + REPLICATION + " AND replication_type='" + replicationType + "'"); + } + + private void insertConsistent(int start, int count) + { + for (int i = start; i < start + count; i++) + { + CLUSTER.coordinator(1).execute( + "INSERT INTO " + ksName + ".tbl (k, v) VALUES (?, ?)", + ConsistencyLevel.ALL, i, i); + } + } + + /** + * Create inconsistency by isolating a node during QUORUM writes. + * After this call, the isolated node is missing the written data. + */ + private void insertWithInconsistency(int isolatedNode, int start, int count) + { + CLUSTER.filters().allVerbs().to(isolatedNode).drop(); + CLUSTER.filters().allVerbs().from(isolatedNode).drop(); + + for (int i = start; i < start + count; i++) + { + int coordinatorNode = isolatedNode == 1 ? 2 : 1; + CLUSTER.coordinator(coordinatorNode).execute( + "INSERT INTO " + ksName + ".tbl (k, v) VALUES (?, ?)", + ConsistencyLevel.QUORUM, i, i); + } + + // Verify the isolated node actually missed the data + Object[][] results = CLUSTER.get(isolatedNode).executeInternal( + "SELECT k FROM " + ksName + ".tbl WHERE k >= ? AND k < ? ALLOW FILTERING", + start, start + count); + assertEquals("Node " + isolatedNode + " should not have data written while isolated", + 0, results.length); + CLUSTER.filters().reset(); + } + + private void assertNodeMissingData(int node, int start, int count) + { + Object[][] results = CLUSTER.get(node).executeInternal( + "SELECT k FROM " + ksName + ".tbl WHERE k >= ? AND k < ? ALLOW FILTERING", + start, start + count); + assertEquals("Node " + node + " should still be missing data for keys [" + start + ", " + (start + count) + + ") after migration started — if present, data was delivered by an unexpected mechanism", + 0, results.length); + } + + private void assertDataOnAllNodes(int start, int count) + { + for (int node = 1; node <= CLUSTER.size(); node++) + { + for (int i = start; i < start + count; i++) + { + Object[][] results = CLUSTER.get(node).executeInternal( + "SELECT k, v FROM " + ksName + ".tbl WHERE k = ?", i); + assertEquals("Node " + node + " missing row k=" + i, 1, results.length); + assertEquals(i, results[0][0]); + assertEquals(i, results[0][1]); + } + } + } + + private long getRepairedBlocking(int node) + { + //noinspection Convert2MethodRef + return CLUSTER.get(node).callOnInstance(() -> ReadRepairMetrics.repairedBlocking.getCount()); + } + + private long getRepairedBlockingViaTrackedWrite(int node) + { + //noinspection Convert2MethodRef + return CLUSTER.get(node).callOnInstance(() -> ReadRepairMetrics.repairedBlockingViaTrackedWrite.getCount()); + } + + private long getTrackedReconcile(int node) + { + //noinspection Convert2MethodRef + return CLUSTER.get(node).callOnInstance(() -> ReadRepairMetrics.trackedReconcile.getCount()); + } + + private boolean isMigrationInProgress() + { + String ks = ksName; + return CLUSTER.get(1).callOnInstance(() -> { + ClusterMetadata metadata = ClusterMetadata.current(); + return MutationTrackingIncrementalRepairTask.isMutationTrackingMigrationInProgress(metadata, ks); + }); + } + + private NodeToolResult nodetoolRepair(int node, String... args) + { + String[] cmd = new String[args.length + 1]; + cmd[0] = "repair"; + System.arraycopy(args, 0, cmd, 1, args.length); + return CLUSTER.get(node).nodetoolResult(cmd); + } + + private List repairConcurrently(List nodes, String... args) + { + List> futures = new ArrayList<>(); + for (int node : nodes) + futures.add(executor.submit(() -> nodetoolRepair(node, args))); + List results = new ArrayList<>(); + for (Future f : futures) + { + try + { + results.add(f.get(60, TimeUnit.SECONDS)); + } + catch (Exception e) + { + throw new RuntimeException("Repair future failed", e); + } + } + return results; + } + + private void completeMigrationViaRepair() + { + assertTrue("Migration should be in progress before repair", isMigrationInProgress()); + // The mutation tracking sync coordinator needs the regular-priority reconciler to + // deliver mutations between replicas. Temporarily resume it for the duration of repair. + CLUSTER.forEach(() -> MutationTrackingService.instance().resumeActiveReconcilerRegularPriority()); + try + { + List results = repairConcurrently(ALL_NODES, ksName, "-pr"); + for (NodeToolResult r : results) + r.asserts().success(); + assertFalse("Migration should complete after repair", isMigrationInProgress()); + } + finally + { + CLUSTER.forEach(() -> MutationTrackingService.instance().pauseActiveReconcilerRegularPriority()); + } + } + + /** + * Read all specified keys using a mix of point reads and a range scan to exercise + * both single-partition and multi-partition read repair code paths. + * + *

The first half of keys are read via individual point reads ({@code WHERE k = ?}). + * Then a full table scan exercises the range read path, repairing any remaining + * inconsistent partitions (the second half). + * + *

Per-read-command metrics (repairedBlocking) increment by the returned count. + * Per-partition metrics (repairedBlockingViaTrackedWrite) increment by the total + * number of inconsistent partitions regardless of how they were read. + * + * @return the number of CQL read commands issued + */ + private int readAllKeys(int coordinator, int start, int count) + { + int mid = start + count / 2; + + // First half: individual point reads (single-partition read path) + for (int i = start; i < mid; i++) + { + CLUSTER.coordinator(coordinator).execute( + "SELECT k, v FROM " + ksName + ".tbl WHERE k = ?", + ConsistencyLevel.ALL, i); + } + + // Full table scan (multi-partition range read path). + // Point reads above already repaired keys [start, mid). + // The range scan repairs remaining inconsistent partitions. + CLUSTER.coordinator(coordinator).execute( + "SELECT k, v FROM " + ksName + ".tbl", + ConsistencyLevel.ALL); + + return (mid - start) + 1; + } + + /** + * Baseline: BRR in a pure untracked keyspace uses the classic READ_REPAIR_REQ path. + * Verify that repairedBlocking increments and repairedBlockingViaTrackedWrite does NOT. + */ + @Test + public void testBlockingReadRepairUntracked() + { + createKeyspace(untracked); + insertWithInconsistency(3, 0, 10); + + long blockingBefore = getRepairedBlocking(1); + long trackedWriteBefore = getRepairedBlockingViaTrackedWrite(1); + + int readOps = readAllKeys(1, 0, 10); + + long blockingAfter = getRepairedBlocking(1); + long trackedWriteAfter = getRepairedBlockingViaTrackedWrite(1); + assertEquals("repairedBlocking should increment for each BRR read command", + blockingBefore + readOps, blockingAfter); + assertEquals("repairedBlockingViaTrackedWrite should NOT increment for untracked BRR", + trackedWriteBefore, trackedWriteAfter); + assertDataOnAllNodes(0, 10); + } + + /** + * During migration from untracked → tracked, reads for pending ranges use the untracked + * read executor with BRR. BRR detects that writes should go through the tracked path + * and uses repairViaTrackedWrite, incrementing repairedBlockingViaTrackedWrite. + */ + @Test + public void testBlockingReadRepairDuringMigrationUntrackedToTracked() + { + createKeyspace(untracked); + insertWithInconsistency(3, 0, 10); + + // Start migration — all ranges are pending + alterKeyspace(tracked); + assertTrue("Migration should be in progress", isMigrationInProgress()); + + long blockingBefore = getRepairedBlocking(1); + long trackedWriteBefore = getRepairedBlockingViaTrackedWrite(1); + + // Pending ranges use untracked reads, BRR fires, writes go through tracked path + int readOps = readAllKeys(1, 0, 10); + + long blockingAfter = getRepairedBlocking(1); + long trackedWriteAfter = getRepairedBlockingViaTrackedWrite(1); + assertEquals("repairedBlocking should increment for each BRR read command", + blockingBefore + readOps, blockingAfter); + assertEquals("repairedBlockingViaTrackedWrite should increment for all repaired partitions", + trackedWriteBefore + 10, trackedWriteAfter); + assertDataOnAllNodes(0, 10); + + completeMigrationViaRepair(); + assertDataOnAllNodes(0, 10); + } + + /** + * After migration to tracked completes, reads use TrackedReadExecutor with + * reconciliation — not BRR. Verify trackedReconcile increments and repairedBlocking does not. + */ + @Test + public void testReconciliationAfterMigrationToTracked() + { + createKeyspace(untracked); + insertConsistent(0, 5); + + alterKeyspace(tracked); + completeMigrationViaRepair(); + + // Inconsistency goes through tracked coordination (gets MutationId) but node 3 misses the write + insertWithInconsistency(3, 100, 10); + + long blockingBefore = getRepairedBlocking(1); + long reconcileBefore = getTrackedReconcile(1); + + // Fully tracked — uses TrackedReadExecutor + reconciliation, not BRR + int readOps = readAllKeys(1, 100, 10); + + long blockingAfter = getRepairedBlocking(1); + long reconcileAfter = getTrackedReconcile(1); + + assertEquals("repairedBlocking should NOT increment for fully tracked reads", + blockingBefore, blockingAfter); + assertEquals("trackedReconcile should increment for each tracked read", + reconcileBefore + readOps, reconcileAfter); + + assertDataOnAllNodes(0, 5); + assertDataOnAllNodes(100, 10); + } + + /** + * After switching from tracked → untracked (instant, no migration), reads use the + * untracked executor with classic BRR via {@code READ_REPAIR_REQ} messages. + * Since the keyspace is fully untracked, repairedBlockingViaTrackedWrite should NOT increment. + * Tests both inconsistencies created while tracked (with MutationIds) and inconsistencies + * created after switching to untracked (without MutationIds). + */ + @Test + public void testBlockingReadRepairAfterSwitchToUntracked() + { + createKeyspace(tracked); + insertConsistent(0, 5); + + // Inconsistency created while tracked — mutations have MutationIds + insertWithInconsistency(3, 100, 10); + + // Switch to untracked — instant, no migration + alterKeyspace(untracked); + assertFalse("Migration should NOT be in progress (tracked->untracked is instant)", + isMigrationInProgress()); + + // Inconsistency created after switch — mutations have no MutationIds + insertWithInconsistency(3, 200, 10); + + long blockingBefore = getRepairedBlocking(1); + long trackedWriteBefore = getRepairedBlockingViaTrackedWrite(1); + + // Read first range with point reads only — a table scan would cross-repair the + // second range, preventing us from verifying BRR fires for both ranges independently. + int readOps = 0; + for (int i = 100; i < 110; i++) + { + CLUSTER.coordinator(1).execute( + "SELECT k, v FROM " + ksName + ".tbl WHERE k = ?", + ConsistencyLevel.ALL, i); + readOps++; + } + + // Second range via readAllKeys (includes point reads + table scan) + readOps += readAllKeys(1, 200, 10); + + long blockingAfter = getRepairedBlocking(1); + long trackedWriteAfter = getRepairedBlockingViaTrackedWrite(1); + + assertEquals("repairedBlocking should increment for each BRR read command", + blockingBefore + readOps, blockingAfter); + assertEquals("repairedBlockingViaTrackedWrite should NOT increment (fully untracked)", + trackedWriteBefore, trackedWriteAfter); + assertDataOnAllNodes(100, 10); + assertDataOnAllNodes(200, 10); + } + + /** + * Round trip: untracked → tracked → untracked. + * Verify BRR uses the correct path at each stage. + */ + @Test + public void testRoundTripUntrackedToTrackedToUntracked() + { + createKeyspace(untracked); + + // Phase 1: BRR while untracked (classic path) + insertWithInconsistency(3, 0, 5); + long blockingBefore = getRepairedBlocking(1); + long trackedWriteBefore = getRepairedBlockingViaTrackedWrite(1); + int readOps = readAllKeys(1, 0, 5); + assertEquals("Phase 1: repairedBlocking should increment for each BRR read command", + blockingBefore + readOps, getRepairedBlocking(1)); + assertEquals("Phase 1: trackedWrite should NOT increment", + trackedWriteBefore, getRepairedBlockingViaTrackedWrite(1)); + assertDataOnAllNodes(0, 5); + + // Phase 2: Migrate to tracked, BRR during migration uses tracked writes + insertWithInconsistency(3, 100, 5); + alterKeyspace(tracked); + assertTrue("Migration to tracked should be in progress", isMigrationInProgress()); + + blockingBefore = getRepairedBlocking(1); + trackedWriteBefore = getRepairedBlockingViaTrackedWrite(1); + readOps = readAllKeys(1, 100, 5); + assertEquals("Phase 2: repairedBlocking should increment for each BRR read command", + blockingBefore + readOps, getRepairedBlocking(1)); + assertEquals("Phase 2: trackedWrite should increment for all repaired partitions", + trackedWriteBefore + 5, getRepairedBlockingViaTrackedWrite(1)); + assertDataOnAllNodes(100, 5); + + completeMigrationViaRepair(); + + // Phase 3: Fully tracked — uses reconciliation, not BRR + insertWithInconsistency(3, 200, 5); + blockingBefore = getRepairedBlocking(1); + long reconcileBefore = getTrackedReconcile(1); + readOps = readAllKeys(1, 200, 5); + assertEquals("Phase 3: repairedBlocking should NOT increment for tracked reads", + blockingBefore, getRepairedBlocking(1)); + assertEquals("Phase 3: trackedReconcile should increment for each tracked read", + reconcileBefore + readOps, getTrackedReconcile(1)); + + // Phase 4: Migrate back to untracked — instant, no migration needed + insertWithInconsistency(3, 300, 5); + alterKeyspace(untracked); + assertFalse("Migration to untracked should NOT be in progress (instant)", isMigrationInProgress()); + + blockingBefore = getRepairedBlocking(1); + trackedWriteBefore = getRepairedBlockingViaTrackedWrite(1); + readOps = readAllKeys(1, 300, 5); + assertEquals("Phase 4: repairedBlocking should increment for each BRR read command", + blockingBefore + readOps, getRepairedBlocking(1)); + assertEquals("Phase 4: trackedWrite should NOT increment (fully untracked)", + trackedWriteBefore, getRepairedBlockingViaTrackedWrite(1)); + + // Phase 5: Fully untracked again — classic BRR + insertWithInconsistency(3, 400, 5); + blockingBefore = getRepairedBlocking(1); + trackedWriteBefore = getRepairedBlockingViaTrackedWrite(1); + readOps = readAllKeys(1, 400, 5); + assertEquals("Phase 5: repairedBlocking should increment for each BRR read command", + blockingBefore + readOps, getRepairedBlocking(1)); + assertEquals("Phase 5: trackedWrite should NOT increment after migration complete", + trackedWriteBefore, getRepairedBlockingViaTrackedWrite(1)); + assertDataOnAllNodes(400, 5); + } + + /** + * Round trip: tracked → untracked → tracked. + * Verify BRR uses the correct path at each stage. + * + * NOTE: The second completeMigrationViaRepair (untracked→tracked) is known to hit + * "Could not find shard for logId" during SSTable flush when shards from the first + * tracked phase were cleaned up. This is a pre-existing migration infrastructure issue. + * This test therefore only verifies the BRR behavior up through the untracked phase + * and does not attempt the second migration completion. + */ + @Test + public void testRoundTripTrackedToUntrackedToTracked() + { + createKeyspace(tracked); + + // Phase 1: Fully tracked — reconciliation, not BRR + insertWithInconsistency(3, 0, 5); + long blockingBefore = getRepairedBlocking(1); + long reconcileBefore = getTrackedReconcile(1); + int readOps = readAllKeys(1, 0, 5); + assertEquals("Phase 1: repairedBlocking should NOT increment for tracked reads", + blockingBefore, getRepairedBlocking(1)); + assertEquals("Phase 1: trackedReconcile should increment for each tracked read", + reconcileBefore + readOps, getTrackedReconcile(1)); + + // Phase 2: Switch to untracked — instant, no migration + insertWithInconsistency(3, 100, 5); + alterKeyspace(untracked); + assertFalse("Migration to untracked should NOT be in progress (instant)", isMigrationInProgress()); + + blockingBefore = getRepairedBlocking(1); + long trackedWriteBefore = getRepairedBlockingViaTrackedWrite(1); + readOps = readAllKeys(1, 100, 5); + assertEquals("Phase 2: repairedBlocking should increment for each BRR read command", + blockingBefore + readOps, getRepairedBlocking(1)); + assertEquals("Phase 2: trackedWrite should NOT increment (fully untracked)", + trackedWriteBefore, getRepairedBlockingViaTrackedWrite(1)); + + // Phase 3: Fully untracked — classic BRR + insertWithInconsistency(3, 200, 5); + blockingBefore = getRepairedBlocking(1); + trackedWriteBefore = getRepairedBlockingViaTrackedWrite(1); + readOps = readAllKeys(1, 200, 5); + assertEquals("Phase 3: repairedBlocking should increment for each BRR read command", + blockingBefore + readOps, getRepairedBlocking(1)); + assertEquals("Phase 3: trackedWrite should NOT increment for fully untracked", + trackedWriteBefore, getRepairedBlockingViaTrackedWrite(1)); + assertDataOnAllNodes(200, 5); + + // Phase 4: Start migration back to tracked, BRR uses tracked writes + insertWithInconsistency(3, 300, 5); + alterKeyspace(tracked); + assertTrue("Migration to tracked should be in progress", isMigrationInProgress()); + + // Verify node 3 is still missing the data after migration starts. + // If this fails, something delivered the data between insertWithInconsistency + // and the reads below (e.g. reconciler, delayed message, epoch-triggered sync). + assertNodeMissingData(3, 300, 5); + + blockingBefore = getRepairedBlocking(1); + trackedWriteBefore = getRepairedBlockingViaTrackedWrite(1); + long reconcileBefore4 = getTrackedReconcile(1); + readOps = readAllKeys(1, 300, 5); + long blockingAfter = getRepairedBlocking(1); + long trackedWriteAfter = getRepairedBlockingViaTrackedWrite(1); + long reconcileAfter4 = getTrackedReconcile(1); + assertEquals("Phase 4: repairedBlocking should increment for each BRR read command" + + " (trackedReconcile delta=" + (reconcileAfter4 - reconcileBefore4) + + ", trackedWrite delta=" + (trackedWriteAfter - trackedWriteBefore) + ")", + blockingBefore + readOps, blockingAfter); + assertEquals("Phase 4: trackedWrite should increment for all repaired partitions", + trackedWriteBefore + 5, trackedWriteAfter); + assertDataOnAllNodes(300, 5); + + // TODO: The second completeMigrationViaRepair (untracked→tracked) hits + // "Could not find shard for logId" in MutationTrackingService.isDurablyReconciled() + // during SSTable flush when shards from the first tracked phase were cleaned up + // during the tracked→untracked migration. This is a pre-existing migration + // infrastructure issue. Uncomment once fixed. + // + // Stack trace: + // java.lang.IllegalStateException: Could not find shard for logId 4294967300 + // at o.a.c.replication.MutationTrackingService.isDurablyReconciled(MutationTrackingService.java:827) + // at o.a.c.io.sstable.format.SSTableWriter.finalizeMetadata(SSTableWriter.java:361) + // at o.a.c.io.sstable.format.SSTableWriter$TransactionalProxy.doPrepare(SSTableWriter.java:418) + // at o.a.c.db.ColumnFamilyStore$Flush.flushMemtable(ColumnFamilyStore.java:1348) + // +// // Complete migration to tracked +// completeMigrationViaRepair(); +// +// // Phase 5: Fully tracked again — reconciliation, not BRR +// insertWithInconsistency(3, 400, 5); +// blockingBefore = getRepairedBlocking(1); +// reconcileBefore = getTrackedReconcile(1); +// readOps = readAllKeys(1, 400, 5); +// assertEquals("Phase 5: repairedBlocking should NOT increment for tracked reads", +// blockingBefore, getRepairedBlocking(1)); +// assertEquals("Phase 5: trackedReconcile should increment for each tracked read", +// reconcileBefore + readOps + 1, getTrackedReconcile(1)); + } + + /** + * Test that the CoordinatorBehindException retry path in BlockingReadRepair.startTrackedWriteAttempt + * works correctly. When the coordinator is behind on epoch (still in migration state while + * replicas have reverted to untracked), the tracked write read repair gets a CoordinatorBehindException + * from the receiving node via checkReplicationMigration. The retry logic refetches ClusterMetadata, + * sees that writes are no longer tracked, and falls back to classic READ_REPAIR_REQ. + * + * Setup: + * 1. Create untracked keyspace, write consistent data + * 2. Start migration to tracked — all nodes see migration in progress + * 3. Pause node 3 (non-CMS-leader) before the next epoch + * 4. ALTER KEYSPACE back to untracked from node 1 (CMS leader) — nodes 1, 2 advance + * 5. Create inconsistency via executeInternal + * 6. Unpause node 3, read from node 3 — BRR fires, node 3 still thinks migration in progress + * 7. Tracked write carries old epoch; receiving node is now untracked → CoordinatorBehindException + * 8. Retry logic refetches metadata, sees untracked, falls back to classic READ_REPAIR_REQ + */ + @Ignore("https://issues.apache.org/jira/browse/CASSANDRA-21310") + @Test + public void testTrackedWriteReadRepairRetryOnCoordinatorBehind() throws Throwable + { + createKeyspace(untracked); + insertConsistent(0, 5); + + // Start migration to tracked — all nodes see this epoch + alterKeyspace(tracked); + assertTrue("Migration should be in progress", isMigrationInProgress()); + + // Pause node 3 before the next epoch — it will remain in migration-in-progress state. + // Node 1 is the CMS leader and must stay operational for the ALTER to succeed. + IInvokableInstance behindNode = CLUSTER.get(3); + Epoch nextEpoch = getNextEpoch(behindNode); + Callable paused = pauseBeforeEnacting(behindNode, nextEpoch); + + // ALTER KEYSPACE back to untracked from node 1 (CMS leader). Uses schemaChange(query, node) + // which calls schemaChangeInternal directly, avoiding the schema agreement wait that would + // hang because node 3 is paused. + CLUSTER.schemaChange("ALTER KEYSPACE " + ksName + " WITH replication = " + + REPLICATION + " AND replication_type='untracked'", 1); + + // Wait for the pause to trigger — node 3 has received the log entry but hasn't enacted + paused.call(); + + // Verify: node 3 is behind nodes 1, 2 + assertTrue("Node 3 should be behind node 1", + decode(CLUSTER.get(3).callOnInstance(() -> encode(ClusterMetadata.current().epoch))) + .isBefore(decode(CLUSTER.get(1).callOnInstance(() -> encode(ClusterMetadata.current().epoch))))); + + // Create inconsistency: write new values to nodes 1 and 3, leaving node 2 stale. + String table = ksName + ".tbl"; + for (int i = 0; i < 5; i++) + { + int newValue = i + 100; + CLUSTER.get(1).executeInternal("INSERT INTO " + table + " (k, v) VALUES (?, ?)", i, newValue); + CLUSTER.get(3).executeInternal("INSERT INTO " + table + " (k, v) VALUES (?, ?)", i, newValue); + } + + // Unpause node 3 so it can catch up during the retry after CoordinatorBehindException. + unpauseEnactment(behindNode); + + // Read from node 3 (coordinator). Node 3 still thinks migration is in progress, + // so BRR uses tracked writes. The tracked write carries node 3's old epoch; + // nodes 1, 2 (at new epoch, untracked) see routing mismatch → CoordinatorBehindException. + // Retry: node 3 catches up, sees untracked → falls back to classic READ_REPAIR_REQ. + for (int i = 0; i < 5; i++) + { + CLUSTER.coordinator(3).execute( + "SELECT k, v FROM " + ksName + ".tbl WHERE k = ?", + ConsistencyLevel.ALL, i); + } + + // TODO This doesn't actually validate the retry path was taken + // Verify all nodes have the updated values + for (int i = 0; i < 5; i++) + { + for (int node = 1; node <= CLUSTER.size(); node++) + { + Object[][] results = CLUSTER.get(node).executeInternal( + "SELECT v FROM " + table + " WHERE k = ?", i); + assertEquals("Node " + node + " should have updated value for k=" + i, + i + 100, results[0][0]); + } + } + } +} diff --git a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java index 7e6399c0a0..3b8eeac91d 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/repair/MutationTrackingRepairTest.java @@ -29,6 +29,8 @@ import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicInteger; import java.util.function.BiPredicate; +import com.google.common.util.concurrent.ThreadFactoryBuilder; + import org.junit.After; import org.junit.AfterClass; import org.junit.Before; @@ -86,16 +88,17 @@ public class MutationTrackingRepairTest extends TestBaseImpl @BeforeClass public static void setupCluster() throws IOException { - executor = Executors.newCachedThreadPool(); + executor = Executors.newCachedThreadPool(new ThreadFactoryBuilder().setDaemon(true).build()); CLUSTER = Cluster.build() - .withNodes(NUM_NODES) - .withConfig(cfg -> cfg.set("mutation_tracking_sync_timeout", "10s") - .set("request_timeout", "1000ms") - .set("repair.retries.max_attempts", 10) - .set("repair.retries.base_sleep_time", "100ms") - .set("repair.retries.max_sleep_time", "500ms") - .with(Feature.GOSSIP, Feature.NETWORK)) - .start(); + .withNodes(NUM_NODES) + .withConfig(cfg -> cfg.set("hinted_handoff_enabled", false) + .set("mutation_tracking_sync_timeout", "10s") + .set("request_timeout", "1000ms") + .set("repair.retries.max_attempts", 10) + .set("repair.retries.base_sleep_time", "100ms") + .set("repair.retries.max_sleep_time", "500ms") + .with(Feature.GOSSIP, Feature.NETWORK)) + .start(); } @AfterClass @@ -120,23 +123,19 @@ public class MutationTrackingRepairTest extends TestBaseImpl public void tearDown() { CLUSTER.filters().reset(); - for (int i = 1; i <= CLUSTER.size(); i++) - { - CLUSTER.get(i).runOnInstance(() -> { - Gossiper.runInGossipStageBlocking(() -> { - for (var entry : Gossiper.instance.endpointStateMap.entrySet()) + CLUSTER.forEach(() -> + Gossiper.runInGossipStageBlocking(() -> { + for (var entry : Gossiper.instance.endpointStateMap.entrySet()) + { + InetAddressAndPort ep = entry.getKey(); + EndpointState state = entry.getValue(); + if (!ep.equals(FBUtilities.getBroadcastAddressAndPort()) && !state.isAlive()) { - InetAddressAndPort ep = entry.getKey(); - EndpointState state = entry.getValue(); - if (!ep.equals(FBUtilities.getBroadcastAddressAndPort()) && !state.isAlive()) - { - FailureDetector.instance.report(ep); - Gossiper.instance.realMarkAlive(ep, state); - } + FailureDetector.instance.report(ep); + Gossiper.instance.realMarkAlive(ep, state); } - }); - }); - } + } + })); } private void setupUntracked() @@ -239,10 +238,7 @@ public class MutationTrackingRepairTest extends TestBaseImpl { List> futures = new ArrayList<>(); for (int node : nodes) - { - int n = node; - futures.add(executor.submit(() -> nodetoolRepair(n, args))); - } + futures.add(executor.submit(() -> nodetoolRepair(node, args))); List results = new ArrayList<>(); for (Future f : futures) { @@ -291,10 +287,7 @@ public class MutationTrackingRepairTest extends TestBaseImpl List> futures = new ArrayList<>(); for (int node : nodes) - { - int n = node; - futures.add(executor.submit(() -> nodetoolRepair(n, args))); - } + futures.add(executor.submit(() -> nodetoolRepair(node, args))); Thread.sleep(2000); assertTrue("Repair should be blocked while node " + isolatedNode + " is isolated", @@ -538,9 +531,27 @@ public class MutationTrackingRepairTest extends TestBaseImpl } @Test - public void testDataAccessibleDuringMigrationToUntracked() throws Exception + public void testDataAccessibleAfterSwitchToUntracked() { - dataAccessibleDuringMigration(() -> alterKeyspaceToUntracked()); + insertDataWithInconsistency("tbl", 0, 50); + + alterKeyspaceToUntracked(); + assertFalse("Migration should NOT be in progress (tracked->untracked is instant)", + isMigrationInProgress()); + + // Read at CL.ALL triggers blocking read repair to fix inconsistencies + Object[][] results = CLUSTER.coordinator(1).execute( + "SELECT k, v FROM " + ksName + ".tbl", ConsistencyLevel.ALL); + assertEquals("Pre-switch data should be readable", 50, results.length); + + // Write and read more data + insertData("tbl", 50, 50); + + results = CLUSTER.coordinator(1).execute( + "SELECT k, v FROM " + ksName + ".tbl", ConsistencyLevel.ALL); + assertEquals("All data should be readable after switch", 100, results.length); + + assertDataOnAllNodes("tbl", 0, 100); } private void dataAccessibleDuringMigration(Runnable alterKeyspace) throws Exception @@ -575,15 +586,14 @@ public class MutationTrackingRepairTest extends TestBaseImpl } @Test - public void testMigrationTrackedToUntrackedCompletesViaRepair() throws Exception + public void testMigrationTrackedToUntrackedIsInstant() { - insertDataWithInconsistency("tbl", 0, 100); + insertData("tbl", 0, 100); alterKeyspaceToUntracked(); - assertTrue("Migration should be in progress after ALTER", isMigrationInProgress()); - - repairResolvingInconsistency(ksName); - assertTrue("Migration should complete after repair", isMigrationComplete()); + assertFalse("Migration should NOT be in progress after ALTER (tracked->untracked is instant)", + isMigrationInProgress()); + assertTrue("Migration should be complete", isMigrationComplete()); assertDataOnAllNodes("tbl", 0, 100); } diff --git a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java index 5757d8cb00..6d04864dc3 100644 --- a/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java +++ b/test/distributed/org/apache/cassandra/distributed/test/tracking/MutationTrackingPendingReadTest.java @@ -63,7 +63,6 @@ import org.apache.cassandra.utils.FBUtilities; import static java.lang.String.format; import static org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils.assertIdsForKey; import static org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils.assertMatchingSummaryIdSpaceForKey; -import static org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils.row; import static org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils.summaryForKey; import static org.apache.cassandra.distributed.test.tracking.MutationTrackingUtils.summaryIdSpace; import static org.apache.cassandra.utils.ByteBufferUtil.bytes; diff --git a/test/unit/org/apache/cassandra/service/replication/migration/MigrationRouterTest.java b/test/unit/org/apache/cassandra/service/replication/migration/MigrationRouterTest.java index fe11cd4559..767d1d7c6f 100644 --- a/test/unit/org/apache/cassandra/service/replication/migration/MigrationRouterTest.java +++ b/test/unit/org/apache/cassandra/service/replication/migration/MigrationRouterTest.java @@ -394,33 +394,31 @@ public class MigrationRouterTest assertTrue(splits.get(2).useTracked); } + /** + * tracked→untracked migration is instant (no migration state). Attempting to construct + * ClusterMetadata with migration state for an untracked keyspace should be rejected. + */ @Test - public void testToUntrackedDirection_CorrectProtocols() + public void testMigrationStateRejectedForUntrackedKeyspace() { - // For migration to untracked: all reads use untracked (no splitting needed) - // This matches single partition behavior: shouldUseTrackedForReads returns false for all ranges - Token queryStart = createToken(-800L); - Token queryEnd = createToken(800L); - Token pendingStart = createToken(0L); - Token pendingEnd = createToken(200L); + ClusterMetadata metadata = new ClusterMetadata(partitioner); + KeyspaceMetadata ksm = createKeyspaceMetadata(TEST_KEYSPACE, ReplicationType.untracked, TEST_TABLE); + metadata = withKeyspace(metadata, ksm); - Range pendingRange = new Range<>(pendingStart, pendingEnd); + Range pendingRange = new Range<>(createToken(-200L), createToken(200L)); + KeyspaceMigrationInfo migrationInfo = createMigrationInfo(ksm, Collections.singletonList(pendingRange)); + MutationTrackingMigrationState migrationState = new MutationTrackingMigrationState( + Epoch.create(1), Collections.singletonMap(TEST_KEYSPACE, migrationInfo)); - ClusterMetadata metadata = createMetadata(false, Collections.singletonList(pendingRange)); - TableMetadata testTable = metadata.schema.getKeyspaceMetadata(TEST_KEYSPACE).getTableOrViewNullable(TEST_TABLE); - - PartitionRangeReadCommand command = createRangeCommand(testTable, queryStart, queryEnd); - List splits = MigrationRouter.splitRangeRead(metadata, command); - - // Should have 1 split (no splitting needed - all reads use untracked) - assertEquals(1, splits.size()); - - // Entire range: untracked - assertFalse(splits.get(0).useTracked); - - // Verify it covers the entire query range - assertEquals(queryStart, splits.get(0).read.dataRange().startKey().getToken()); - assertEquals(queryEnd, splits.get(0).read.dataRange().stopKey().getToken()); + try + { + metadata.transformer().with(migrationState).build(); + Assert.fail("Should have thrown IllegalStateException for migration state on untracked keyspace"); + } + catch (IllegalStateException e) + { + assertTrue(e.getMessage().contains("tracked-to-untracked migration should be instant")); + } } @Test @@ -592,60 +590,6 @@ public class MigrationRouterTest assertTrue(MigrationRouter.shouldUseTrackedForWrites(metadata, TEST_KEYSPACE, testTable.id, tokenOutsidePending)); } - /** - * Test write routing through MigrationRouter for migration to untracked replication. - * Writes use tracked for tokens in pending ranges (still migrating),untracked for completed ranges. - */ - @Test - public void testWriteRoutingToUntracked_PerRangeRouting() - { - Token tokenInPending = createToken(0L); - Token tokenOutsidePending = createToken(500L); - - Range pendingRange = new Range<>(createToken(-200L), createToken(200L)); - - ClusterMetadata metadata = createMetadata(false, Collections.singletonList(pendingRange)); - TableMetadata testTable = metadata.schema.getKeyspaceMetadata(TEST_KEYSPACE).getTableOrViewNullable(TEST_TABLE); - - assertTrue(MigrationRouter.shouldUseTrackedForWrites(metadata, TEST_KEYSPACE, testTable.id, tokenInPending)); - assertFalse(MigrationRouter.shouldUseTrackedForWrites(metadata, TEST_KEYSPACE, testTable.id, tokenOutsidePending)); - } - - @Test - public void testMultiTableMutationRouting_ToUntracked() - { - ClusterMetadata metadata = new ClusterMetadata(partitioner); - - KeyspaceMetadata ksm = createKeyspaceMetadata(TEST_KEYSPACE, ReplicationType.untracked, "table1", "table2"); - metadata = withKeyspace(metadata, ksm); - - TableMetadata table1 = ksm.getTableNullable("table1"); - TableMetadata table2 = ksm.getTableNullable("table2"); - - ClusterMetadata.Transformer transformer = metadata.transformer(); - - // table1 migrating to untracked, table2 complete - MutationTrackingMigrationState migrationState = metadata.mutationTrackingMigrationState.withKeyspaceMigrating(ksm.name, Collections.singleton(table1.id), transformer.epoch()); - metadata = transformer.with(migrationState).build().metadata; - - // Create a mutation with both tables - DecoratedKey key = partitioner.decorateKey(UTF8Type.instance.decompose("key")); - PartitionUpdate update1 = PartitionUpdate.emptyUpdate(table1, key); - PartitionUpdate update2 = PartitionUpdate.emptyUpdate(table2, key); - - Mutation mutation = new Mutation(MutationId.none(), TEST_KEYSPACE, key, ImmutableMap.of(table1.id, update1, table2.id, update2), Clock.Global.nanoTime(), ReadCommand.PotentialTxnConflicts.ALLOW); - - MigrationRouter.RoutedMutations routed = MigrationRouter.routeMutations(metadata, Collections.singletonList(mutation)); - - Mutation trackedMutation = (Mutation) routed.trackedMutations.get(0); - Mutation untrackedMutation = (Mutation) routed.untrackedMutations.get(0); - - // table 1 is still migrating, so it should be in the tracked mutation - assertEquals(Collections.singleton(table1.id), trackedMutation.getTableIds()); - - // table 2 is done migrating, so it should appear in the untracked mutation - assertEquals(Collections.singleton(table2.id), untrackedMutation.getTableIds()); - } /** * Test mutation routing with multiple tables - some tracked, some untracked. diff --git a/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java b/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java index 80409b9515..5c9581b71d 100644 --- a/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java +++ b/test/unit/org/apache/cassandra/tcm/transformations/AlterSchemaMutationTrackingTest.java @@ -104,7 +104,7 @@ public class AlterSchemaMutationTrackingTest } @Test - public void testAutoStartToUntrackedMigration() throws Throwable + public void testTrackedToUntrackedIsInstant() throws Throwable { String ksName = nextKsName(); // tracked replication @@ -118,25 +118,12 @@ public class AlterSchemaMutationTrackingTest ClusterMetadata metadata = ClusterMetadata.current(); assertFalse(metadata.mutationTrackingMigrationState.isMigrating(ksName)); - // Alter keyspace to untracked + // Alter keyspace to untracked — should be instant with no migration state schemaChange(String.format("ALTER KEYSPACE %s WITH replication_type = 'untracked'", ksName)); metadata = ClusterMetadata.current(); - TableId tableId = metadata.schema.getKeyspaceMetadata(ksName).getTableOrViewNullable("tbl").id; - Range fullRing = new Range<>(partitioner.getMinimumToken(), partitioner.getMinimumToken()); - - MutationTrackingMigrationState actualState = metadata.mutationTrackingMigrationState; - KeyspaceMigrationInfo actualInfo = actualState.getKeyspaceInfo(ksName); - - MutationTrackingMigrationState expectedState = createExpectedState( - actualState.lastModified, - ksName, - tableId, - fullRing, - actualInfo.startedAtEpoch - ); - - assertStatesEqual(expectedState, actualState, ksName); + assertFalse("tracked->untracked should be instant with no migration state", + metadata.mutationTrackingMigrationState.isMigrating(ksName)); } @Test @@ -183,29 +170,31 @@ public class AlterSchemaMutationTrackingTest // Alter to tracked schemaChange(String.format("ALTER KEYSPACE %s WITH replication_type = 'tracked'", ks1)); - // Alter to untracked + // Alter to untracked — instant, no migration state schemaChange(String.format("ALTER KEYSPACE %s WITH replication_type = 'untracked'", ks2)); ClusterMetadata metadata = ClusterMetadata.current(); Range fullRing = new Range<>(partitioner.getMinimumToken(), partitioner.getMinimumToken()); TableId table1Id = metadata.schema.getKeyspaceMetadata(ks1).getTableOrViewNullable("tbl").id; - TableId table2Id = metadata.schema.getKeyspaceMetadata(ks2).getTableOrViewNullable("tbl").id; MutationTrackingMigrationState actualState = metadata.mutationTrackingMigrationState; KeyspaceMigrationInfo actual1 = actualState.getKeyspaceInfo(ks1); - KeyspaceMigrationInfo actual2 = actualState.getKeyspaceInfo(ks2); - ImmutableMap.Builder expectedKeyspaces = ImmutableMap.builder(); - expectedKeyspaces.put(ks1, createKeyspaceMigrationInfo(ks1, table1Id, fullRing, actual1.startedAtEpoch)); - expectedKeyspaces.put(ks2, createKeyspaceMigrationInfo(ks2, table2Id, fullRing, actual2.startedAtEpoch)); - - MutationTrackingMigrationState expectedState = new MutationTrackingMigrationState( + // ks1 (untracked→tracked) should have migration state + MutationTrackingMigrationState expectedState = createExpectedState( actualState.lastModified, - expectedKeyspaces.build() + ks1, + table1Id, + fullRing, + actual1.startedAtEpoch ); - assertStatesEqual(expectedState, actualState, ks1, ks2); + assertStatesEqual(expectedState, actualState, ks1); + + // ks2 (tracked→untracked) should have no migration state — instant transition + assertFalse("tracked->untracked should be instant with no migration state", + actualState.isMigrating(ks2)); } @Test