From d267cf77bd9f87389997d0c4b990aa799bc6cfda Mon Sep 17 00:00:00 2001
From: Jonathan Ellis <jbellis@apache.org>
Date: Wed, 9 Dec 2009 05:49:03 +0000
Subject: [PATCH] support ConsistencyLevel.ALL on read.  patch by jbellis;
 reviewed by goffinet for CASSANDRA-584

git-svn-id: https://svn.apache.org/repos/asf/incubator/cassandra/trunk@888707 13f79535-47bb-0310-9956-ffa450edef68
---
 CHANGES.txt                                     |  1 +
 .../apache/cassandra/service/StorageProxy.java  | 17 +++++++----------
 2 files changed, 8 insertions(+), 10 deletions(-)

diff --git a/CHANGES.txt b/CHANGES.txt
index a90baf64bb..5c7902cea5 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -19,6 +19,7 @@
  * increase failure conviction threshold, resulting in less nodes
    incorrectly (and temporarily) marked as down (CASSANDRA-610)
  * respect memtable thresholds during log replay (CASSANDRA-609)
+ * support ConsistencyLevel.ALL on read (CASSANDRA-584)
 
 
 0.5.0 beta
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index ae02daf17d..ce2a93d953 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -407,6 +407,7 @@ public class StorageProxy implements StorageProxyMBean
         List<InetAddress[]> commandEndPoints = new ArrayList<InetAddress[]>();
         List<Row> rows = new ArrayList<Row>();
 
+        int responseCount = determineBlockFor(DatabaseDescriptor.getReplicationFactor(), DatabaseDescriptor.getReplicationFactor(), consistency_level);
         int commandIndex = 0;
 
         for (ReadCommand command: commands)
@@ -419,28 +420,24 @@ public class StorageProxy implements StorageProxyMBean
             Message messageDigestOnly = readMessageDigestOnly.makeReadMessage();
 
             InetAddress dataPoint = StorageService.instance().findSuitableEndPoint(command.key);
-            List<InetAddress> endpointList = StorageService.instance().getNaturalEndpoints(command.key);
+            List<InetAddress> endpointList = StorageService.instance().getLiveNaturalEndpoints(command.key);
+            if (endpointList.size() < responseCount)
+                throw new UnavailableException();
 
             InetAddress[] endPoints = new InetAddress[endpointList.size()];
             Message messages[] = new Message[endpointList.size()];
-            /*
-             * data-request message is sent to dataPoint, the node that will actually get
-             * the data for us. The other replicas are only sent a digest query.
-            */
+            // data-request message is sent to dataPoint, the node that will actually get
+            // the data for us. The other replicas are only sent a digest query.
             int n = 0;
             for (InetAddress endpoint : endpointList)
             {
-                if (!FailureDetector.instance().isAlive(endpoint))
-                    continue;
                 Message m = endpoint.equals(dataPoint) ? message : messageDigestOnly;
                 endPoints[n] = endpoint;
                 messages[n++] = m;
                 if (logger.isDebugEnabled())
                     logger.debug("strongread reading " + (m == message ? "data" : "digest") + " for " + command + " from " + m.getMessageId() + "@" + endpoint);
             }
-            if (n < DatabaseDescriptor.getQuorum())
-                throw new UnavailableException();
-            QuorumResponseHandler<Row> quorumResponseHandler = new QuorumResponseHandler<Row>(DatabaseDescriptor.getQuorum(), new ReadResponseResolver(command.table, DatabaseDescriptor.getQuorum()));
+            QuorumResponseHandler<Row> quorumResponseHandler = new QuorumResponseHandler<Row>(DatabaseDescriptor.getQuorum(), new ReadResponseResolver(command.table, responseCount));
             MessagingService.instance().sendRR(messages, endPoints, quorumResponseHandler);
             quorumResponseHandlers.add(quorumResponseHandler);
             commandEndPoints.add(endPoints);