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