Paxos support for live migration to mutation tracking

During mutation tracking migration (tracked <-> untracked), the per-range
migration state must be consulted for every routing decision. Before this
change, all Paxos V1 and V2 paths used the static keyspace-level
replicationType().isTracked() check, which does not reflect per-range
migration state and produces incorrect routing decisions during migration.

Coordinator-side routing: Replace all static isTracked() checks with
MigrationRouter calls across StorageProxy (commitPaxos, sendCommit,
isTrackedKeyspaceRequiringPaxosCommitForwarding, checkAndForwardCasIfNeeded,
checkAndForwardConsensusReadIfNeeded), PaxosCommit (constructor,
isTrackedKeyspaceRequiringForwarding), PaxosCommitAndPrepare,
PaxosPrepare (start + isTracked field), PaxosPrepareRefresh, and
PaxosState truncation acknowledgment.

Handler-side validation: Add migration state validation to four Paxos
replica handlers that receive messages carrying mutations or tracked reads:
PaxosCommit.RequestHandler (direct V1/V2 commits), PaxosPrepare.RequestHandler
(V2 prepare with tracked read), PaxosCommitAndPrepare.RequestHandler
(combined commit+prepare), and PaxosPrepareRefresh.RequestHandler (refresh
commits). Each uses the conditional-fetch pattern from
AbstractMutationVerbHandler.checkReplicationMigration: compare the
coordinator routing decision against the handler MigrationRouter result,
fetch only on mismatch when the coordinator epoch is ahead (handler is
behind and needs to catch up), throw CoordinatorBehindException when the
coordinator epoch is behind.

Coordinator-side commit retry: Add commit-level COORDINATOR_BEHIND retry
in Paxos.cas() (V2) and commitPaxos() (V1). When replicas reject a commit
due to migration state mismatch, ResponseVerbHandler.maybeFetchLogs()
catches up the coordinator synchronously before delivering the failure.
The retry re-creates the commit with fresh MigrationRouter routing. This
retries only the commit phase, not the entire prepare+propose protocol.

Stale mutation ID reconciliation: Commits saved in system.paxos may have
a mutation ID from when the keyspace was tracked. When replayed after
migration to untracked (via PaxosPrepareRefresh, PaxosCommitAndPrepare,
sendCommit, or commitPaxos), the stale ID must be stripped to avoid
Keyspace.apply() rejecting the mutation. Uses Commit.withMutationId() to
reconcile in all four replay paths.

Forward handlers: Forwarding is harmless -- the receiving replica
re-executes the full CAS/read with its own fresh routing decisions, so no
migration validation is needed at the forward boundary itself. Removed the
"reject if keyspace not tracked" guards from CasForwardHandler and
ConsensusReadForwardHandler (forwarding is now valid in either direction).
Replaced unconditional fetchLogFromPeerOrCMS with the conditional-fetch
pattern in Paxos2CommitForwardHandler, PaxosCommitForwardHandler,
PrepareRefreshForwardHandler, and PaxosCommitAndPrepare.RequestHandler
(unconditional fetch added unnecessary latency on the no-mismatch case).

PaxosCommit failure tracking: Added super.onFailure() call to
PaxosCommit.onFailure() so FailureRecordingCallback.failureResponses is
populated, enabling failureReasonsAsMap() to return actual failure reasons.
This was required for the V2 commit retry to detect COORDINATOR_BEHIND
in MaybeFailure.failures.

PaxosCommit hint suppression: Tracked mutations must not be written as
hints because hint replay routes through Keyspace.applyInternalTracked()
based on the mutation ID presence, which fails after migration to untracked.
Guard submitHint with !isTracked() -- tracked mutations use
MutationTrackingService for retries, not the hint system.

MigrationRouter null safety: Replace getKeyspaceMetadata() (throws
NoSuchElementException on missing keyspace) with
maybeGetKeyspaceMetadata().orElse(null) at four call sites so the existing
null guards actually protect against concurrent keyspace drops.

Author: aweisberg

src/java/org/apache/cassandra/exceptions/CoordinatorBehindException.java

@@ -25,7 +25,7 @@ public CoordinatorBehindException(String msg)
         super(msg);
     }
 
-    public CoordinatorBehindException(String msg, UnknownTableException cause)
+    public CoordinatorBehindException(String msg, Throwable cause)
     {
         super(msg, cause);
     }

