Merge commit '080280dc0177da6176dd4ba970e5a35aa7e2a729' into trunk

CHANGES.txt

@@ -15,6 +15,7 @@
 Merged from 3.11:
  * SASI's `max_compaction_flush_memory_in_mb` settings over 100GB revert to default of 1GB (CASSANDRA-16071)
 Merged from 3.0:
+ * Fix serial read/non-applying CAS linearizability (CASSANDRA-12126)
  * Avoid potential NPE in JVMStabilityInspector (CASSANDRA-16294)
  * Improved check of num_tokens against the length of initial_token (CASSANDRA-14477)
  * Fix a race condition on ColumnFamilyStore and TableMetrics (CASSANDRA-16228)

NEWS.txt

@@ -283,6 +283,14 @@ Materialized Views
 
 Upgrading
 ---------
+    - This release fix a correctness issue with SERIAL reads, and LWT writes that do not apply.
+      Unfortunately, this fix has a performance impact on read performance at the SERIAL or
+      LOCAL_SERIAL consistency levels. For heavy users of such SERIAL reads, the performance
+      impact may be noticeable and may also result in an increased of timeouts. For that
+      reason, a opt-in system property has been added to disable the fix:
+        -Dcassandra.unsafe.disable-serial-reads-linearizability=true
+      Use this flag at your own risk as it revert SERIAL reads to the incorrect behavior of
+      previous versions. See CASSANDRA-12126 for details. 
     - SASI's `max_compaction_flush_memory_in_mb` setting was previously getting interpreted in bytes. From 3.11.8
       it is correctly interpreted in megabytes, but prior to 3.11.10 previous configurations of this setting will
       lead to nodes OOM during compaction. From 3.11.10 previous configurations will be detected as incorrect,

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

@@ -18,7 +18,6 @@
 package org.apache.cassandra.service;
 
 import java.nio.ByteBuffer;
-import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
@@ -41,6 +40,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.Supplier;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
@@ -202,6 +202,10 @@ public class StorageProxy implements StorageProxyMBean
      */
     private static final int MAX_CONCURRENT_RANGE_REQUESTS = Math.max(1, Integer.getInteger("cassandra.max_concurrent_range_requests", FBUtilities.getAvailableProcessors() * 10));
 
+    private static final String DISABLE_SERIAL_READ_LINEARIZABILITY_KEY = "cassandra.unsafe.disable-serial-reads-linearizability";
+    private static final boolean disableSerialReadLinearizability =
+        Boolean.parseBoolean(System.getProperty(DISABLE_SERIAL_READ_LINEARIZABILITY_KEY, "false"));
+
     private StorageProxy()
     {
     }
@@ -245,6 +249,16 @@ public class StorageProxy implements StorageProxyMBean
         }
 
         ReadRepairMetrics.init();
