CEP-45: Incremental repair for mutation tracking

src/java/org/apache/cassandra/config/Config.java

@@ -169,6 +169,8 @@ public static Set<String> splitCommaDelimited(String src)
     @Replaces(oldName = "repair_request_timeout_in_ms", converter = Converters.MILLIS_DURATION_LONG, deprecated = true)
     public volatile DurationSpec.LongMillisecondsBound repair_request_timeout = new DurationSpec.LongMillisecondsBound("120000ms");
 
+    public volatile DurationSpec.LongMillisecondsBound mutation_tracking_sync_timeout = new DurationSpec.LongMillisecondsBound("2m");
+
     public Integer streaming_connections_per_host = 1;
     @Replaces(oldName = "streaming_keep_alive_period_in_secs", converter = Converters.SECONDS_DURATION, deprecated = true)
     public DurationSpec.IntSecondsBound streaming_keep_alive_period = new DurationSpec.IntSecondsBound("300s");

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -2592,6 +2592,16 @@ public static void setRepairRpcTimeout(Long timeOutInMillis)
         conf.repair_request_timeout = new DurationSpec.LongMillisecondsBound(timeOutInMillis);
     }
 
+    public static long getMutationTrackingSyncTimeout(TimeUnit unit)
+    {
+        return conf.mutation_tracking_sync_timeout.to(unit);
+    }
+
+    public static void setMutationTrackingSyncTimeout(long timeoutInMillis)
+    {
+        conf.mutation_tracking_sync_timeout = new DurationSpec.LongMillisecondsBound(timeoutInMillis);
+    }
+
     public static boolean hasCrossNodeTimeout()
     {
         return conf.internode_timeout;

src/java/org/apache/cassandra/db/AbstractMutationVerbHandler.java

@@ -202,6 +202,13 @@ else if (message.epoch().isBefore(metadata.schema.lastModified()))
      */
     protected ClusterMetadata checkReplicationMigration(ClusterMetadata metadata, Message<T> 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))

src/java/org/apache/cassandra/db/CassandraKeyspaceWriteHandler.java