src/java/org/apache/cassandra/exceptions/WriteTimeoutException.java

@@ -42,6 +42,12 @@ public WriteTimeoutException(WriteType writeType, ConsistencyLevel consistency,
         this.writeType = writeType;
     }
 
+    public WriteTimeoutException(WriteType writeType, ConsistencyLevel consistency, int received, int blockFor, Throwable cause)
+    {
+        super(ExceptionCode.WRITE_TIMEOUT, consistency, received, blockFor, cause);
+        this.writeType = writeType;
+    }
+
     @Override
     protected void serializeSpecificFields(DataOutputPlus out, int version) throws IOException
     {

src/java/org/apache/cassandra/hints/HintsService.java

@@ -73,6 +73,14 @@
  */
 public final class HintsService implements HintsServiceMBean
 {
+    private static long REJECT_HINTS_BEFORE_NANOS;
+
+    @VisibleForTesting
+    public static void setRejectHintsBeforeNanos(long nanos)
+    {
+        REJECT_HINTS_BEFORE_NANOS = nanos;
+    }
+
     // Dummy address to use for storing metrics for hints that will be retried on a different transaction system
     // and aren't being sent to a specific node
     public static final InetAddressAndPort RETRY_ON_DIFFERENT_SYSTEM_ADDRESS;
@@ -189,6 +197,9 @@ public void write(Collection<UUID> hostIds, Hint hint)
         if (isShutDown)
             throw new IllegalStateException("HintsService is shut down and can't accept new hints");
 
+        if (hint.mutation.getApproxCreatedAtNanos() < REJECT_HINTS_BEFORE_NANOS)
+            return;
+
         // we have to make sure that the HintsStore instances get properly initialized - otherwise dispatch will not trigger
         catalog.maybeLoadStores(hostIds);
 

src/java/org/apache/cassandra/service/StorageProxy.java

@@ -40,7 +40,7 @@
 /**
  * Handler for forwarded CAS (Compare-And-Set) operations.
  * Executes the CAS operation on behalf of the original coordinator,
- * ensuring that MutationId generation happens on a replica coordinator for tracked keyspaces.
+ * ensuring that MutationId generation happens on a replica coordinator.
  *
  * TODO (expected): more comprehensive testing
  */
@@ -60,7 +60,6 @@ public void doVerb(Message<CasForwardRequest> message)
         ClientWarn.instance.captureWarnings();
         try
         {
-            // Validate keyspace exists and is tracked
             KeyspaceMetadata ksMetadata = Schema.instance.getKeyspaceMetadata(request.keyspaceName);
             if (ksMetadata == null)
             {
@@ -69,13 +68,6 @@ public void doVerb(Message<CasForwardRequest> message)
                 return;
             }
 
-            if (!ksMetadata.params.replicationType.isTracked())
-            {
-                MessagingService.instance().respondWithFailure(RequestFailureReason.INCOMPATIBLE_SCHEMA, message);
-                logger.error("Asked to perform forwarded CAS operation, but keyspace {} is not tracked", request.keyspaceName);
-                return;
-            }
-
             // Execute the forwarded CAS operation
             logger.debug("Executing CAS operation for table {}.{} with key {}",
                          request.keyspaceName, request.cfName, request.key);

src/java/org/apache/cassandra/service/paxos/CasForwardHandler.java

@@ -180,6 +180,12 @@ public AcceptedWithTTL(Ballot ballot, PartitionUpdate update, long localDeletion
             this.localDeletionTime = localDeletionTime;
         }
 
+        public AcceptedWithTTL(Ballot ballot, Mutation mutation, long localDeletionTime)
+        {
+            super(ballot, mutation);
+            this.localDeletionTime = localDeletionTime;
+        }
+
         boolean isExpired(long nowInSec)
         {
             return nowInSec >= localDeletionTime;
@@ -194,7 +200,7 @@ Accepted lastDeleted(Accepted b)
         @Override
         public AcceptedWithTTL withMutationId(MutationId mutationId)
         {
-            return new AcceptedWithTTL(ballot, makeMutation(mutationId).getOnlyUpdate(), localDeletionTime);
+            return new AcceptedWithTTL(ballot, makeMutation(mutationId), localDeletionTime);
         }
     }
 
@@ -385,6 +391,12 @@ public static Commit newProposal(Ballot ballot, PartitionUpdate update)
         return new Commit(ballot, update);
     }
 
+    public static Commit newProposal(Ballot ballot, Mutation mutation)
+    {
+        PartitionUpdate update = withTimestamp(mutation.getOnlyUpdate(), ballot.unixMicros());
+        return new Commit(ballot, new Mutation(mutation.id(), update, mutation.potentialTxnConflicts()));
+    }
+
     public boolean isAfter(Commit other)
     {
         return other == null || ballot.uuidTimestamp() > other.ballot.uuidTimestamp();
@@ -651,7 +663,7 @@ public T deserialize(DataInputPlus in, int version) throws IOException
             if (version >= MessagingService.VERSION_61)
             {
                 // New format: deserialize Mutation
-                Mutation mutation = org.apache.cassandra.db.Mutation.serializer.deserialize(in, version);
+                Mutation mutation = Mutation.serializer.deserialize(in, version);
                 return mutationConstructor.apply(ballot, mutation);
             }
             else

src/java/org/apache/cassandra/service/paxos/Commit.java

@@ -40,8 +40,7 @@
 
 /**
  * Handler for forwarded consensus read operations.
- * Executes the consensus read operation on behalf of the original coordinator,
- * ensuring proper coordination for tracked keyspaces on a replica coordinator.
+ * Executes the consensus read operation on behalf of the original coordinator.
  *
  * TODO (expected): more comprehensive testing
  */
@@ -55,13 +54,13 @@ public void doVerb(Message<ConsensusReadForwardRequest> message)
     {
         ConsensusReadForwardRequest request = message.payload;
         SinglePartitionReadCommand command = request.command;
+
         Tracing.trace("Executing forwarded consensus read operation for {}", command.partitionKey());
 
         // Start capturing client warnings for the forwarded operation
         ClientWarn.instance.captureWarnings();
         try
         {
-            // Validate keyspace exists and is tracked
             String keyspaceName = command.metadata().keyspace;
             KeyspaceMetadata ksMetadata = Schema.instance.getKeyspaceMetadata(keyspaceName);
             if (ksMetadata == null)
@@ -71,21 +70,14 @@ public void doVerb(Message<ConsensusReadForwardRequest> message)
                 return;
             }
 
-            if (!ksMetadata.params.replicationType.isTracked())
-            {
-                MessagingService.instance().respondWithFailure(RequestFailureReason.INCOMPATIBLE_SCHEMA, message);
-                logger.error("Asked to perform forwarded consensus read operation, but keyspace {} is not tracked", keyspaceName);
-                return;
-            }
-
             // Create a Group from the single command for reading
             SinglePartitionReadCommand.Group group = SinglePartitionReadCommand.Group.one(command);
 
             // Execute the read using StorageProxy.read() which will:
             // 1. Check forwarding (returns null since we're on a replica)
             // 2. Execute the consensus read with the appropriate protocol
             logger.debug("Executing consensus read operation for table {}.{} with key {}",
-                         keyspaceName, command.metadata().name, command.partitionKey());
+                         command.metadata().keyspace, command.metadata().name, command.partitionKey());
 
             Dispatcher.RequestTime requestTime = Dispatcher.RequestTime.forImmediateExecution();
             PartitionIterator result = StorageProxy.readWithConsensusForwarded(group, request.consistencyLevel, requestTime);

src/java/org/apache/cassandra/service/paxos/ConsensusReadForwardHandler.java

@@ -20,6 +20,7 @@
 
 import java.io.IOException;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
@@ -94,6 +95,7 @@
 import org.apache.cassandra.service.ClientState;
 import org.apache.cassandra.service.FailureRecordingCallback.AsMap;
 import org.apache.cassandra.service.consensus.migration.ConsensusRequestRouter;
+import org.apache.cassandra.service.paxos.Commit.Agreed;
 import org.apache.cassandra.service.paxos.Commit.Proposal;
 import org.apache.cassandra.service.paxos.PaxosPrepare.FoundIncompleteAccepted;
 import org.apache.cassandra.service.paxos.PaxosPrepare.FoundIncompleteCommitted;
@@ -740,6 +742,8 @@ public static ConsensusAttemptResult cas(DecoratedKey partitionKey,
         try (PaxosOperationLock lock = PaxosState.lock(partitionKey, metadata, proposeDeadline, consistencyForConsensus, true))
         {
             Paxos.Async<PaxosCommit.Status> commit = null;
+            Agreed agreed = null;
+            Participants commitParticipants = null;
             done: while (true)
             {
                 // read the current values and check they validate the conditions
@@ -839,7 +843,11 @@ else if (begin.isPromised)
                         //   1) reached a majority, in which case it was agreed, had no effect and we can do nothing; or
                         //   2) did not reach a majority, was not agreed, and was not user visible as a result so we can ignore it
                         if (!proposal.isEmpty())
-                            commit = commit(proposal.agreed(), participants, consistencyForConsensus, consistencyForCommit, true);
+                        {
+                            agreed = proposal.agreed();
+                            commitParticipants = participants;
+                            commit = commit(agreed, participants, consistencyForConsensus, consistencyForCommit, true);
+                        }
 
                         break done;
                     }
@@ -872,8 +880,26 @@ else if (begin.isPromised)
             if (commit != null)
             {
                 PaxosCommit.Status result = commit.awaitUntil(commitDeadline);
-                if (!result.isSuccess())
-                    throw result.maybeFailure().markAndThrowAsTimeoutOrFailure(true, consistencyForCommit, failedAttemptsDueToContention);
+                while (!result.isSuccess())
+                {
+                    Paxos.MaybeFailure failure = result.maybeFailure();
+                    long coordinatorBehindCount = Collections.frequency(failure.failures.values(),
+                                                                        RequestFailureReason.COORDINATOR_BEHIND);
+                    if (coordinatorBehindCount == 0
+                        || failure.successes + coordinatorBehindCount < failure.required
+                        || nanoTime() >= commitDeadline)
+                    {
+                        throw failure.markAndThrowAsTimeoutOrFailure(true, consistencyForCommit, failedAttemptsDueToContention);
+                    }
+
+                    casWriteMetrics.retryCoordinatorBehind.mark();
+                    Tracing.trace("Retrying V2 Paxos commit after COORDINATOR_BEHIND for {}.{} partition {}, {} behind replicas out of {} required",
+                                  metadata.keyspace, metadata.name, partitionKey, coordinatorBehindCount, failure.required);
+                    logger.warn("Retrying V2 Paxos commit after COORDINATOR_BEHIND for {}.{} partition {}, {} behind replicas out of {} required",
+                                metadata.keyspace, metadata.name, partitionKey, coordinatorBehindCount, failure.required);
+                    commit = commit(agreed, commitParticipants, consistencyForConsensus, consistencyForCommit, true);
+                    result = commit.awaitUntil(commitDeadline);
+                }
             }
             Tracing.trace("CAS successful");
             return casResult((RowIterator)null);
@@ -1103,7 +1129,7 @@ private static BeginResult begin(long deadline,
                     // is equal to the latest commit (even if the ballots aren't) we're done and can abort earlier,
                     // and in fact it's possible for a CAS to sometimes determine if side effects occurred by reading
                     // the underlying data and not witnessing the timestamp of its ballot (or any newer for the relevant data).
-                    Proposal repropose = new Proposal(inProgress.ballot, inProgress.accepted.update);
+                    Proposal repropose = new Proposal(inProgress.ballot, inProgress.accepted.mutation);
                     PaxosPropose.Status proposeResult = propose(repropose, inProgress.participants, false).awaitUntil(deadline);
                     switch (proposeResult.outcome)
                     {

src/java/org/apache/cassandra/service/paxos/Paxos.java

@@ -29,8 +29,8 @@
 import org.apache.cassandra.net.Message;
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.net.NoPayload;
-import org.apache.cassandra.schema.KeyspaceMetadata;
-import org.apache.cassandra.schema.Schema;
+import org.apache.cassandra.service.replication.migration.MigrationRouter;
+import org.apache.cassandra.tcm.ClusterMetadata;
 import org.apache.cassandra.tcm.ClusterMetadataService;
 import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.utils.concurrent.ConditionAsConsumer;
@@ -39,13 +39,11 @@
 
 /**
  * Handler for forwarded Paxos V2 commit requests.
- * Executes the commit operation on behalf of the original coordinator,
- * ensuring that MutationId generation happens on a replica coordinator.
- *
- * The PaxosCommit constructor handles mutation ID generation, so this handler
- * simply delegates to PaxosCommit.commit() with the original commit.
+ * Delegates to PaxosCommit.commit() which handles mutation ID generation
+ * in its constructor.
  *
  * TODO (expected): more comprehensive testing
+ * TODO: should loop on CoordinatorBehindException rather than propagating failure to the forwarding coordinator
  */
 public class Paxos2CommitForwardHandler implements IVerbHandler<Paxos2CommitForwardRequest>
 {
@@ -55,27 +53,38 @@ public class Paxos2CommitForwardHandler implements IVerbHandler<Paxos2CommitForw
     @Override
     public void doVerb(Message<Paxos2CommitForwardRequest> message)
     {
-        // Ensure we have up-to-date cluster metadata before executing the forwarded commit
-        ClusterMetadataService.instance().fetchLogFromPeerOrCMS(message.from(), message.header.epoch);
         Paxos2CommitForwardRequest request = message.payload;
+        Commit.Agreed commit = request.commit;
 
-        Tracing.trace("Executing forwarded Paxos V2 commit for {}", request.commit.partitionKey());
+        Tracing.trace("Executing forwarded Paxos V2 commit for {}.{} partition {}",
+                      commit.metadata().keyspace, commit.metadata().name, commit.partitionKey());
 
         try
         {
-            String ksName = request.commit.metadata().keyspace;
-            KeyspaceMetadata ksMetadata = Schema.instance.getKeyspaceMetadata(ksName);
-            if (ksMetadata == null)
+            String ksName = commit.metadata().keyspace;
+            ClusterMetadata metadata = ClusterMetadata.current();
+            boolean shouldBeTracked = MigrationRouter.shouldUseTrackedForWrites(metadata,
+                                                                                ksName,
+                                                                                commit.metadata().id,
+                                                                                commit.partitionKey().getToken());
+
+            if (!shouldBeTracked && message.epoch().isAfter(metadata.epoch))
             {
-                MessagingService.instance().respondWithFailure(RequestFailureReason.INCOMPATIBLE_SCHEMA, message);
-                logger.error("Failed to forward paxos commit for non-existent keyspace {}", ksName);
-                return;
+                metadata = ClusterMetadataService.instance().fetchLogFromPeerOrCMS(metadata, message.from(), message.epoch());
+                // shouldBeTracked isn't used after this, but is kept up to date just in case
+                shouldBeTracked = MigrationRouter.shouldUseTrackedForWrites(metadata,
+                                                                            ksName,
+                                                                            commit.metadata().id,
+                                                                            commit.partitionKey().getToken());
             }
 
-            if (!ksMetadata.params.replicationType.isTracked())
+            if (metadata.schema.getKeyspaces().getNullable(ksName) == null)
             {
                 MessagingService.instance().respondWithFailure(RequestFailureReason.INCOMPATIBLE_SCHEMA, message);
-                logger.error("Asked to perform forwarded paxos commit, but keyspace {} is not tracked", ksName);
+                logger.error("Failed to forward paxos commit for non-existent keyspace {}.{} partition {}",
+                             ksName, commit.metadata().name, commit.partitionKey());
+                Tracing.trace("Failed to forward paxos commit for non-existent keyspace {}.{} partition {}",
+                              ksName, commit.metadata().name, commit.partitionKey());
                 return;
             }
 
@@ -112,15 +121,29 @@ public void doVerb(Message<Paxos2CommitForwardRequest> message)
                 }
                 else
                 {
-                    MessagingService.instance().respondWithFailure(RequestFailureReason.UNKNOWN, message);
+                    RequestFailureReason reason = RequestFailureReason.UNKNOWN;
+                    if (status != null && status.maybeFailure() != null)
+                    {
+                        for (RequestFailureReason r : status.maybeFailure().failures.values())
+                        {
+                            if (r == RequestFailureReason.COORDINATOR_BEHIND)
+                            {
+                                reason = RequestFailureReason.COORDINATOR_BEHIND;
+                                break;
+                            }
+                        }
+                    }
+                    MessagingService.instance().respondWithFailure(reason, message);
                     logger.error("Forwarded Paxos V2 commit failed with status: {}", status);
+                    Tracing.trace("Forwarded Paxos V2 commit failed with status: {}", status);
                 }
             }
             catch (InterruptedException e)
             {
                 Thread.currentThread().interrupt();
                 MessagingService.instance().respondWithFailure(RequestFailure.forException(e), message);
                 logger.error("Forwarded Paxos V2 commit interrupted", e);
+                Tracing.trace("Forwarded Paxos V2 commit interrupted");
             }
         }
         catch (Exception e)