+
+        if (disableSerialReadLinearizability)
+        {
+            logger.warn("This node was started with -D{}. SERIAL (and LOCAL_SERIAL) reads coordinated by this node " +
+                        "will not offer linearizability (see CASSANDRA-12126 for details on what this mean) with " +
+                        "respect to other SERIAL operations. Please note that, with this flag, SERIAL reads will be " +
+                        "slower than QUORUM reads, yet offer no more guarantee. This flag should only be used in " +
+                        "the restricted case of upgrading from a pre-CASSANDRA-12126 version, and only if you " +
+                        "understand the tradeoff.", DISABLE_SERIAL_READ_LINEARIZABILITY_KEY);
+        }
     }
 
     /**
@@ -300,23 +314,12 @@ public class StorageProxy implements StorageProxyMBean
     throws UnavailableException, IsBootstrappingException, RequestFailureException, RequestTimeoutException, InvalidRequestException, CasWriteUnknownResultException
     {
         final long startTimeForMetrics = System.nanoTime();
-        TableMetadata metadata = Schema.instance.getTableMetadata(keyspaceName, cfName);
-        int contentions = 0;
         try
         {
-            consistencyForPaxos.validateForCas();
+            TableMetadata metadata = Schema.instance.validateTable(keyspaceName, cfName);
-            consistencyForCommit.validateForCasCommit(keyspaceName);
 
-            long timeoutNanos = DatabaseDescriptor.getCasContentionTimeout(NANOSECONDS);
+            Supplier<Pair<PartitionUpdate, RowIterator>> updateProposer = () ->
-            while (System.nanoTime() - queryStartNanoTime < timeoutNanos)
             {
-                // for simplicity, we'll do a single liveness check at the start of each attempt
-                ReplicaPlan.ForPaxosWrite replicaPlan = ReplicaPlans.forPaxos(Keyspace.open(keyspaceName), key, consistencyForPaxos);
-
-                final PaxosBallotAndContention pair = beginAndRepairPaxos(queryStartNanoTime, key, metadata, replicaPlan, consistencyForPaxos, consistencyForCommit, true, state);
-                final UUID ballot = pair.ballot;
-                contentions += pair.contentions;
-
                 // read the current values and check they validate the conditions
                 Tracing.trace("Reading existing values for CAS precondition");
                 SinglePartitionReadCommand readCommand = (SinglePartitionReadCommand) request.readCommand(nowInSeconds);
@@ -332,11 +335,10 @@ public class StorageProxy implements StorageProxyMBean
                 {
                     Tracing.trace("CAS precondition does not match current values {}", current);
                     casWriteMetrics.conditionNotMet.inc();
-                    return current.rowIterator();
+                    return Pair.create(PartitionUpdate.emptyUpdate(metadata, key), current.rowIterator());
                 }
 
-                // finish the paxos round w/ the desired updates
+                // Create the desired updates
-                // TODO turn null updates into delete?
                 PartitionUpdate updates = request.makeUpdates(current);
 
                 long size = updates.dataSize();
@@ -352,34 +354,30 @@ public class StorageProxy implements StorageProxyMBean
                 // InvalidRequestException) any which aren't.
                 updates = TriggerExecutor.instance.execute(updates);
 
+                return Pair.create(updates, null);
+            };
 
-                Commit proposal = Commit.newProposal(ballot, updates);
+            return doPaxos(metadata,
-                Tracing.trace("CAS precondition is met; proposing client-requested updates for {}", ballot);
+                           key,
-                if (proposePaxos(proposal, replicaPlan, true, queryStartNanoTime))
+                           consistencyForPaxos,
-                {
+                           consistencyForCommit,
-                    commitPaxos(proposal, consistencyForCommit, true, queryStartNanoTime);
+                           consistencyForCommit,
-                    Tracing.trace("CAS successful");
+                           state,
-                    return null;
+                           queryStartNanoTime,
-                }
+                           casWriteMetrics,
+                           updateProposer);
 
-                Tracing.trace("Paxos proposal not accepted (pre-empted by a higher ballot)");
-                contentions++;
-                Uninterruptibles.sleepUninterruptibly(ThreadLocalRandom.current().nextInt(100), MILLISECONDS);
-                // continue to retry
-            }
-
-            throw new WriteTimeoutException(WriteType.CAS, consistencyForPaxos, 0, consistencyForPaxos.blockFor(Keyspace.open(keyspaceName)));
         }
         catch (CasWriteUnknownResultException e)
         {
             casWriteMetrics.unknownResult.mark();
             throw e;
         }
-        catch (WriteTimeoutException wte)
+        catch (CasWriteTimeoutException wte)
         {
             casWriteMetrics.timeouts.mark();
             writeMetricsMap.get(consistencyForPaxos).timeouts.mark();
-            throw new CasWriteTimeoutException(wte.writeType, wte.consistency, wte.received, wte.blockFor, contentions);
+            throw new CasWriteTimeoutException(wte.writeType, wte.consistency, wte.received, wte.blockFor, wte.contentions);
         }
         catch (ReadTimeoutException e)
         {
@@ -387,7 +385,7 @@ public class StorageProxy implements StorageProxyMBean
             writeMetricsMap.get(consistencyForPaxos).timeouts.mark();
             throw e;
         }
-        catch (WriteFailureException|ReadFailureException e)
+        catch (WriteFailureException | ReadFailureException e)
         {
             casWriteMetrics.failures.mark();
             writeMetricsMap.get(consistencyForPaxos).failures.mark();
@@ -401,18 +399,137 @@ public class StorageProxy implements StorageProxyMBean
         }
         finally
         {
-            recordCasContention(contentions);
-            Keyspace.open(keyspaceName).getColumnFamilyStore(cfName).metric.topCasPartitionContention.addSample(key.getKey(), contentions);
             final long latency = System.nanoTime() - startTimeForMetrics;
             casWriteMetrics.addNano(latency);
             writeMetricsMap.get(consistencyForPaxos).addNano(latency);
         }
     }
 
-    private static void recordCasContention(int contentions)
+    private static void recordCasContention(TableMetadata table,
+                                            DecoratedKey key,
+                                            CASClientRequestMetrics casMetrics,
+                                            int contentions)
     {
-        if(contentions > 0)
+        if (contentions == 0)
-            casWriteMetrics.contention.update(contentions);
+            return;
+
+        casMetrics.contention.update(contentions);
+        Keyspace.open(table.keyspace)
+                .getColumnFamilyStore(table.name)
+                .metric
+                .topCasPartitionContention
+                .addSample(key.getKey(), contentions);
+    }
+
+    /**
+     * Performs the Paxos rounds for a given proposal, retrying when preempted until the timeout.
+     *
+     * <p>The main 'configurable' of this method is the {@code createUpdateProposal} method: it is called by the method
+     * once a ballot has been successfully 'prepared' to generate the update to 'propose' (and commit if the proposal is
+     * successful). That method also generates the result that the whole method will return. Note that due to retrying,
+     * this method may be called multiple times and does not have to return the same results.
+     *
+     * @param metadata the table to update with Paxos.
+     * @param key the partition updated.
+     * @param consistencyForPaxos the serial consistency of the operation (either {@link ConsistencyLevel#SERIAL} or
+     *     {@link ConsistencyLevel#LOCAL_SERIAL}).
+     * @param consistencyForReplayCommits the consistency for the commit phase of "replayed" in-progress operations.
+     * @param consistencyForCommit the consistency for the commit phase of _this_ operation update.
+     * @param state the client state.
+     * @param queryStartNanoTime the nano time for the start of the query this is part of. This is the base time for
+     *     timeouts.
+     * @param casMetrics the metrics to update for this operation.
+     * @param createUpdateProposal method called after a successful 'prepare' phase to obtain 1) the actual update of
+     *     this operation and 2) the result that the whole method should return. This can return {@code null} in the
+     *     special where, after having "prepared" (and thus potentially replayed in-progress upgdates), we don't want
+     *     to propose anything (the whole method then return {@code null}).
+     * @return the second element of the pair returned by {@code createUpdateProposal} (for the last call of that method
+     *     if that method is called multiple times due to retries).
+     */
+    private static RowIterator doPaxos(TableMetadata metadata,
+                                       DecoratedKey key,
+                                       ConsistencyLevel consistencyForPaxos,
+                                       ConsistencyLevel consistencyForReplayCommits,
+                                       ConsistencyLevel consistencyForCommit,
+                                       ClientState state,
+                                       long queryStartNanoTime,
+                                       CASClientRequestMetrics casMetrics,
+                                       Supplier<Pair<PartitionUpdate, RowIterator>> createUpdateProposal)
+    throws UnavailableException, IsBootstrappingException, RequestFailureException, RequestTimeoutException, InvalidRequestException
+    {
+        int contentions = 0;
+        Keyspace keyspace = Keyspace.open(metadata.keyspace);
+        try
+        {
+            consistencyForPaxos.validateForCas();
+            consistencyForReplayCommits.validateForCasCommit(metadata.keyspace);
+            consistencyForCommit.validateForCasCommit(metadata.keyspace);
+
+            long timeoutNanos = DatabaseDescriptor.getCasContentionTimeout(NANOSECONDS);
+            while (System.nanoTime() - queryStartNanoTime < timeoutNanos)
+            {
+                // for simplicity, we'll do a single liveness check at the start of each attempt
+                ReplicaPlan.ForPaxosWrite replicaPlan = ReplicaPlans.forPaxos(keyspace, key, consistencyForPaxos);
+                PaxosBallotAndContention pair = beginAndRepairPaxos(queryStartNanoTime,
+                                                                    key,
+                                                                    metadata,
+                                                                    replicaPlan,
+                                                                    consistencyForPaxos,
+                                                                    consistencyForReplayCommits,
+                                                                    casMetrics,
+                                                                    state);
+
+                final UUID ballot = pair.ballot;
+                contentions += pair.contentions;
+
+                Pair<PartitionUpdate, RowIterator> proposalPair = createUpdateProposal.get();
+                // See method javadoc: null here is code for "stop here and return null".
+                if (proposalPair == null)
+                    return null;
+
+                Commit proposal = Commit.newProposal(ballot, proposalPair.left);
+                Tracing.trace("CAS precondition is met; proposing client-requested updates for {}", ballot);
+                if (proposePaxos(proposal, replicaPlan, true, queryStartNanoTime))
+                {
+                    // We skip committing accepted updates when they are empty. This is an optimization which works
+                    // because we also skip replaying those same empty update in beginAndRepairPaxos (see the longer
+                    // comment there). As empty update are somewhat common (serial reads and non-applying CAS propose
+                    // them), this is worth bothering.
+                    if (!proposal.update.isEmpty())
+                        commitPaxos(proposal, consistencyForCommit, true, queryStartNanoTime);
+                    RowIterator result = proposalPair.right;
+                    if (result != null)
+                        Tracing.trace("CAS did not apply");
+                    else
+                        Tracing.trace("CAS applied successfully");
+                    return result;
+                }
+
+                Tracing.trace("Paxos proposal not accepted (pre-empted by a higher ballot)");
+                contentions++;
+                Uninterruptibles.sleepUninterruptibly(ThreadLocalRandom.current().nextInt(100), TimeUnit.MILLISECONDS);
+                // continue to retry
+            }
+        }
+        catch (CasWriteTimeoutException e)
+        {
+            // Might be thrown by beginRepairAndPaxos. In that case, any contention that happened within the method and
+            // led up to the timeout was not accounted in our local 'contentions' variable and we add it now so it the
+            // contention recorded in the finally is correct.
+            contentions += e.contentions;
+            throw e;
+        }
+        catch (WriteTimeoutException e)
+        {
+            // Might be thrown by proposePaxos or commitPaxos
+            throw new CasWriteTimeoutException(e.writeType, e.consistency, e.received, e.blockFor, contentions);
+        }
+        finally
+        {
+            recordCasContention(metadata, key, casMetrics, contentions);
+        }
+
+        throw new CasWriteTimeoutException(WriteType.CAS, consistencyForPaxos, 0, consistencyForPaxos.blockFor(keyspace), contentions);
     }
 
     /**
@@ -427,7 +544,7 @@ public class StorageProxy implements StorageProxyMBean
                                                                 ReplicaPlan.ForPaxosWrite paxosPlan,
                                                                 ConsistencyLevel consistencyForPaxos,
                                                                 ConsistencyLevel consistencyForCommit,
-                                                                final boolean isWrite,
+                                                                CASClientRequestMetrics casMetrics,
                                                                 ClientState state)
     throws WriteTimeoutException, WriteFailureException
     {
@@ -448,76 +565,87 @@ public class StorageProxy implements StorageProxyMBean
             UUID ballot = UUIDGen.getRandomTimeUUIDFromMicros(ballotMicros);
 
             // prepare
-            Tracing.trace("Preparing {}", ballot);
+            try
-            Commit toPrepare = Commit.newPrepare(key, metadata, ballot);
-            summary = preparePaxos(toPrepare, paxosPlan, queryStartNanoTime);
-            if (!summary.promised)
             {
-                Tracing.trace("Some replicas have already promised a higher ballot than ours; aborting");
+                Tracing.trace("Preparing {}", ballot);
-                contentions++;
+                Commit toPrepare = Commit.newPrepare(key, metadata, ballot);
-                // sleep a random amount to give the other proposer a chance to finish
+                summary = preparePaxos(toPrepare, paxosPlan, queryStartNanoTime);
-                Uninterruptibles.sleepUninterruptibly(ThreadLocalRandom.current().nextInt(100), MILLISECONDS);
+                if (!summary.promised)
-                continue;
-            }
-
-            Commit inProgress = summary.mostRecentInProgressCommitWithUpdate;
-            Commit mostRecent = summary.mostRecentCommit;
-
-            // If we have an in-progress ballot greater than the MRC we know, then it's an in-progress round that
-            // needs to be completed, so do it.
-            if (!inProgress.update.isEmpty() && inProgress.isAfter(mostRecent))
-            {
-                Tracing.trace("Finishing incomplete paxos round {}", inProgress);
-                if(isWrite)
-                    casWriteMetrics.unfinishedCommit.inc();
-                else
-                    casReadMetrics.unfinishedCommit.inc();
-                Commit refreshedInProgress = Commit.newProposal(ballot, inProgress.update);
-                if (proposePaxos(refreshedInProgress, paxosPlan, false, queryStartNanoTime))
                 {
-                    try
+                    Tracing.trace("Some replicas have already promised a higher ballot than ours; aborting");
+                    contentions++;
+                    // sleep a random amount to give the other proposer a chance to finish
+                    Uninterruptibles.sleepUninterruptibly(ThreadLocalRandom.current().nextInt(100), MILLISECONDS);
+                    continue;
+                }
+
+                Commit inProgress = summary.mostRecentInProgressCommit;
+                Commit mostRecent = summary.mostRecentCommit;
+
+                // If we have an in-progress ballot greater than the MRC we know, then it's an in-progress round that
+                // needs to be completed, so do it.
+                // One special case we make is for update that are empty (which are proposed by serial reads and
+                // non-applying CAS). While we could handle those as any other updates, we can optimize this somewhat by
+                // neither committing those empty updates, nor replaying in-progress ones. The reasoning is this: as the
+                // update is empty, we have nothing to apply to storage in the commit phase, so the only reason to commit
+                // would be to update the MRC. However, if we skip replaying those empty updates, then we don't need to
+                // update the MRC for following updates to make progress (that is, if we didn't had the empty update skip
+                // below _but_ skipped updating the MRC on empty updates, then we'd be stuck always proposing that same
+                // empty update). And the reason skipping that replay is safe is that when an operation tries to propose
+                // an empty value, there can be only 2 cases:
+                //  1) the propose succeed, meaning a quorum of nodes accept it, in which case we are guaranteed no earlier
+                //     pending operation can ever be replayed (which is what we want to guarantee with the empty update).
+                //  2) the propose does not succeed. But then the operation proposing the empty update will not succeed
+                //     either (it will retry or ultimately timeout), and we're actually ok if earlier pending operation gets
+                //     replayed in that case.
+                // Tl;dr, it is safe to skip committing empty updates _as long as_ we also skip replying them below. And
+                // doing is more efficient, so we do so.
+                if (!inProgress.update.isEmpty() && inProgress.isAfter(mostRecent))
+                {
+                    Tracing.trace("Finishing incomplete paxos round {}", inProgress);
+                    casMetrics.unfinishedCommit.inc();
+                    Commit refreshedInProgress = Commit.newProposal(ballot, inProgress.update);
+                    if (proposePaxos(refreshedInProgress, paxosPlan, false, queryStartNanoTime))
                     {
                         commitPaxos(refreshedInProgress, consistencyForCommit, false, queryStartNanoTime);
                     }
-                    catch (WriteTimeoutException e)
+                    else
                     {
-                        recordCasContention(contentions);
+                        Tracing.trace("Some replicas have already promised a higher ballot than ours; aborting");
-                        // We're still doing preparation for the paxos rounds, so we want to use the CAS (see CASSANDRA-8672)
+                        // sleep a random amount to give the other proposer a chance to finish
-                        throw new WriteTimeoutException(WriteType.CAS, e.consistency, e.received, e.blockFor);
+                        contentions++;
+                        Uninterruptibles.sleepUninterruptibly(ThreadLocalRandom.current().nextInt(100), MILLISECONDS);
                     }
+                    continue;
                 }
-                else
+
+                // To be able to propose our value on a new round, we need a quorum of replica to have learn the previous one. Why is explained at:
+                // https://issues.apache.org/jira/browse/CASSANDRA-5062?focusedCommentId=13619810&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-13619810)
+                // Since we waited for quorum nodes, if some of them haven't seen the last commit (which may just be a timing issue, but may also
+                // mean we lost messages), we pro-actively "repair" those nodes, and retry.
+                int nowInSec = Ints.checkedCast(TimeUnit.MICROSECONDS.toSeconds(ballotMicros));
+                Iterable<InetAddressAndPort> missingMRC = summary.replicasMissingMostRecentCommit(metadata, nowInSec);
+                if (Iterables.size(missingMRC) > 0)
                 {
-                    Tracing.trace("Some replicas have already promised a higher ballot than ours; aborting");
+                    Tracing.trace("Repairing replicas that missed the most recent commit");
-                    // sleep a random amount to give the other proposer a chance to finish
+                    sendCommit(mostRecent, missingMRC);
-                    contentions++;
+                    // TODO: provided commits don't invalid the prepare we just did above (which they don't), we could just wait
-                    Uninterruptibles.sleepUninterruptibly(ThreadLocalRandom.current().nextInt(100), MILLISECONDS);
+                    // for all the missingMRC to acknowledge this commit and then move on with proposing our value. But that means
+                    // adding the ability to have commitPaxos block, which is exactly CASSANDRA-5442 will do. So once we have that
+                    // latter ticket, we can pass CL.ALL to the commit above and remove the 'continue'.
+                    continue;
                 }
-                continue;
-            }
 
-            // To be able to propose our value on a new round, we need a quorum of replica to have learn the previous one. Why is explained at:
+                return new PaxosBallotAndContention(ballot, contentions);
-            // https://issues.apache.org/jira/browse/CASSANDRA-5062?focusedCommentId=13619810&page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel#comment-13619810)
+            }
-            // Since we waited for quorum nodes, if some of them haven't seen the last commit (which may just be a timing issue, but may also
+            catch (WriteTimeoutException e)
-            // mean we lost messages), we pro-actively "repair" those nodes, and retry.
-            int nowInSec = Ints.checkedCast(TimeUnit.MICROSECONDS.toSeconds(ballotMicros));
-            Iterable<InetAddressAndPort> missingMRC = summary.replicasMissingMostRecentCommit(metadata, nowInSec);
-            if (Iterables.size(missingMRC) > 0)
             {
-                Tracing.trace("Repairing replicas that missed the most recent commit");
+                // We're still doing preparation for the paxos rounds, so we want to use the CAS (see CASSANDRA-8672)
-                sendCommit(mostRecent, missingMRC);
+                throw new CasWriteTimeoutException(WriteType.CAS, e.consistency, e.received, e.blockFor, contentions);
-                // TODO: provided commits don't invalid the prepare we just did above (which they don't), we could just wait
-                // for all the missingMRC to acknowledge this commit and then move on with proposing our value. But that means
-                // adding the ability to have commitPaxos block, which is exactly CASSANDRA-5442 will do. So once we have that
-                // latter ticket, we can pass CL.ALL to the commit above and remove the 'continue'.
-                continue;
             }
-
-            return new PaxosBallotAndContention(ballot, contentions);
         }
 
-        recordCasContention(contentions);
+        throw new CasWriteTimeoutException(WriteType.CAS, consistencyForPaxos, 0, consistencyForPaxos.blockFor(Keyspace.open(metadata.keyspace)), contentions);
-        throw new WriteTimeoutException(WriteType.CAS, consistencyForPaxos, 0, consistencyForPaxos.blockFor(Keyspace.open(metadata.keyspace)));
     }
 
     /**
@@ -1643,19 +1771,31 @@ public class StorageProxy implements StorageProxyMBean
         PartitionIterator result = null;
         try
         {
-            // make sure any in-progress paxos writes are done (i.e., committed to a majority of replicas), before performing a quorum read
+            final ConsistencyLevel consistencyForReplayCommitsOrFetch = consistencyLevel == ConsistencyLevel.LOCAL_SERIAL
-            ReplicaPlan.ForPaxosWrite replicaPlan = ReplicaPlans.forPaxos(Keyspace.open(metadata.keyspace), key, consistencyLevel);
+                                                                        ? ConsistencyLevel.LOCAL_QUORUM
-
+                                                                        : ConsistencyLevel.QUORUM;
-            // does the work of applying in-progress writes; throws UAE or timeout if it can't
-            final ConsistencyLevel consistencyForCommitOrFetch = consistencyLevel == ConsistencyLevel.LOCAL_SERIAL
-                                                                                   ? ConsistencyLevel.LOCAL_QUORUM
-                                                                                   : ConsistencyLevel.QUORUM;
 
             try
             {
-                final PaxosBallotAndContention pair = beginAndRepairPaxos(start, key, metadata, replicaPlan, consistencyLevel, consistencyForCommitOrFetch, false, state);
+                // Commit an empty update to make sure all in-progress updates that should be finished first is, _and_
-                if (pair.contentions > 0)
+                // that no other in-progress can get resurrected.
-                    casReadMetrics.contention.update(pair.contentions);
+                Supplier<Pair<PartitionUpdate, RowIterator>> updateProposer =
+                    disableSerialReadLinearizability
+                    ? () -> null
+                    : () -> Pair.create(PartitionUpdate.emptyUpdate(metadata, key), null);
+                // When replaying, we commit at quorum/local quorum, as we want to be sure the following read (done at
+                // quorum/local_quorum) sees any replayed updates. Our own update is however empty, and those don't even
+                // get committed due to an optimiation described in doPaxos/beingRepairAndPaxos, so the commit
+                // consistency is irrelevant (we use ANY just to emphasis that we don't wait on our commit).
+                doPaxos(metadata,
+                        key,
+                        consistencyLevel,
+                        consistencyForReplayCommitsOrFetch,
+                        ConsistencyLevel.ANY,
+                        state,
+                        start,
+                        casReadMetrics,
+                        updateProposer);
             }
             catch (WriteTimeoutException e)
             {
@@ -1666,7 +1806,7 @@ public class StorageProxy implements StorageProxyMBean
                 throw new ReadFailureException(consistencyLevel, e.received, e.blockFor, false, e.failureReasonByEndpoint);
             }
 
-            result = fetchRows(group.queries, consistencyForCommitOrFetch, queryStartNanoTime);
+            result = fetchRows(group.queries, consistencyForReplayCommitsOrFetch, queryStartNanoTime);
         }
         catch (UnavailableException e)
         {

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

@@ -78,6 +78,12 @@ public class Commit
         return this.ballot.equals(ballot);
     }
 
+    /** Whether this is an empty commit, that is one with no updates. */
+    public boolean isEmpty()
+    {
+        return update.isEmpty();
+    }
+
     public Mutation makeMutation()
     {
         return new Mutation(update);

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

@@ -46,7 +46,6 @@ public class PrepareCallback extends AbstractPaxosCallback<PrepareResponse>
     public boolean promised = true;
     public Commit mostRecentCommit;
     public Commit mostRecentInProgressCommit;
-    public Commit mostRecentInProgressCommitWithUpdate;
 
     private final Map<InetAddressAndPort, Commit> commitsByReplica = new ConcurrentHashMap<>();
 
@@ -56,7 +55,6 @@ public class PrepareCallback extends AbstractPaxosCallback<PrepareResponse>
         // need to inject the right key in the empty commit so comparing with empty commits in the response works as expected
         mostRecentCommit = Commit.emptyCommit(key, metadata);
         mostRecentInProgressCommit = Commit.emptyCommit(key, metadata);
-        mostRecentInProgressCommitWithUpdate = Commit.emptyCommit(key, metadata);
     }
 
     public synchronized void onResponse(Message<PrepareResponse> message)
