CASSANDRA-20798: Fix non-deterministic reads in Transient Replication

Author: maoling

doc/modules/cassandra/pages/managing/operating/transientreplication.adoc

@@ -45,10 +45,11 @@ can be discarded.
 
 Transient replication is not enabled by default. Transient replication
 must be enabled on each node in a cluster separately by setting the
-following configuration property in `cassandra.yaml`.
+following configuration property in `cassandra.yaml`. Transient nodes are not allowed with multiple tokens
 
 ....
 transient_replication_enabled: true
+num_tokens: 1
 ....
 
 Transient replication may be configured with both `SimpleStrategy` and
@@ -59,7 +60,7 @@ As an example, create a keyspace with replication factor (RF) 3.
 
 ....
 CREATE KEYSPACE CassandraKeyspaceSimple WITH replication = {'class': 'SimpleStrategy',
-'replication_factor' : 4/1};
+'replication_factor' : '4/1'};
 ....
 
 As another example, `some_keysopace keyspace` will have 3 replicas in
@@ -68,12 +69,21 @@ transient:
 
 ....
 CREATE KEYSPACE some_keysopace WITH replication = {'class': 'NetworkTopologyStrategy',
-'DC1' : '3/1'', 'DC2' : '5/2'};
+'DC1' : '3/1', 'DC2' : '5/2'};
 ....
 
 Transiently replicated keyspaces only support tables with `read_repair`
 set to `NONE`.
 
+....
+CREATE TABLE users (
+    user_id varchar PRIMARY KEY,
+    first varchar,
+    last varchar,
+    age int
+) WITH read_repair = 'NONE';
+....
+
 Important Restrictions:
 
 * RF cannot be altered while some endpoints are not in a normal state
@@ -124,7 +134,7 @@ quorum write will be upgraded to include transient replicas.
 
 Pending ranges refers to the movement of token ranges between transient
 replicas. When a transient range is moved, there will be a period of
-time where both transient replicas would need to receive any write
+time when both transient replicas would need to receive any write
 intended for the logical transient replica so that after the movement
 takes effect a read quorum is able to return a response. Nodes are _not_
 temporarily transient replicas during expansion. They stream data like a

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

@@ -36,15 +36,15 @@ public static UnavailableException create(ConsistencyLevel consistency, int requ
         if (required > alive)
             return new UnavailableException(String.format("Cannot achieve consistency level %s. Required %s but only %s alive.", consistency, required, alive),
                                             consistency, required, alive);
-        assert requiredFull < aliveFull;
+        assert aliveFull < requiredFull;
         return new UnavailableException("Insufficient full replicas", consistency, required, alive);
     }
 
     public static UnavailableException create(ConsistencyLevel consistency, String dc, int required, int requiredFull, int alive, int aliveFull)
     {
         if (required > alive)
             return new UnavailableException("Cannot achieve consistency level " + consistency + " in DC " + dc, consistency, required, alive);
-        assert requiredFull < aliveFull;
+        assert aliveFull < requiredFull;
         return new UnavailableException("Insufficient full replicas in DC " + dc, consistency, required, alive);
     }
 

src/java/org/apache/cassandra/locator/ReplicaPlans.java

@@ -186,7 +186,7 @@ static void assureSufficientLiveReplicas(Locator locator, AbstractReplicationStr
                         total += dcCount.allReplicas();
                     }
                     if (totalFull < blockForFullReplicas)
-                        throw UnavailableException.create(consistencyLevel, blockFor, total, blockForFullReplicas, totalFull);
+                        throw UnavailableException.create(consistencyLevel, blockFor, blockForFullReplicas, total, totalFull);
                     break;
                 }
                 // Fallthough on purpose for SimpleStrategy
@@ -812,7 +812,8 @@ private static <E extends Endpoints<E>> E contactForEachQuorumRead(Locator locat
         });
     }
 
-    private static <E extends Endpoints<E>> E contactForRead(Locator locator, AbstractReplicationStrategy replicationStrategy, ConsistencyLevel consistencyLevel, boolean alwaysSpeculate, E candidates)
+    @VisibleForTesting
+    public static <E extends Endpoints<E>> E contactForRead(Locator locator, AbstractReplicationStrategy replicationStrategy, ConsistencyLevel consistencyLevel, boolean alwaysSpeculate, E candidates)
     {
         /*
          * If we are doing an each quorum query, we have to make sure that the endpoints we select
@@ -823,13 +824,45 @@ private static <E extends Endpoints<E>> E contactForRead(Locator locator, Abstra
          *
          * TODO: this is still very inconistently managed between {LOCAL,EACH}_QUORUM and other consistency levels - should address this in a follow-up
          */
+        candidates = reorderWithOneFullReplicaFirst(candidates);
         if (consistencyLevel == EACH_QUORUM && replicationStrategy instanceof NetworkTopologyStrategy)
             return contactForEachQuorumRead(locator, (NetworkTopologyStrategy) replicationStrategy, candidates);
 
         int count = consistencyLevel.blockFor(replicationStrategy) + (alwaysSpeculate ? 1 : 0);
         return candidates.subList(0, Math.min(count, candidates.size()));
     }
 
+    /**
+     * Reorders the provided list of replicas such that, if any transient replicas are present,
+     * a single full replica (if available) is placed first. The remaining replicas retain their original order(order by NodeProximity),
+     *
+     * @param endpoints the original set of replicas
+     * @param <E> a subtype of Endpoints
+     * @return the same set if there are no transient replicas or no full replica is found,
+     *         otherwise a reordered set with the first full replica placed first
+     */
+    private static <E extends Endpoints<E>> E reorderWithOneFullReplicaFirst(E endpoints) {
+        boolean hasTransient = false;
+        Replica firstFull = null;
+
+        for (Replica r : endpoints) {
+            if (r.isTransient()) hasTransient = true;
+            if (firstFull == null && r.isFull()) firstFull = r;
+        }
+
+        if (!hasTransient || firstFull == null)
+            return endpoints;
+
+        ReplicaCollection.Builder<E> builder = endpoints.newBuilder(endpoints.size());
+        builder.add(firstFull);
+        for (Replica r : endpoints) {
+            if (!r.equals(firstFull)) {
+                builder.add(r);
+            }
+        }
+
+        return builder.build();
+    }
 
     /**
      * Construct a plan for reading from a single node - this permits no speculation or read-repair