@@ -47,7 +47,8 @@ public WriteContext beginWrite(Mutation mutation, boolean makeDurable) throws Re
         {
             group = Keyspace.writeOrder.start();
 
-            MigrationRouter.validateUntrackedMutation(mutation);
+            if (!mutation.isReadRepair())
+                MigrationRouter.validateUntrackedMutation(mutation);
             // write the mutation to the commitlog and memtables
             CommitLogPosition position = null;
             if (makeDurable)

src/java/org/apache/cassandra/db/Keyspace.java

@@ -437,7 +437,7 @@ public void apply(final Mutation mutation,
                       boolean updateIndexes,
                       boolean isDroppable)
     {
-        if (MigrationRouter.isFullyTracked(mutation))
+        if (MigrationRouter.isFullyTracked(mutation) && !mutation.isReadRepair())
             applyInternalTracked(mutation, null);
         else
             applyInternal(mutation, makeDurable, updateIndexes, isDroppable, false, null);
@@ -460,7 +460,7 @@ private Future<?> applyInternal(final Mutation mutation,
                                                boolean isDeferrable,
                                                Promise<?> future)
     {
-        Preconditions.checkState(!getMetadata().useMutationTracking() && mutation.id().isNone());
+        Preconditions.checkState((!getMetadata().useMutationTracking() || mutation.isReadRepair()) && mutation.id().isNone());
 
         if (TEST_FAIL_WRITES && getMetadata().name.equals(TEST_FAIL_WRITES_KS))
             throw new RuntimeException("Testing write failures");

src/java/org/apache/cassandra/db/Mutation.java

@@ -108,6 +108,21 @@ public class Mutation implements IMutation, Supplier<Mutation>, 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);
@@ -148,7 +163,9 @@ public MutationId id()
     @Override
     public Mutation withMutationId(MutationId mutationId)
     {
-        return new Mutation(mutationId, keyspaceName, key, modifications, approxCreatedAtNanos, cdcEnabled, potentialTxnConflicts);
+        Mutation m = new Mutation(mutationId, keyspaceName, key, modifications, approxCreatedAtNanos, cdcEnabled, potentialTxnConflicts);
+        m.isReadRepair = this.isReadRepair;
+        return m;
     }
 
     private static boolean cdcEnabled(Iterable<PartitionUpdate> modifications)
@@ -182,7 +199,9 @@ private static boolean cdcEnabled(Iterable<PartitionUpdate> modifications)
 
         Map<TableId, PartitionUpdate> updates = builder.build();
         checkState(!updates.isEmpty(), "Updates should not be empty");
-        return new Mutation(id, keyspaceName, key, builder.build(), approxCreatedAtNanos, potentialTxnConflicts);
+        Mutation result = new Mutation(id, keyspaceName, key, builder.build(), approxCreatedAtNanos, potentialTxnConflicts);
+        result.isReadRepair = this.isReadRepair;
+        return result;
     }
 
     public @Nullable Mutation without(TableId tableId)

src/java/org/apache/cassandra/db/ReadRepairVerbHandler.java

@@ -27,6 +27,7 @@ public class ReadRepairVerbHandler extends AbstractMutationVerbHandler<Mutation>
 
     public void applyMutation(Mutation mutation)
     {
+        mutation.setReadRepair(true);
         mutation.apply();
     }
 

src/java/org/apache/cassandra/db/lifecycle/Tracker.java

@@ -63,6 +63,7 @@
 import org.apache.cassandra.notifications.TablePreScrubNotification;
 import org.apache.cassandra.notifications.TruncationNotification;
 import org.apache.cassandra.replication.ImmutableCoordinatorLogOffsets;
+import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.utils.Pair;
 import org.apache.cassandra.utils.Throwables;
 import org.apache.cassandra.utils.TimeUUID;
@@ -272,7 +273,10 @@ public void updateInitialSSTableSize(Iterable<SSTableReader> sstables)
 
     public void addSSTables(Collection<SSTableReader> sstables)
     {
-        Preconditions.checkState(!cfstore.metadata().replicationType().isTracked());
+        // Tracked tables may legitimately use this path during migration from untracked to tracked,
+        // when incremental repair streams SSTables that were written before tracking was enabled.
+        Preconditions.checkState(!cfstore.metadata().replicationType().isTracked()
+                                 || ClusterMetadata.current().mutationTrackingMigrationState.isMigrating(cfstore.metadata().keyspace));
         addSSTablesInternal(sstables, false, true, true);
     }
 

src/java/org/apache/cassandra/db/streaming/CassandraStreamReceiver.java

@@ -53,6 +53,7 @@
 import org.apache.cassandra.service.accord.IAccordService;
 import org.apache.cassandra.service.accord.TimeOnlyRequestBookkeeping.LatencyRequestBookkeeping;
 import org.apache.cassandra.replication.MutationTrackingService;
+import org.apache.cassandra.service.replication.migration.KeyspaceMigrationInfo;
 import org.apache.cassandra.replication.PendingLocalTransfer;
 import org.apache.cassandra.streaming.IncomingStream;
 import org.apache.cassandra.streaming.StreamReceiver;
@@ -104,6 +105,19 @@ public CassandraStreamReceiver(ColumnFamilyStore cfs, StreamSession session, Lis
         this.requiresWritePath = requiresWritePath(cfs);
     }
 
+    /**
+     * Whether this stream should use the tracked transfer path (pending until activation).
+     * Returns false during mutation tracking migration for ranges that are still pending,
+     * since migration repair uses the untracked streaming path for those ranges.
+     */
+    private boolean useTrackedTransferPath()
+    {
+        if (!cfs.metadata().replicationType().isTracked() || !session.streamOperation().isTrackable())
+            return false;
+
+        return KeyspaceMigrationInfo.shouldUseTrackedTransfers(ClusterMetadata.current(), cfs.getKeyspaceName(), cfs.metadata().id, ranges);
+    }
+
     public static CassandraStreamReceiver fromReceiver(StreamReceiver receiver)
     {
         Preconditions.checkArgument(receiver instanceof CassandraStreamReceiver);
@@ -135,7 +149,7 @@ public synchronized void received(IncomingStream stream)
         sstables.addAll(finished);
         receivedEntireSSTable = file.isEntireSSTable();
 
-        if (cfs.metadata().replicationType().isTracked() && session.streamOperation().isTrackable())
+        if (useTrackedTransferPath())
         {
             PendingLocalTransfer transfer = new PendingLocalTransfer(cfs.metadata().id, session.planId(), sstables);
             MutationTrackingService.instance().received(transfer);
@@ -266,7 +280,7 @@ public void finished()
                 logger.debug("[Stream #{}] Received {} sstables from {} ({})", session.planId(), readers.size(), session.peer, readers);
 
                 // SSTables involved in a coordinated transfer become live when the transfer is activated
-                if (cfs.metadata().replicationType().isTracked() && session.streamOperation().isTrackable())
+                if (useTrackedTransferPath())
                     return;
 
                 if (session.streamOperation() == StreamOperation.BOOTSTRAP)

src/java/org/apache/cassandra/dht/NormalizedRanges.java

@@ -134,6 +134,19 @@ public boolean intersects(T token)
         return isIn;
     }
 
+    /**
+     * Check if any of these ranges intersect with the given range.
+     */
+    public boolean intersects(Range<T> range)
+    {
+        for (Range<T> r : this)
+        {
+            if (r.intersects(range))
+                return true;
+        }
+        return false;
+    }
+
     public NormalizedRanges<T> subtract(NormalizedRanges<T> b)
     {
         if (b.isEmpty())

src/java/org/apache/cassandra/io/sstable/format/SSTableWriter.java

@@ -62,6 +62,8 @@
 import org.apache.cassandra.replication.ImmutableCoordinatorLogOffsets;
 import org.apache.cassandra.replication.MutationTrackingService;
 import org.apache.cassandra.service.ActiveRepairService;
+import org.apache.cassandra.service.replication.migration.KeyspaceMigrationInfo;
+import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.utils.Clock;
 import org.apache.cassandra.utils.Throwables;
 import org.apache.cassandra.utils.TimeUUID;
@@ -337,16 +339,28 @@ public final void abort()
 
     protected Map<MetadataType, MetadataComponent> finalizeMetadata()
     {
-        // Reconciliation should not occur before activation for coordinated transfer streams for tracked keyspaces. 
+        // Reconciliation should not occur before activation for coordinated transfer streams for tracked keyspaces.
         boolean reconcile = txn.opType() != OperationType.STREAM;
 
+        // During migration, incremental repair handles repair status for ranges still pending migration.
+        // Only apply mutation tracking reconciliation for ranges NOT in the migration pending set.
+        // For SSTables whose range falls within pending migration ranges, IR sets pendingRepair/repairedAt.
         if (metadata().replicationType().isTracked() && repairedAt == ActiveRepairService.UNREPAIRED_SSTABLE && reconcile)
         {
-            Preconditions.checkState(Objects.equals(pendingRepair, ActiveRepairService.NO_PENDING_REPAIR));
-            if (MutationTrackingService.instance().isDurablyReconciled(coordinatorLogOffsets))
+            KeyspaceMigrationInfo migrationInfo = ClusterMetadata.current().mutationTrackingMigrationState.getKeyspaceInfo(metadata().keyspace);
+            boolean inMigrationPendingRange = migrationInfo != null
+                                              && migrationInfo.isRangeInPendingMigration(metadata().id,
+                                                                                         first.getToken(),
+                                                                                         last.getToken());
+
+            if (!inMigrationPendingRange)
             {
-                repairedAt = Clock.Global.currentTimeMillis();
-                logger.debug("Marking SSTable {} as reconciled with repairedAt {}", descriptor, repairedAt);
+                Preconditions.checkState(Objects.equals(pendingRepair, ActiveRepairService.NO_PENDING_REPAIR));
+                if (MutationTrackingService.instance().isDurablyReconciled(coordinatorLogOffsets))
+                {
+                    repairedAt = Clock.Global.currentTimeMillis();
+                    logger.debug("Marking SSTable {} as reconciled with repairedAt {}", descriptor, repairedAt);
+                }
             }
         }
 

src/java/org/apache/cassandra/net/Verb.java

@@ -64,6 +64,8 @@
 import org.apache.cassandra.repair.messages.FinalizeCommit;
 import org.apache.cassandra.repair.messages.FinalizePromise;
 import org.apache.cassandra.repair.messages.FinalizePropose;
+import org.apache.cassandra.repair.messages.MutationTrackingSyncRequest;
+import org.apache.cassandra.repair.messages.MutationTrackingSyncResponse;
 import org.apache.cassandra.repair.messages.PrepareConsistentRequest;
 import org.apache.cassandra.repair.messages.PrepareConsistentResponse;
 import org.apache.cassandra.repair.messages.PrepareMessage;
@@ -287,8 +289,10 @@ public enum Verb
     FINALIZE_PROMISE_MSG   (111, P1, repairWithBackoffTimeout,      ANTI_ENTROPY,      () -> FinalizePromise.serializer,           () -> RepairMessageVerbHandler.instance(),   REPAIR_RSP          ),
     FINALIZE_COMMIT_MSG    (112, P1, repairWithBackoffTimeout,      ANTI_ENTROPY,      () -> FinalizeCommit.serializer,            () -> RepairMessageVerbHandler.instance(),   REPAIR_RSP          ),
     FAILED_SESSION_MSG     (113, P1, repairWithBackoffTimeout,      ANTI_ENTROPY,      () -> FailSession.serializer,               () -> RepairMessageVerbHandler.instance(),   REPAIR_RSP          ),
-    STATUS_RSP             (115, P1, repairTimeout,   ANTI_ENTROPY,      () -> StatusResponse.serializer,            () -> RepairMessageVerbHandler.instance(),   REPAIR_RSP          ),
-    STATUS_REQ             (114, P1, repairTimeout,   ANTI_ENTROPY,      () -> StatusRequest.serializer,             () -> RepairMessageVerbHandler.instance(),   REPAIR_RSP          ),
+    STATUS_RSP             (115, P1, repairTimeout,                 ANTI_ENTROPY,      () -> StatusResponse.serializer,            () -> RepairMessageVerbHandler.instance(),   REPAIR_RSP          ),
+    STATUS_REQ             (114, P1, repairTimeout,                 ANTI_ENTROPY,      () -> StatusRequest.serializer,             () -> RepairMessageVerbHandler.instance(),   REPAIR_RSP          ),
+    MT_SYNC_RSP            (117, P1, repairWithBackoffTimeout,      REQUEST_RESPONSE,  () -> MutationTrackingSyncResponse.serializer, RESPONSE_HANDLER                                              ),
+    MT_SYNC_REQ            (116, P1, repairWithBackoffTimeout,      ANTI_ENTROPY,      () -> MutationTrackingSyncRequest.serializer, () -> RepairMessageVerbHandler.instance(), MT_SYNC_RSP         ),
 
     REPLICATION_DONE_RSP   (82,  P0, rpcTimeout,      MISC,              () -> NoPayload.serializer,                 RESPONSE_HANDLER                             ),
     REPLICATION_DONE_REQ   (22,  P0, rpcTimeout,      MISC,              () -> NoPayload.serializer,                 () -> ReplicationDoneVerbHandler.instance, REPLICATION_DONE_RSP),