@@ -64,9 +62,8 @@ public class PrepareCallback extends AbstractPaxosCallback<PrepareResponse>
         PrepareResponse response = message.payload;
         logger.trace("Prepare response {} from {}", response, message.from());
 
-        // In case of clock skew, another node could be proposing with ballot that are quite a bit
+        // We set the mostRecentInProgressCommit even if we're not promised as, in that case, the ballot of that commit
-        // older than our own. In that case, we record the more recent commit we've received to make
+        // will be used to avoid generating a ballot that has not chance to win on retry (think clock skew).
-        // sure we re-prepare on an older ballot.
         if (response.inProgressCommit.isAfter(mostRecentInProgressCommit))
             mostRecentInProgressCommit = response.inProgressCommit;
 
@@ -82,11 +79,6 @@ public class PrepareCallback extends AbstractPaxosCallback<PrepareResponse>
         if (response.mostRecentCommit.isAfter(mostRecentCommit))
             mostRecentCommit = response.mostRecentCommit;
 
-        // If some response has an update, then we should replay the update with the highest ballot. So find
-        // the the highest commit that actually have an update
-        if (response.inProgressCommit.isAfter(mostRecentInProgressCommitWithUpdate) && !response.inProgressCommit.update.isEmpty())
-            mostRecentInProgressCommitWithUpdate = response.inProgressCommit;
-
         latch.countDown();
     }
 

