Allow EACH_QUORUM for reads

patch by Carl Yeksigian; revieiwed by Ariel Weisberg for CASSANDRA-9602

CHANGES.txt

@@ -1,4 +1,5 @@
 3.0-rc2
+ * Allow EACH_QUORUM for reads (CASSANDRA-9602)
  * Fix potential ClassCastException while upgrading (CASSANDRA-10468)
  * Fix NPE in MVs on update (CASSANDRA-10503)
  * Only include modified cell data in indexing deltas (CASSANDRA-10438)

NEWS.txt

@@ -18,6 +18,7 @@ using the provided 'sstableupgrade' tool.
 
 New features
 ------------
+   - EACH_QUORUM is now a supported consistency level for read requests.
    - Support for IN restrictions on any partition key component or clustering key
      as well as support for EQ and IN multicolumn restrictions has been added to
      UPDATE and DELETE statement.

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

@@ -182,6 +182,11 @@ public enum ConsistencyLevel
 
     public List<InetAddress> filterForQuery(Keyspace keyspace, List<InetAddress> liveEndpoints, ReadRepairDecision readRepair)
     {
+        // If we are doing an each quorum, we have to make sure that the endpoints we select provide a quorum for each
+        // data center
+        if (this == EACH_QUORUM)
+            return filterForEachQuorum(keyspace, liveEndpoints, readRepair);
+
         /*
          * Endpoints are expected to be restricted to live replicas, sorted by snitch preference.
          * For LOCAL_QUORUM, move local-DC replicas in front first as we need them there whether
@@ -217,6 +222,37 @@ public enum ConsistencyLevel
         }
     }
 
+    private List<InetAddress> filterForEachQuorum(Keyspace keyspace, List<InetAddress> liveEndpoints, ReadRepairDecision readRepair)
+    {
+        NetworkTopologyStrategy strategy = (NetworkTopologyStrategy) keyspace.getReplicationStrategy();
+
+        // quickly drop out if read repair is GLOBAL, since we just use all of the live endpoints
+        if (readRepair == ReadRepairDecision.GLOBAL)
+            return liveEndpoints;
+
+        Map<String, List<InetAddress>> dcsEndpoints = new HashMap<>();
+        for (String dc: strategy.getDatacenters())
+            dcsEndpoints.put(dc, new ArrayList<>());
+
+        for (InetAddress add : liveEndpoints)
+        {
+            String dc = DatabaseDescriptor.getEndpointSnitch().getDatacenter(add);
+            dcsEndpoints.get(dc).add(add);
+        }
+
+        List<InetAddress> waitSet = new ArrayList<>();
+        for (Map.Entry<String, List<InetAddress>> dcEndpoints : dcsEndpoints.entrySet())
+        {
+            List<InetAddress> dcEndpoint = dcEndpoints.getValue();
+            if (readRepair == ReadRepairDecision.DC_LOCAL && dcEndpoints.getKey().equals(DatabaseDescriptor.getLocalDataCenter()))
+                waitSet.addAll(dcEndpoint);
+            else
+                waitSet.addAll(dcEndpoint.subList(0, Math.min(localQuorumFor(keyspace, dcEndpoints.getKey()), dcEndpoint.size())));
+        }
+
+        return waitSet;
+    }
+
     public boolean isSufficientLiveNodes(Keyspace keyspace, Iterable<InetAddress> liveEndpoints)
     {
         switch (this)
@@ -282,7 +318,7 @@ public enum ConsistencyLevel
                         int dcBlockFor = localQuorumFor(keyspace, entry.getKey());
                         int dcLive = entry.getValue();
                         if (dcLive < dcBlockFor)
-                            throw new UnavailableException(this, dcBlockFor, dcLive);
+                            throw new UnavailableException(this, entry.getKey(), dcBlockFor, dcLive);
                     }
                     break;
                 }
@@ -304,8 +340,6 @@ public enum ConsistencyLevel
         {
             case ANY:
                 throw new InvalidRequestException("ANY ConsistencyLevel is only supported for writes");
-            case EACH_QUORUM:
-                throw new InvalidRequestException("EACH_QUORUM ConsistencyLevel is only supported for writes");
         }
     }
 

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

@@ -30,6 +30,11 @@ public class UnavailableException extends RequestExecutionException
         this("Cannot achieve consistency level " + consistency, consistency, required, alive);
     }
 
+    public UnavailableException(ConsistencyLevel consistency, String dc, int required, int alive)
+    {
+        this("Cannot achieve consistency level " + consistency + " in DC " + dc, consistency, required, alive);
+    }
+
     public UnavailableException(String msg, ConsistencyLevel consistency, int required, int alive)
     {
         super(ExceptionCode.UNAVAILABLE, msg);