use correct list of replicas for LOCAL_QUORUM reads when read repair is disabled

patch by jbellis; reviewed by Vijay for CASSANDRA-3696

CHANGES.txt

@@ -22,6 +22,9 @@ Merged from 0.8:
  * prevent new nodes from thinking down nodes are up forever (CASSANDRA-3626)
  * Flush non-cfs backed secondary indexes (CASSANDRA-3659)
  * Secondary Indexes should report memory consumption (CASSANDRA-3155)
+ * use correct list of replicas for LOCAL_QUORUM reads when read repair
+   is disabled (CASSANDRA-3696)
+
 
 1.0.6
  * (CQL) fix cqlsh support for replicate_on_write (CASSANDRA-3596)

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

@@ -23,6 +23,8 @@ package org.apache.cassandra.service;
 
 import java.net.InetAddress;
 import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
 import java.util.List;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
@@ -42,6 +44,19 @@ public class DatacenterReadCallback<T> extends ReadCallback<T>
 {
     private static final IEndpointSnitch snitch = DatabaseDescriptor.getEndpointSnitch();
     private static final String localdc = snitch.getDatacenter(FBUtilities.getBroadcastAddress());
+    private static final Comparator<InetAddress> localComparator = new Comparator<InetAddress>()
+    {
+        public int compare(InetAddress endpoint1, InetAddress endpoint2)
+        {
+            boolean local1 = localdc.equals(snitch.getDatacenter(endpoint1));
+            boolean local2 = localdc.equals(snitch.getDatacenter(endpoint2));
+            if (local1 && !local2)
+                return -1;
+            if (local2 && !local1)
+                return 1;
+            return 0;
+        }
+    };
 
     public DatacenterReadCallback(IResponseResolver resolver, ConsistencyLevel consistencyLevel, IReadCommand command, List<InetAddress> endpoints)
     {
@@ -49,17 +64,9 @@ public class DatacenterReadCallback<T> extends ReadCallback<T>
     }
 
     @Override
-    protected List<InetAddress> preferredEndpoints(List<InetAddress> endpoints)
+    protected void sortForConsistencyLevel(List<InetAddress> endpoints)
     {
-        ArrayList<InetAddress> preferred = new ArrayList<InetAddress>(blockfor);
-        for (InetAddress endpoint : endpoints)
-        {
-            if (localdc.equals(snitch.getDatacenter(endpoint)))
-                preferred.add(endpoint);
-            if (preferred.size() == blockfor)
-                break;
-        }
-        return preferred;
+        Collections.sort(endpoints, localComparator);
     }
 
     @Override

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

@@ -21,7 +21,6 @@ package org.apache.cassandra.service;
 import java.io.IOException;
 import java.net.InetAddress;
 import java.util.List;
-import java.util.Random;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -69,18 +68,25 @@ public class ReadCallback<T> implements IAsyncCallback
         this.resolver = resolver;
         this.startTime = System.currentTimeMillis();
         boolean repair = randomlyReadRepair();
+        sortForConsistencyLevel(endpoints);
         this.endpoints = repair || resolver instanceof RowRepairResolver
                        ? endpoints
-                       : preferredEndpoints(endpoints);
+                       : endpoints.subList(0, Math.min(endpoints.size(), blockfor));
 
         if (logger.isDebugEnabled())
             logger.debug(String.format("Blockfor/repair is %s/%s; setting up requests to %s",
                                        blockfor, repair, StringUtils.join(this.endpoints, ",")));
     }
 
-    protected List<InetAddress> preferredEndpoints(List<InetAddress> endpoints)
+    /**
+     * Endpoints is already restricted to live replicas, sorted by snitch preference.  This is a hook for
+     * DatacenterReadCallback to move local-DC replicas to the front of the list.  We need this both
+     * when doing read repair (because the first replica gets the data read) and otherwise (because
+     * only the first 1..blockfor replicas will get digest reads).
+     */
+    protected void sortForConsistencyLevel(List<InetAddress> endpoints)
     {
-        return endpoints.subList(0, Math.min(endpoints.size(), blockfor)); // min so as to not throw exception until assureSufficient is called
+        // no-op except in DRC
     }
 
     private boolean randomlyReadRepair()