test/distributed/org/apache/cassandra/distributed/impl/Instance.java

@@ -64,6 +64,8 @@ import org.apache.cassandra.db.SystemKeyspace;
 import org.apache.cassandra.db.SystemKeyspaceMigrator40;
 import org.apache.cassandra.db.commitlog.CommitLog;
 import org.apache.cassandra.db.compaction.CompactionManager;
+import org.apache.cassandra.dht.IPartitioner;
+import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.distributed.Cluster;
 import org.apache.cassandra.distributed.Constants;
 import org.apache.cassandra.distributed.action.GossipHelper;
@@ -79,7 +81,9 @@ import org.apache.cassandra.distributed.api.NodeToolResult;
 import org.apache.cassandra.distributed.api.SimpleQueryResult;
 import org.apache.cassandra.distributed.mock.nodetool.InternalNodeProbe;
 import org.apache.cassandra.distributed.mock.nodetool.InternalNodeProbeFactory;
+import org.apache.cassandra.gms.ApplicationState;
 import org.apache.cassandra.gms.Gossiper;
+import org.apache.cassandra.gms.VersionedValue;
 import org.apache.cassandra.hints.HintsService;
 import org.apache.cassandra.index.SecondaryIndexManager;
 import org.apache.cassandra.io.IVersionedAsymmetricSerializer;
@@ -538,6 +542,85 @@ public class Instance extends IsolatedExecutor implements IInvokableInstance
         return YamlConfigurationLoader.fromMap(params, check, Config.class);
     }
 
+    public static void addToRing(boolean bootstrapping, IInstance peer)
+    {
+        try
+        {
+            IInstanceConfig config = peer.config();
+            IPartitioner partitioner = FBUtilities.newPartitioner(config.getString("partitioner"));
+            Token token = partitioner.getTokenFactory().fromString(config.getString("initial_token"));
+            InetAddressAndPort addressAndPort = toCassandraInetAddressAndPort(peer.broadcastAddress());
+
+            UUID hostId = config.hostId();
+            Gossiper.runInGossipStageBlocking(() -> {
+                Gossiper.instance.initializeNodeUnsafe(addressAndPort, hostId, 1);
+                Gossiper.instance.injectApplicationState(addressAndPort,
+                                                         ApplicationState.TOKENS,
+                                                         new VersionedValue.VersionedValueFactory(partitioner).tokens(Collections.singleton(token)));
+                StorageService.instance.onChange(addressAndPort,
+                                                 ApplicationState.STATUS,
+                                                 bootstrapping
+                                                 ? new VersionedValue.VersionedValueFactory(partitioner).bootstrapping(Collections.singleton(token))
+                                                 : new VersionedValue.VersionedValueFactory(partitioner).normal(Collections.singleton(token)));
+                Gossiper.instance.realMarkAlive(addressAndPort, Gossiper.instance.getEndpointStateForEndpoint(addressAndPort));
+            });
+            int messagingVersion = peer.isShutdown()
+                    ? MessagingService.current_version
+                    : Math.min(MessagingService.current_version, peer.getMessagingVersion());
+            MessagingService.instance().versions.set(addressAndPort, messagingVersion);
+
+            assert bootstrapping || StorageService.instance.getTokenMetadata().isMember(addressAndPort);
+            PendingRangeCalculatorService.instance.blockUntilFinished();
+        }
+        catch (Throwable e) // UnknownHostException
+        {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public static void removeFromRing(IInstance peer)
+    {
+        try
+        {
+            IInstanceConfig config = peer.config();
+            IPartitioner partitioner = FBUtilities.newPartitioner(config.getString("partitioner"));
+            Token token = partitioner.getTokenFactory().fromString(config.getString("initial_token"));
+            InetAddressAndPort addressAndPort = toCassandraInetAddressAndPort(peer.broadcastAddress());
+
+            Gossiper.runInGossipStageBlocking(() -> {
+                StorageService.instance.onChange(addressAndPort,
+                        ApplicationState.STATUS,
+                        new VersionedValue.VersionedValueFactory(partitioner).left(Collections.singleton(token), 0L));
+                Gossiper.instance.removeEndpoint(addressAndPort);
+            });
+            PendingRangeCalculatorService.instance.blockUntilFinished();
+        }
+        catch (Throwable e) // UnknownHostException
+        {
+            throw new RuntimeException(e);
+        }
+    }
+
+    public static void addToRingNormal(IInstance peer)
+    {
+        addToRing(false, peer);
+        assert StorageService.instance.getTokenMetadata().isMember(toCassandraInetAddressAndPort(peer.broadcastAddress()));
+    }
+
+    public static void addToRingBootstrapping(IInstance peer)
+    {
+        addToRing(true, peer);
+    }
+
+    private static void initializeRing(ICluster cluster)
+    {
+        for (int i = 1 ; i <= cluster.size() ; ++i)
+            addToRing(false, cluster.get(i));
+
+        for (int i = 1; i <= cluster.size(); ++i)
+            assert StorageService.instance.getTokenMetadata().isMember(toCassandraInetAddressAndPort(cluster.get(i).broadcastAddress()));
+    }
+
     public Future<Void> shutdown()
     {
         return shutdown(true);

test/distributed/org/apache/cassandra/distributed/test/CASTest.java

@@ -0,0 +1,688 @@
+/*
+ * 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;
+
+import java.io.IOException;
+import java.util.UUID;
+import java.util.function.BiConsumer;
+
+
+import org.junit.Assert;
+import org.junit.Ignore;
+import org.junit.Test;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.marshal.Int32Type;
+import org.apache.cassandra.dht.Murmur3Partitioner;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+import org.apache.cassandra.distributed.api.ICoordinator;
+import org.apache.cassandra.distributed.api.IInstance;
+import org.apache.cassandra.distributed.api.IMessageFilters;
+import org.apache.cassandra.distributed.impl.Instance;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.service.StorageService;
+import org.apache.cassandra.utils.UUIDGen;
+
+import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
+import static org.apache.cassandra.distributed.shared.AssertUtils.fail;
+import static org.apache.cassandra.distributed.shared.AssertUtils.row;
+import static org.apache.cassandra.net.Verb.PAXOS_COMMIT_REQ;
+import static org.apache.cassandra.net.Verb.PAXOS_PREPARE_REQ;
+import static org.apache.cassandra.net.Verb.PAXOS_PROPOSE_REQ;
+import static org.apache.cassandra.net.Verb.READ_REQ;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class CASTest extends TestBaseImpl
+{
+    private static final Logger logger = LoggerFactory.getLogger(CASTest.class);
+
+    @Test
+    public void simpleUpdate() throws Throwable
+    {
+        try (Cluster cluster = init(Cluster.create(3)))
+        {
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+            cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1) IF NOT EXISTS", ConsistencyLevel.QUORUM);
+            assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", ConsistencyLevel.SERIAL),
+                    row(1, 1, 1));
+            cluster.coordinator(1).execute("UPDATE " + KEYSPACE + ".tbl SET v = 3 WHERE pk = 1 and ck = 1 IF v = 2", ConsistencyLevel.QUORUM);
+            assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", ConsistencyLevel.SERIAL),
+                    row(1, 1, 1));
+            cluster.coordinator(1).execute("UPDATE " + KEYSPACE + ".tbl SET v = 2 WHERE pk = 1 and ck = 1 IF v = 1", ConsistencyLevel.QUORUM);
+            assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", ConsistencyLevel.SERIAL),
+                    row(1, 1, 2));
+        }
+    }
+
+    @Test
+    public void incompletePrepare() throws Throwable
+    {
+        try (Cluster cluster = init(Cluster.create(3, config -> config.set("write_request_timeout_in_ms", 200L).set("cas_contention_timeout_in_ms", 200L))))
+        {
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+            IMessageFilters.Filter drop = cluster.filters().verbs(PAXOS_PREPARE_REQ.id).from(1).to(2, 3).drop();
+            try
+            {
+                cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1) IF NOT EXISTS", ConsistencyLevel.QUORUM);
+                Assert.fail();
+            }
+            catch (RuntimeException e)
+            {
+                Assert.assertEquals("CAS operation timed out - encountered contentions: 0", e.getMessage());
+            }
+            drop.off();
+            cluster.coordinator(1).execute("UPDATE " + KEYSPACE + ".tbl SET v = 2 WHERE pk = 1 and ck = 1 IF v = 1", ConsistencyLevel.QUORUM);
+            assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", ConsistencyLevel.SERIAL));
+        }
+    }
+
+    @Test
+    public void incompletePropose() throws Throwable
+    {
+        try (Cluster cluster = init(Cluster.create(3, config -> config.set("write_request_timeout_in_ms", 200L).set("cas_contention_timeout_in_ms", 200L))))
+        {
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+            IMessageFilters.Filter drop1 = cluster.filters().verbs(PAXOS_PROPOSE_REQ.id).from(1).to(2, 3).drop();
+            try
+            {
+                cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1) IF NOT EXISTS", ConsistencyLevel.QUORUM);
+                Assert.fail();
+            }
+            catch (RuntimeException e)
+            {
+                Assert.assertEquals("CAS operation timed out - encountered contentions: 0", e.getMessage());
+            }
+            drop1.off();
+            // make sure we encounter one of the in-progress proposals so we complete it
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id).from(1).to(2).drop();
+            cluster.coordinator(1).execute("UPDATE " + KEYSPACE + ".tbl SET v = 2 WHERE pk = 1 and ck = 1 IF v = 1", ConsistencyLevel.QUORUM);
+            assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", ConsistencyLevel.SERIAL),
+                    row(1, 1, 2));
+        }
+    }
+
+    @Test
+    public void incompleteCommit() throws Throwable
+    {
+        try (Cluster cluster = init(Cluster.create(3, config -> config.set("write_request_timeout_in_ms", 200L).set("cas_contention_timeout_in_ms", 200L))))
+        {
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+            IMessageFilters.Filter drop1 = cluster.filters().verbs(PAXOS_COMMIT_REQ.id).from(1).to(2, 3).drop();
+            try
+            {
+                cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (1, 1, 1) IF NOT EXISTS", ConsistencyLevel.QUORUM);
+                Assert.fail();
+            }
+            catch (RuntimeException e)
+            {
+                Assert.assertEquals("CAS operation timed out - encountered contentions: 0", e.getMessage());
+            }
+            drop1.off();
+            // make sure we see one of the successful commits
+            cluster.filters().verbs(PAXOS_PROPOSE_REQ.id).from(1).to(2).drop();
+            cluster.coordinator(1).execute("UPDATE " + KEYSPACE + ".tbl SET v = 2 WHERE pk = 1 and ck = 1 IF v = 1", ConsistencyLevel.QUORUM);
+            assertRows(cluster.coordinator(1).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = 1", ConsistencyLevel.SERIAL),
+                    row(1, 1, 2));
+        }
+    }
+
+    private int[] paxosAndReadVerbs() {
+        return new int[] { PAXOS_PREPARE_REQ.id, PAXOS_PROPOSE_REQ.id, PAXOS_COMMIT_REQ.id, READ_REQ.id };
+    }
+
+    /**
+     * Base test to ensure that if a write times out but with a proposal accepted by some nodes (less then quorum), and
+     * a following SERIAL operation does not observe that write (the node having accepted it do not participate in that
+     * following operation), then that write is never applied, even when the nodes having accepted the original proposal
+     * participate.
+     *
+     * <p>In other words, if an operation timeout, it may or may not be applied, but that "fate" is persistently decided
+     * by the very SERIAL operation that "succeed" (in the sense of 'not timing out or throwing some other exception').
+     *
+     * @param postTimeoutOperation1 a SERIAL operation executed after an initial write that inserts the row [0, 0] times
+     *                              out. It is executed with a QUORUM of nodes that have _not_ see the timed out
+     *                              proposal, and so that operation should expect that the [0, 0] write has not taken
+     *                              place.
+     * @param postTimeoutOperation2 a 2nd SERIAL operation executed _after_ {@code postTimeoutOperation1}, with no
+     *                              write executed between the 2 operation. Contrarily to the 1st operation, the QORUM
+     *                              for this operation _will_ include the node that got the proposal for the [0, 0]
+     *                              insert but didn't participated to {@code postTimeoutOperation1}}. That operation
+     *                              should also no witness that [0, 0] write (since {@code postTimeoutOperation1}
+     *                              didn't).
+     * @param loseCommitOfOperation1 if {@code true}, the test will also drop the "commits" messages for
+     *                               {@code postTimeoutOperation1}. In general, the test should behave the same with or
+     *                               without that flag since a value is decided as soon as it has been "accepted by
+     *                               quorum" and the commits should always be properly replayed.
+     */
+    private void consistencyAfterWriteTimeoutTest(BiConsumer<String, ICoordinator> postTimeoutOperation1,
+                                                  BiConsumer<String, ICoordinator> postTimeoutOperation2,
+                                                  boolean loseCommitOfOperation1) throws IOException
+    {
+        // It's unclear why (haven't dug), but in some of the instance of this test method, there is a consistent 2+
+        // seconds pauses between the prepare and propose phases during the execution of 'postTimeoutOperation2'. This
+        // does not happen on 3.0 and there is no report of such long pauses otherwise, so an hypothesis is that this
+        // is due to the in-jvm dtest framework. This is is why we use a 4 seconds timeout here. Given this test is
+        // not about performance, this is probably ok, even if we ideally should dug into the underlying reason.
+        try (Cluster cluster = init(Cluster.create(3, config -> config.set("write_request_timeout_in_ms", 4000L)
+                                                                      .set("cas_contention_timeout_in_ms", 200L))))
+        {
+            String table = KEYSPACE + ".t";
+            cluster.schemaChange("CREATE TABLE " + table + " (k int PRIMARY KEY, v int)");
+
+            // We do a CAS insertion, but have with the PROPOSE message dropped on node 1 and 2. The CAS will not get
+            // through and should timeout. Importantly, node 3 does receive and answer the PROPOSE.
+            IMessageFilters.Filter dropProposeFilter = cluster.filters()
+                                                              .inbound()
+                                                              .verbs(PAXOS_PROPOSE_REQ.id)
+                                                              .from(3)
+                                                              .to(1, 2)
+                                                              .drop();
+            try
+            {
+                // NOTE: the consistency below is the "commit" one, so it doesn't matter at all here.
+                // NOTE 2: we use node 3 as coordinator because message filters don't currently work for locally
+                //   delivered messages and as we want to drop messages to 1 and 2, we can't use them.
+                cluster.coordinator(3)
+                       .execute("INSERT INTO " + table + "(k, v) VALUES (0, 0) IF NOT EXISTS", ConsistencyLevel.ONE);
+                fail("The insertion should have timed-out");
+            }
+            catch (Exception e)
+            {
+                // We expect a write timeout. If we get one, the test can continue, otherwise, we rethrow. Note that we
+                // look at the root cause because the dtest framework effectively wrap the exception in a RuntimeException
+                // (we could just look at the immediate cause, but this feel a bit more resilient this way).
+                // TODO: we can't use an instanceof below because the WriteTimeoutException we get is from a different class
+                //  loader than the one the test run under, and that's our poor-man work-around. This kind of things should
+                //  be improved at the dtest API level.
+                if (!e.getClass().getSimpleName().equals("CasWriteTimeoutException"))
+                    throw e;
+            }
+            finally
+            {
+                dropProposeFilter.off();
+            }
+
+            // Isolates node 3 and executes the SERIAL operation. As neither node 1 or 2 got the initial insert proposal,
+            // there is nothing to "replay" and the operation should assert the table is still empty.
+            IMessageFilters.Filter ignoreNode3Filter = cluster.filters().verbs(paxosAndReadVerbs()).to(3).drop();
+            IMessageFilters.Filter dropCommitFilter = null;
+            if (loseCommitOfOperation1)
+            {
+                dropCommitFilter = cluster.filters().verbs(PAXOS_COMMIT_REQ.id).to(1, 2).drop();
+            }
+            try
+            {
+                postTimeoutOperation1.accept(table, cluster.coordinator(1));
+            }
+            finally
+            {
+                ignoreNode3Filter.off();
+                if (dropCommitFilter != null)
+                    dropCommitFilter.off();
+            }
+
+            // Node 3 is now back and we isolate node 2 to ensure the next read hits node 1 and 3.
+            // What we want to ensure is that despite node 3 having the initial insert in its paxos state in a position of
+            // being replayed, that insert is _not_ replayed (it would contradict serializability since the previous
+            // operation asserted nothing was inserted). It is this execution that failed before CASSANDRA-12126.
+            IMessageFilters.Filter ignoreNode2Filter = cluster.filters().verbs(paxosAndReadVerbs()).to(2).drop();
+            try
+            {
+                postTimeoutOperation2.accept(table, cluster.coordinator(1));
+            }
+            finally
+            {
+                ignoreNode2Filter.off();
+            }
+        }
+    }
+
+    /**
+     * Tests that if a write timeouts and a following serial read does not see that write, then no following reads sees
+     * it, even if some nodes still have the write in their paxos state.
+     *
+     * <p>This specifically test for the inconsistency described/fixed by CASSANDRA-12126.
+     */
+    @Test
+    public void readConsistencyAfterWriteTimeoutTest() throws IOException
+    {
+        BiConsumer<String, ICoordinator> operation =
+            (table, coordinator) -> assertRows(coordinator.execute("SELECT * FROM " + table + " WHERE k=0",
+                                                                   ConsistencyLevel.SERIAL));
+
+        consistencyAfterWriteTimeoutTest(operation, operation, false);
+        consistencyAfterWriteTimeoutTest(operation, operation, true);
+    }
+
+    /**
+     * Tests that if a write timeouts, then a following CAS succeed but does not apply in a way that indicate the write
+     * has not applied, then no following CAS can see that initial insert , even if some nodes still have the write in
+     * their paxos state.
+     *
+     * <p>This specifically test for the inconsistency described/fixed by CASSANDRA-12126.
+     */
+    @Test
+    public void nonApplyingCasConsistencyAfterWriteTimeout() throws IOException
+    {
+        // Note: we use CL.ANY so that the operation don't timeout in the case where we "lost" the operation1 commits.
+        // The commit CL shouldn't have impact on this test anyway, so this doesn't diminishes the test.
+        BiConsumer<String, ICoordinator> operation =
+            (table, coordinator) -> assertCasNotApplied(coordinator.execute("UPDATE " + table + " SET v = 1 WHERE k = 0 IF v = 0",
+                                                                            ConsistencyLevel.ANY));
+        consistencyAfterWriteTimeoutTest(operation, operation, false);
+        consistencyAfterWriteTimeoutTest(operation, operation, true);
+    }
+
+    /**
+     * Tests that if a write timeouts and a following serial read does not see that write, then no following CAS see
+     * that initial insert, even if some nodes still have the write in their paxos state.
+     *
+     * <p>This specifically test for the inconsistency described/fixed by CASSANDRA-12126.
+     */
+    @Test
+    public void mixedReadAndNonApplyingCasConsistencyAfterWriteTimeout() throws IOException
+    {
+        BiConsumer<String, ICoordinator> operation1 =
+            (table, coordinator) -> assertRows(coordinator.execute("SELECT * FROM " + table + " WHERE k=0",
+                                                                   ConsistencyLevel.SERIAL));
+        BiConsumer<String, ICoordinator> operation2 =
+            (table, coordinator) -> assertCasNotApplied(coordinator.execute("UPDATE " + table + " SET v = 1 WHERE k = 0 IF v = 0",
+                                                                            ConsistencyLevel.QUORUM));
+        consistencyAfterWriteTimeoutTest(operation1, operation2, false);
+        consistencyAfterWriteTimeoutTest(operation1, operation2, true);
+    }
+
+    /**
+     * Tests that if a write timeouts and a following CAS succeed but does not apply in a way that indicate the write
+     * has not applied, then following serial reads do no see that write, even if some nodes still have the write in
+     * their paxos state.
+     *
+     * <p>This specifically test for the inconsistency described/fixed by CASSANDRA-12126.
+     */
+    @Test
+    public void mixedNonApplyingCasAndReadConsistencyAfterWriteTimeout() throws IOException
+    {
+        // Note: we use CL.ANY so that the operation don't timeout in the case where we "lost" the operation1 commits.
+        // The commit CL shouldn't have impact on this test anyway, so this doesn't diminishes the test.
+        BiConsumer<String, ICoordinator> operation1 =
+            (table, coordinator) -> assertCasNotApplied(coordinator.execute("UPDATE " + table + " SET v = 1 WHERE k = 0 IF v = 0",
+                                                                            ConsistencyLevel.ANY));
+        BiConsumer<String, ICoordinator> operation2 =
+            (table, coordinator) -> assertRows(coordinator.execute("SELECT * FROM " + table + " WHERE k=0",
+                                                                   ConsistencyLevel.SERIAL));
+        consistencyAfterWriteTimeoutTest(operation1, operation2, false);
+        consistencyAfterWriteTimeoutTest(operation1, operation2, true);
+    }
+
+    // TODO: this shoud probably be moved into the dtest API.
+    private void assertCasNotApplied(Object[][] resultSet)
+    {
+        assertFalse("Expected a CAS resultSet (with at least application result) but got an empty one.",
+                    resultSet.length == 0);
+        assertFalse("Invalid empty first row in CAS resultSet.", resultSet[0].length == 0);
+        Object wasApplied = resultSet[0][0];
+        assertTrue("Expected 1st column of CAS resultSet to be a boolean, but got a " + wasApplied.getClass(),
+                   wasApplied instanceof Boolean);
+        assertFalse("Expected CAS to not be applied, but was applied.", (Boolean)wasApplied);
+    }
+
+    /**
+     * Failed write (by node that did not yet witness a range movement via gossip) is witnessed later as successful
+     * conflicting with another successful write performed by a node that did witness the range movement
+     * Prepare, Propose and Commit A to {1, 2}
+     * Range moves to {2, 3, 4}
+     * Prepare and Propose B (=> !A) to {3, 4}
+     */
+    @Ignore
+    @Test
+    public void testSuccessfulWriteBeforeRangeMovement() throws Throwable
+    {
+        try (Cluster cluster = Cluster.create(4, config -> config
+                .set("write_request_timeout_in_ms", 200L)
+                .set("cas_contention_timeout_in_ms", 200L)))
+        {
+            cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3};");
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v1 int, v2 int, PRIMARY KEY (pk, ck))");
+
+            // make it so {1} is unaware (yet) that {4} is an owner of the token
+            cluster.get(1).acceptsOnInstance(Instance::removeFromRing).accept(cluster.get(4));
+
+            int pk = pk(cluster, 1, 2);
+
+            // {1} promises and accepts on !{3} => {1, 2}; commits on !{2,3} => {1}
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id, READ_REQ.id).from(1).to(3).drop();
+            cluster.filters().verbs(PAXOS_PROPOSE_REQ.id).from(1).to(3).drop();
+            cluster.filters().verbs(PAXOS_COMMIT_REQ.id).from(1).to(2, 3).drop();
+            assertRows(cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v1) VALUES (?, 1, 1) IF NOT EXISTS", ConsistencyLevel.ONE, pk),
+                    row(true));
+
+            for (int i = 1 ; i <= 3 ; ++i)
+                cluster.get(i).acceptsOnInstance(Instance::addToRingNormal).accept(cluster.get(4));
+
+            // {4} reads from !{2} => {3, 4}
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id, READ_REQ.id).from(4).to(2).drop();
+            cluster.filters().verbs(PAXOS_PROPOSE_REQ.id).from(4).to(2).drop();
+            assertRows(cluster.coordinator(4).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v2) VALUES (?, 1, 2) IF NOT EXISTS", ConsistencyLevel.ONE, pk),
+                    row(false, pk, 1, 1, null));
+        }
+    }
+
+    /**
+     * Failed write (by node that did not yet witness a range movement via gossip) is witnessed later as successful
+     * conflicting with another successful write performed by a node that did witness the range movement
+     *  - Range moves from {1, 2, 3} to {2, 3, 4}, witnessed by X (not by !X)
+     *  -  X: Prepare, Propose and Commit A to {3, 4}
+     *  - !X: Prepare and Propose B (=>!A) to {1, 2}
+     */
+    @Ignore
+    @Test
+    public void testConflictingWritesWithStaleRingInformation() throws Throwable
+    {
+        try (Cluster cluster = Cluster.create(4, config -> config
+                .set("write_request_timeout_in_ms", 200L)
+                .set("cas_contention_timeout_in_ms", 200L)))
+        {
+            cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3};");
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v1 int, v2 int, PRIMARY KEY (pk, ck))");
+
+            // make it so {1} is unaware (yet) that {4} is an owner of the token
+            cluster.get(1).acceptsOnInstance(Instance::removeFromRing).accept(cluster.get(4));
+
+            // {4} promises, accepts and commits on !{2} => {3, 4}
+            int pk = pk(cluster, 1, 2);
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id, READ_REQ.id).from(4).to(2).drop();
+            cluster.filters().verbs(PAXOS_PROPOSE_REQ.id).from(4).to(2).drop();
+            cluster.filters().verbs(PAXOS_COMMIT_REQ.id).from(4).to(2).drop();
+            assertRows(cluster.coordinator(4).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v1) VALUES (?, 1, 1) IF NOT EXISTS", ConsistencyLevel.ONE, pk),
+                    row(true));
+
+            // {1} promises, accepts and commmits on !{3} => {1, 2}
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id, READ_REQ.id).from(1).to(3).drop();
+            cluster.filters().verbs(PAXOS_PROPOSE_REQ.id).from(1).to(3).drop();
+            cluster.filters().verbs(PAXOS_COMMIT_REQ.id).from(1).to(3).drop();
+            assertRows(cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v2) VALUES (?, 1, 2) IF NOT EXISTS", ConsistencyLevel.ONE, pk),
+                    row(false, pk, 1, 1, null));
+        }
+    }
+
+    /**
+     * Successful write during range movement, not witnessed by read after range movement.
+     * Very similar to {@link #testConflictingWritesWithStaleRingInformation}.
+     *
+     *  - Range moves from {1, 2, 3} to {2, 3, 4}; witnessed by X (not by !X)
+     *  -  !X: Prepare and Propose to {1, 2}
+     *  - Range movement witnessed by !X
+     *  - Any: Prepare and Read from {3, 4}
+     */
+    @Ignore
+    @Test
+    public void testSucccessfulWriteDuringRangeMovementFollowedByRead() throws Throwable
+    {
+        try (Cluster cluster = Cluster.create(4, config -> config
+                .set("write_request_timeout_in_ms", 200L)
+                .set("cas_contention_timeout_in_ms", 200L)))
+        {
+            cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3};");
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v int, PRIMARY KEY (pk, ck))");
+
+            // make it so {4} is bootstrapping, and this has not propagated to other nodes yet
+            for (int i = 1 ; i <= 4 ; ++i)
+                cluster.get(1).acceptsOnInstance(Instance::removeFromRing).accept(cluster.get(4));
+            cluster.get(4).acceptsOnInstance(Instance::addToRingBootstrapping).accept(cluster.get(4));
+
+            int pk = pk(cluster, 1, 2);
+
+            // {1} promises and accepts on !{3} => {1, 2}; commmits on !{2, 3} => {1}
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id, READ_REQ.id).from(1).to(3).drop();
+            cluster.filters().verbs(PAXOS_PROPOSE_REQ.id).from(1).to(3).drop();
+            cluster.filters().verbs(PAXOS_COMMIT_REQ.id).from(1).to(2, 3).drop();
+            assertRows(cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v) VALUES (?, 1, 1) IF NOT EXISTS", ConsistencyLevel.ONE, pk),
+                    row(true));
+
+            // finish topology change
+            for (int i = 1 ; i <= 4 ; ++i)
+                cluster.get(i).acceptsOnInstance(Instance::addToRingNormal).accept(cluster.get(4));
+
+            // {3} reads from !{2} => {3, 4}
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id, READ_REQ.id).from(3).to(2).drop();
+            assertRows(cluster.coordinator(3).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = ?", ConsistencyLevel.SERIAL, pk),
+                    row(pk, 1, 1));
+        }
+    }
+
+    /**
+     * Successful write during range movement not witnessed by write after range movement
+     *
+     *  - Range moves from {1, 2, 3} to {2, 3, 4}; witnessed by X (not by !X)
+     *  -  !X: Prepare and Propose to {1, 2}
+     *  - Range movement witnessed by !X
+     *  - Any: Prepare and Propose to {3, 4}
+     */
+    @Ignore
+    @Test
+    public void testSuccessfulWriteDuringRangeMovementFollowedByConflicting() throws Throwable
+    {
+        try (Cluster cluster = Cluster.create(4, config -> config
+                .set("write_request_timeout_in_ms", 200L)
+                .set("cas_contention_timeout_in_ms", 200L)))
+        {
+            cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3};");
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v1 int, v2 int, PRIMARY KEY (pk, ck))");
+
+            // make it so {4} is bootstrapping, and this has not propagated to other nodes yet
+            for (int i = 1 ; i <= 4 ; ++i)
+                cluster.get(1).acceptsOnInstance(Instance::removeFromRing).accept(cluster.get(4));
+            cluster.get(4).acceptsOnInstance(Instance::addToRingBootstrapping).accept(cluster.get(4));
+
+            int pk = pk(cluster, 1, 2);
+
+            // {1} promises and accepts on !{3} => {1, 2}; commits on !{2, 3} => {1}
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id, READ_REQ.id).from(1).to(3).drop();
+            cluster.filters().verbs(PAXOS_PROPOSE_REQ.id).from(1).to(3).drop();
+            cluster.filters().verbs(PAXOS_COMMIT_REQ.id).from(1).to(2, 3).drop();
+            assertRows(cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v1) VALUES (?, 1, 1) IF NOT EXISTS", ConsistencyLevel.ONE, pk),
+                    row(true));
+
+            // finish topology change
+            for (int i = 1 ; i <= 4 ; ++i)
+                cluster.get(i).acceptsOnInstance(Instance::addToRingNormal).accept(cluster.get(4));
+
+            // {3} reads from !{2} => {3, 4}
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id, READ_REQ.id).from(3).to(2).drop();
+            assertRows(cluster.coordinator(3).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v2) VALUES (?, 1, 2) IF NOT EXISTS", ConsistencyLevel.ONE, pk),
+                    row(false, pk, 1, 1, null));
+
+            // TODO: repair and verify base table state
+        }
+    }
+
+    /**
+     * During a range movement, a CAS may fail leaving side effects that are not witnessed by another operation
+     * being performed with stale ring information.
+     * This is a particular special case of stale ring information sequencing, which probably would be resolved
+     * by fixing each of the more isolated cases (but is unique, so deserving of its own test case).
+     * See CASSANDRA-15745
+     *
+     *  - Range moves from {1, 2, 3} to {2, 3, 4}; witnessed by X (not by !X)
+     *  -   X: Prepare to {2, 3, 4}
+     *  -   X: Propose to {4}
+     *  -  !X: Prepare and Propose to {1, 2}
+     *  - Range move visible by !X
+     *  - Any: Prepare and Read from {3, 4}
+     */
+    @Ignore
+    @Test
+    public void testIncompleteWriteFollowedBySuccessfulWriteWithStaleRingDuringRangeMovementFollowedByRead() throws Throwable
+    {
+        try (Cluster cluster = Cluster.create(4, config -> config
+                .set("write_request_timeout_in_ms", 200L)
+                .set("cas_contention_timeout_in_ms", 200L)))
+        {
+            cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3};");
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v1 int, v2 int, PRIMARY KEY (pk, ck))");
+
+            // make it so {4} is bootstrapping, and this has not propagated to other nodes yet
+            for (int i = 1 ; i <= 4 ; ++i)
+                cluster.get(1).acceptsOnInstance(Instance::removeFromRing).accept(cluster.get(4));
+            cluster.get(4).acceptsOnInstance(Instance::addToRingBootstrapping).accept(cluster.get(4));
+
+            int pk = pk(cluster, 1, 2);
+
+            // {4} promises and accepts on !{1} => {2, 3, 4}; commits on !{1, 2, 3} => {4}
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id, READ_REQ.id).from(4).to(1).drop();
+            cluster.filters().verbs(PAXOS_PROPOSE_REQ.id).from(4).to(1, 2, 3).drop();
+            try
+            {
+                cluster.coordinator(4).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v1) VALUES (?, 1, 1) IF NOT EXISTS", ConsistencyLevel.QUORUM, pk);
+                Assert.assertTrue(false);
+            }
+            catch (RuntimeException wrapped)
+            {
+                Assert.assertEquals("Operation timed out - received only 1 responses.", wrapped.getCause().getMessage());
+            }
+
+            // {1} promises and accepts on !{3} => {1, 2}; commits on !{2, 3} => {1}
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id, READ_REQ.id).from(1).to(3).drop();
+            cluster.filters().verbs(PAXOS_PROPOSE_REQ.id).from(1).to(3).drop();
+            cluster.filters().verbs(PAXOS_COMMIT_REQ.id).from(1).to(2, 3).drop();
+            assertRows(cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v2) VALUES (?, 1, 2) IF NOT EXISTS", ConsistencyLevel.ONE, pk),
+                    row(true));
+
+            // finish topology change
+            for (int i = 1 ; i <= 4 ; ++i)
+                cluster.get(i).acceptsOnInstance(Instance::addToRingNormal).accept(cluster.get(4));
+
+            // {3} reads from !{2} => {3, 4}
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id, READ_REQ.id).from(3).to(2).drop();
+            cluster.filters().verbs(PAXOS_PROPOSE_REQ.id).from(3).to(2).drop();
+            assertRows(cluster.coordinator(3).execute("SELECT * FROM " + KEYSPACE + ".tbl WHERE pk = ?", ConsistencyLevel.SERIAL, pk),
+                    row(pk, 1, null, 2));
+        }
+    }
+
+    /**
+     * During a range movement, a CAS may fail leaving side effects that are not witnessed by another operation
+     * being performed with stale ring information.
+     * This is a particular special case of stale ring information sequencing, which probably would be resolved
+     * by fixing each of the more isolated cases (but is unique, so deserving of its own test case).
+     * See CASSANDRA-15745
+     *
+     *  - Range moves from {1, 2, 3} to {2, 3, 4}; witnessed by X (not by !X)
+     *  -   X: Prepare to {2, 3, 4}
+     *  -   X: Propose to {4}
+     *  -  !X: Prepare and Propose to {1, 2}
+     *  - Range move visible by !X
+     *  - Any: Prepare and Propose to {3, 4}
+     */
+    @Ignore
+    @Test
+    public void testIncompleteWriteFollowedBySuccessfulWriteWithStaleRingDuringRangeMovementFollowedByWrite() throws Throwable
+    {
+        try (Cluster cluster = Cluster.create(4, config -> config
+                .set("write_request_timeout_in_ms", 200L)
+                .set("cas_contention_timeout_in_ms", 200L)))
+        {
+            cluster.schemaChange("CREATE KEYSPACE " + KEYSPACE + " WITH replication = {'class': 'SimpleStrategy', 'replication_factor': 3};");
+            cluster.schemaChange("CREATE TABLE " + KEYSPACE + ".tbl (pk int, ck int, v1 int, v2 int, PRIMARY KEY (pk, ck))");
+
+            // make it so {4} is bootstrapping, and this has not propagated to other nodes yet
+            for (int i = 1 ; i <= 4 ; ++i)
+                cluster.get(1).acceptsOnInstance(Instance::removeFromRing).accept(cluster.get(4));
+            cluster.get(4).acceptsOnInstance(Instance::addToRingBootstrapping).accept(cluster.get(4));
+
+            int pk = pk(cluster, 1, 2);
+
+            // {4} promises and accepts on !{1} => {2, 3, 4}; commits on !{1, 2, 3} => {4}
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id, READ_REQ.id).from(4).to(1).drop();
+            cluster.filters().verbs(PAXOS_PROPOSE_REQ.id).from(4).to(1, 2, 3).drop();
+            try
+            {
+                cluster.coordinator(4).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v1) VALUES (?, 1, 1) IF NOT EXISTS", ConsistencyLevel.QUORUM, pk);
+                Assert.assertTrue(false);
+            }
+            catch (RuntimeException wrapped)
+            {
+                Assert.assertEquals("Operation timed out - received only 1 responses.", wrapped.getCause().getMessage());
+            }
+
+            // {1} promises and accepts on !{3} => {1, 2}; commits on !{2, 3} => {1}
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id, READ_REQ.id).from(1).to(3).drop();
+            cluster.filters().verbs(PAXOS_PROPOSE_REQ.id).from(1).to(3).drop();
+            cluster.filters().verbs(PAXOS_COMMIT_REQ.id).from(1).to(2, 3).drop();
+            assertRows(cluster.coordinator(1).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v2) VALUES (?, 1, 2) IF NOT EXISTS", ConsistencyLevel.ONE, pk),
+                    row(true));
+
+            // finish topology change
+            for (int i = 1 ; i <= 4 ; ++i)
+                cluster.get(i).acceptsOnInstance(Instance::addToRingNormal).accept(cluster.get(4));
+
+            // {3} reads from !{2} => {3, 4}
+            cluster.filters().verbs(PAXOS_PREPARE_REQ.id, READ_REQ.id).from(3).to(2).drop();
+            cluster.filters().verbs(PAXOS_PROPOSE_REQ.id).from(3).to(2).drop();
+            assertRows(cluster.coordinator(3).execute("INSERT INTO " + KEYSPACE + ".tbl (pk, ck, v2) VALUES (?, 1, 2) IF NOT EXISTS", ConsistencyLevel.ONE, pk),
+                    row(false, 5, 1, null, 2));
+        }
+    }
+
+    private static int pk(Cluster cluster, int lb, int ub)
+    {
+        return pk(cluster.get(lb), cluster.get(ub));
+    }
+
+    private static int pk(IInstance lb, IInstance ub)
+    {
+        return pk(Murmur3Partitioner.instance.getTokenFactory().fromString(lb.config().getString("initial_token")),
+                Murmur3Partitioner.instance.getTokenFactory().fromString(ub.config().getString("initial_token")));
+    }
+
+    private static int pk(Token lb, Token ub)
+    {
+        int pk = 0;
+        Token pkt;
+        while (lb.compareTo(pkt = Murmur3Partitioner.instance.getToken(Int32Type.instance.decompose(pk))) >= 0 || ub.compareTo(pkt) < 0)
+            ++pk;
+        return pk;
+    }
+
+    private static void debugOwnership(Cluster cluster, int pk)
+    {
+        for (int i = 1 ; i <= cluster.size() ; ++i)
+            System.out.println(i + ": " + cluster.get(i).appliesOnInstance((Integer v) -> StorageService.instance.getNaturalEndpointsWithPort(KEYSPACE, Int32Type.instance.decompose(v)))
+                    .apply(pk));
+    }
+
+    private static void debugPaxosState(Cluster cluster, int pk)
+    {
+        TableId tableId = cluster.get(1).callOnInstance(() -> Keyspace.open(KEYSPACE).getColumnFamilyStore("tbl").metadata.id);
+        for (int i = 1 ; i <= cluster.size() ; ++i)
+            for (Object[] row : cluster.get(i).executeInternal("select in_progress_ballot, proposal_ballot, most_recent_commit_at from system.paxos where row_key = ? and cf_id = ?", Int32Type.instance.decompose(pk), tableId))
+                System.out.println(i + ": " + (row[0] == null ? 0L : UUIDGen.microsTimestamp((UUID)row[0])) + ", " + (row[1] == null ? 0L : UUIDGen.microsTimestamp((UUID)row[1])) + ", " + (row[2] == null ? 0L : UUIDGen.microsTimestamp((UUID)row[2])));
+    }
+
+}