cassandra-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bdeggles...@apache.org
Subject cassandra git commit: Improve read repair blocking behavior
Date Tue, 21 Aug 2018 18:04:28 GMT
Repository: cassandra
Updated Branches:
  refs/heads/trunk 994da255c -> 644676b08


Improve read repair blocking behavior

Patch by Blake Eggleston; reviewed by Marcus Eriksson and Alex Petrov
for CASSANDRA-10726


Project: http://git-wip-us.apache.org/repos/asf/cassandra/repo
Commit: http://git-wip-us.apache.org/repos/asf/cassandra/commit/644676b0
Tree: http://git-wip-us.apache.org/repos/asf/cassandra/tree/644676b0
Diff: http://git-wip-us.apache.org/repos/asf/cassandra/diff/644676b0

Branch: refs/heads/trunk
Commit: 644676b088be5177ef1d0cdaf450306ea28d8a12
Parents: 994da25
Author: Blake Eggleston <bdeggleston@gmail.com>
Authored: Mon May 14 14:24:03 2018 -0700
Committer: Blake Eggleston <bdeggleston@gmail.com>
Committed: Tue Aug 21 11:01:10 2018 -0700

----------------------------------------------------------------------
 CHANGES.txt                                     |   1 +
 .../apache/cassandra/db/ConsistencyLevel.java   |  10 +-
 .../cassandra/db/PartitionRangeReadCommand.java |   1 +
 .../org/apache/cassandra/db/ReadCommand.java    |   1 +
 .../db/SinglePartitionReadCommand.java          |   1 +
 .../cassandra/metrics/ReadRepairMetrics.java    |   5 +
 .../apache/cassandra/net/MessagingService.java  |   1 +
 .../apache/cassandra/service/StorageProxy.java  |  65 +++-
 .../service/reads/AbstractReadExecutor.java     |  40 +-
 .../reads/repair/BlockingPartitionRepair.java   | 243 +++++++++++++
 .../reads/repair/BlockingReadRepair.java        | 229 ++++++------
 .../reads/repair/BlockingReadRepairs.java       | 114 ++++++
 .../service/reads/repair/NoopReadRepair.java    |  29 +-
 .../repair/PartitionIteratorMergeListener.java  |  13 +-
 .../service/reads/repair/ReadRepair.java        |  44 ++-
 .../service/reads/repair/RepairListener.java    |  34 --
 .../reads/repair/RowIteratorMergeListener.java  |  31 +-
 .../service/reads/DataResolverTest.java         |  18 +-
 .../service/reads/repair/ReadRepairTest.java    | 361 +++++++++++++++++++
 .../reads/repair/TestableReadRepair.java        |  41 ++-
 20 files changed, 1046 insertions(+), 236 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 9fbaf25..b34979a 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 4.0
+ * Improve read repair blocking behavior (CASSANDRA-10726)
  * Add a virtual table to expose settings (CASSANDRA-14573)
  * Fix up chunk cache handling of metrics (CASSANDRA-14628)
  * Extend IAuthenticator to accept peer SSL certificates (CASSANDRA-14652)

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/db/ConsistencyLevel.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ConsistencyLevel.java b/src/java/org/apache/cassandra/db/ConsistencyLevel.java
index 8f3a51c..d37da0a 100644
--- a/src/java/org/apache/cassandra/db/ConsistencyLevel.java
+++ b/src/java/org/apache/cassandra/db/ConsistencyLevel.java
@@ -138,12 +138,20 @@ public enum ConsistencyLevel
         }
     }
 
+    /**
+     * Determine if this consistency level meets or exceeds the consistency requirements of the given cl for the given keyspace
+     */
+    public boolean satisfies(ConsistencyLevel other, Keyspace keyspace)
+    {
+        return blockFor(keyspace) >= other.blockFor(keyspace);
+    }
+
     public boolean isDatacenterLocal()
     {
         return isDCLocal;
     }
 
-    public boolean isLocal(InetAddressAndPort endpoint)
+    public static boolean isLocal(InetAddressAndPort endpoint)
     {
         return DatabaseDescriptor.getLocalDataCenter().equals(DatabaseDescriptor.getEndpointSnitch().getDatacenter(endpoint));
     }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
index a6641d4..c312acc 100644
--- a/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
+++ b/src/java/org/apache/cassandra/db/PartitionRangeReadCommand.java
@@ -24,6 +24,7 @@ import java.util.List;
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.collect.Iterables;
 
+import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.filter.*;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/db/ReadCommand.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java
index 82ca054..0262140 100644
--- a/src/java/org/apache/cassandra/db/ReadCommand.java
+++ b/src/java/org/apache/cassandra/db/ReadCommand.java
@@ -34,6 +34,7 @@ import org.apache.cassandra.db.transform.RTBoundCloser;
 import org.apache.cassandra.db.transform.RTBoundValidator;
 import org.apache.cassandra.db.transform.StoppingTransformation;
 import org.apache.cassandra.db.transform.Transformation;
+import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.exceptions.UnknownIndexException;
 import org.apache.cassandra.index.Index;
 import org.apache.cassandra.index.IndexNotAvailableException;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
index c091bf1..1fdb11f 100644
--- a/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
+++ b/src/java/org/apache/cassandra/db/SinglePartitionReadCommand.java
@@ -36,6 +36,7 @@ import org.apache.cassandra.db.lifecycle.*;
 import org.apache.cassandra.db.partitions.*;
 import org.apache.cassandra.db.rows.*;
 import org.apache.cassandra.db.transform.Transformation;
+import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.exceptions.RequestExecutionException;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.sstable.format.SSTableReadsListener;

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java b/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java
index c79fe89..e639be9 100644
--- a/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java
+++ b/src/java/org/apache/cassandra/metrics/ReadRepairMetrics.java
@@ -34,4 +34,9 @@ public class ReadRepairMetrics
     public static final Meter repairedBackground = Metrics.meter(factory.createMetricName("RepairedBackground"));
     @Deprecated
     public static final Meter attempted = Metrics.meter(factory.createMetricName("Attempted"));
+
+    public static final Meter speculatedRead = Metrics.meter(factory.createMetricName("SpeculatedRead"));
+    public static final Meter speculatedWrite = Metrics.meter(factory.createMetricName("SpeculatedWrite"));
+
+    public static void init() {}
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/net/MessagingService.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/net/MessagingService.java b/src/java/org/apache/cassandra/net/MessagingService.java
index b2e72f4..7732673 100644
--- a/src/java/org/apache/cassandra/net/MessagingService.java
+++ b/src/java/org/apache/cassandra/net/MessagingService.java
@@ -125,6 +125,7 @@ public final class MessagingService implements MessagingServiceMBean
     public static final int VERSION_30 = 10;
     public static final int VERSION_3014 = 11;
     public static final int VERSION_40 = 12;
+    public static final int minimum_version = VERSION_30;
     public static final int current_version = VERSION_40;
 
     public static final byte[] ONE_BYTE = new byte[1];

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/service/StorageProxy.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 81e6dae..7fdf591 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -183,6 +183,8 @@ public class StorageProxy implements StorageProxyMBean
             readMetricsMap.put(level, new ClientRequestMetrics("Read-" + level.name()));
             writeMetricsMap.put(level, new ClientWriteRequestMetrics("Write-" + level.name()));
         }
+
+        ReadRepairMetrics.init();
     }
 
     /**
@@ -1766,6 +1768,34 @@ public class StorageProxy implements StorageProxyMBean
         }
     }
 
+    private static PartitionIterator concatAndBlockOnRepair(List<PartitionIterator> iterators, List<ReadRepair> repairs)
+    {
+        PartitionIterator concatenated = PartitionIterators.concat(iterators);
+
+        if (repairs.isEmpty())
+            return concatenated;
+
+        return new PartitionIterator()
+        {
+            public void close()
+            {
+                concatenated.close();
+                repairs.forEach(ReadRepair::maybeSendAdditionalWrites);
+                repairs.forEach(ReadRepair::awaitWrites);
+            }
+
+            public boolean hasNext()
+            {
+                return concatenated.hasNext();
+            }
+
+            public RowIterator next()
+            {
+                return concatenated.next();
+            }
+        };
+    }
+
     /**
      * This function executes local and remote reads, and blocks for the results:
      *
@@ -1784,43 +1814,59 @@ public class StorageProxy implements StorageProxyMBean
 
         AbstractReadExecutor[] reads = new AbstractReadExecutor[cmdCount];
 
+        // Get the replica locations, sorted by response time according to the snitch, and create a read executor
+        // for type of speculation we'll use in this read
         for (int i=0; i<cmdCount; i++)
         {
             reads[i] = AbstractReadExecutor.getReadExecutor(commands.get(i), consistencyLevel, queryStartNanoTime);
         }
 
+        // sends a data request to the closest replica, and a digest request to the others. If we have a speculating
+        // read executoe, we'll only send read requests to enough replicas to satisfy the consistency level
         for (int i=0; i<cmdCount; i++)
         {
             reads[i].executeAsync();
         }
 
+        // if we have a speculating read executor and it looks like we may not receive a response from the initial
+        // set of replicas we sent messages to, speculatively send an additional messages to an un-contacted replica
         for (int i=0; i<cmdCount; i++)
         {
             reads[i].maybeTryAdditionalReplicas();
         }
 
+        // wait for enough responses to meet the consistency level. If there's a digest mismatch, begin the read
+        // repair process by sending full data reads to all replicas we received responses from.
         for (int i=0; i<cmdCount; i++)
         {
             reads[i].awaitResponses();
         }
 
+        // read repair - if it looks like we may not receive enough full data responses to meet CL, send
+        // an additional request to any remaining replicas we haven't contacted (if there are any)
         for (int i=0; i<cmdCount; i++)
         {
-            reads[i].maybeRepairAdditionalReplicas();
+            reads[i].maybeSendAdditionalDataRequests();
         }
 
+        // read repair - block on full data responses
         for (int i=0; i<cmdCount; i++)
         {
             reads[i].awaitReadRepair();
         }
 
+        // if we didn't do a read repair, return the contents of the data response, if we did do a read
+        // repair, merge the full data reads
         List<PartitionIterator> results = new ArrayList<>(cmdCount);
+        List<ReadRepair> repairs = new ArrayList<>(cmdCount);
         for (int i=0; i<cmdCount; i++)
         {
             results.add(reads[i].getResult());
+            repairs.add(reads[i].getReadRepair());
         }
 
-        return PartitionIterators.concat(results);
+        // if we did a read repair, assemble repair mutation and block on them
+        return concatAndBlockOnRepair(results, repairs);
     }
 
     public static class LocalReadRunnable extends DroppableRunnable
@@ -2029,12 +2075,14 @@ public class StorageProxy implements StorageProxyMBean
     {
         private final DataResolver resolver;
         private final ReadCallback handler;
+        private final ReadRepair readRepair;
         private PartitionIterator result;
 
-        private SingleRangeResponse(DataResolver resolver, ReadCallback handler)
+        private SingleRangeResponse(DataResolver resolver, ReadCallback handler, ReadRepair readRepair)
         {
             this.resolver = resolver;
             this.handler = handler;
+            this.readRepair = readRepair;
         }
 
         private void waitForResponse() throws ReadTimeoutException
@@ -2165,7 +2213,7 @@ public class StorageProxy implements StorageProxyMBean
         {
             PartitionRangeReadCommand rangeCommand = command.forSubRange(toQuery.range, isFirst);
 
-            ReadRepair readRepair = ReadRepair.create(command, toQuery.filteredEndpoints, queryStartNanoTime, consistency);
+            ReadRepair readRepair = ReadRepair.create(command, queryStartNanoTime, consistency);
             DataResolver resolver = new DataResolver(keyspace, rangeCommand, consistency, toQuery.filteredEndpoints.size(), queryStartNanoTime, readRepair);
 
             int blockFor = consistency.blockFor(keyspace);
@@ -2188,15 +2236,18 @@ public class StorageProxy implements StorageProxyMBean
                 }
             }
 
-            return new SingleRangeResponse(resolver, handler);
+            return new SingleRangeResponse(resolver, handler, readRepair);
         }
 
         private PartitionIterator sendNextRequests()
         {
             List<PartitionIterator> concurrentQueries = new ArrayList<>(concurrencyFactor);
+            List<ReadRepair> readRepairs = new ArrayList<>(concurrencyFactor);
             for (int i = 0; i < concurrencyFactor && ranges.hasNext(); i++)
             {
-                concurrentQueries.add(query(ranges.next(), i == 0));
+                SingleRangeResponse response = query(ranges.next(), i == 0);
+                concurrentQueries.add(response);
+                readRepairs.add(response.readRepair);
                 ++rangesQueried;
             }
 
@@ -2204,7 +2255,7 @@ public class StorageProxy implements StorageProxyMBean
             // We want to count the results for the sake of updating the concurrency factor (see updateConcurrencyFactor) but we don't want to
             // enforce any particular limit at this point (this could break code than rely on postReconciliationProcessing), hence the DataLimits.NONE.
             counter = DataLimits.NONE.newCounter(command.nowInSec(), true, command.selectsFullPartition(), enforceStrictLiveness);
-            return counter.applyTo(PartitionIterators.concat(concurrentQueries));
+            return counter.applyTo(concatAndBlockOnRepair(concurrentQueries, readRepairs));
         }
 
         public void close()

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java
index e531e0c..61b9948 100644
--- a/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java
+++ b/src/java/org/apache/cassandra/service/reads/AbstractReadExecutor.java
@@ -69,17 +69,23 @@ public abstract class AbstractReadExecutor
     protected final long queryStartNanoTime;
     protected volatile PartitionIterator result = null;
 
+    protected final Keyspace keyspace;
+    protected final int blockFor;
+
     AbstractReadExecutor(Keyspace keyspace, ColumnFamilyStore cfs, ReadCommand command, ConsistencyLevel consistency, List<InetAddressAndPort> targetReplicas, long queryStartNanoTime)
     {
         this.command = command;
         this.consistency = consistency;
         this.targetReplicas = targetReplicas;
-        this.readRepair = ReadRepair.create(command, targetReplicas, queryStartNanoTime, consistency);
+        this.readRepair = ReadRepair.create(command, queryStartNanoTime, consistency);
         this.digestResolver = new DigestResolver(keyspace, command, consistency, readRepair, targetReplicas.size());
         this.handler = new ReadCallback(digestResolver, consistency, command, targetReplicas, queryStartNanoTime);
         this.cfs = cfs;
         this.traceState = Tracing.instance.get();
         this.queryStartNanoTime = queryStartNanoTime;
+        this.keyspace = keyspace;
+        this.blockFor = consistency.blockFor(keyspace);
+
 
         // Set the digest version (if we request some digests). This is the smallest version amongst all our target replicas since new nodes
         // knows how to produce older digest but the reverse is not true.
@@ -91,16 +97,16 @@ public abstract class AbstractReadExecutor
         command.setDigestVersion(digestVersion);
     }
 
-    private DecoratedKey getKey()
+    public DecoratedKey getKey()
     {
-        if (command instanceof SinglePartitionReadCommand)
-        {
-            return ((SinglePartitionReadCommand) command).partitionKey();
-        }
-        else
-        {
-            return null;
-        }
+        Preconditions.checkState(command instanceof SinglePartitionReadCommand,
+                                 "Can only get keys for SinglePartitionReadCommand");
+        return ((SinglePartitionReadCommand) command).partitionKey();
+    }
+
+    public ReadRepair getReadRepair()
+    {
+        return readRepair;
     }
 
     protected void makeDataRequests(Iterable<InetAddressAndPort> endpoints)
@@ -409,7 +415,7 @@ public abstract class AbstractReadExecutor
     {
         try
         {
-            readRepair.awaitRepair();
+            readRepair.awaitReads();
         }
         catch (ReadTimeoutException e)
         {
@@ -424,9 +430,17 @@ public abstract class AbstractReadExecutor
         }
     }
 
-    public void maybeRepairAdditionalReplicas()
+    boolean isDone()
     {
-        // TODO: this
+        return result != null;
+    }
+
+    public void maybeSendAdditionalDataRequests()
+    {
+        if (isDone())
+            return;
+
+        readRepair.maybeSendAdditionalReads();
     }
 
     public PartitionIterator getResult() throws ReadFailureException, ReadTimeoutException

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/service/reads/repair/BlockingPartitionRepair.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/reads/repair/BlockingPartitionRepair.java b/src/java/org/apache/cassandra/service/reads/repair/BlockingPartitionRepair.java
new file mode 100644
index 0000000..eb402ba
--- /dev/null
+++ b/src/java/org/apache/cassandra/service/reads/repair/BlockingPartitionRepair.java
@@ -0,0 +1,243 @@
+/*
+ * 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.service.reads.repair;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.AbstractFuture;
+
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.partitions.PartitionUpdate;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.metrics.ReadRepairMetrics;
+import org.apache.cassandra.net.IAsyncCallback;
+import org.apache.cassandra.net.MessageIn;
+import org.apache.cassandra.net.MessageOut;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.schema.TableId;
+import org.apache.cassandra.tracing.Tracing;
+
+public class BlockingPartitionRepair extends AbstractFuture<Object> implements IAsyncCallback<Object>
+{
+    private final Keyspace keyspace;
+    private final DecoratedKey key;
+    private final ConsistencyLevel consistency;
+    private final InetAddressAndPort[] participants;
+    private final ConcurrentMap<InetAddressAndPort, Mutation> pendingRepairs;
+    private final CountDownLatch latch;
+
+    private volatile long mutationsSentTime;
+
+    public BlockingPartitionRepair(Keyspace keyspace, DecoratedKey key, ConsistencyLevel consistency, Map<InetAddressAndPort, Mutation> repairs, int maxBlockFor, InetAddressAndPort[] participants)
+    {
+        this.keyspace = keyspace;
+        this.key = key;
+        this.consistency = consistency;
+        this.pendingRepairs = new ConcurrentHashMap<>(repairs);
+        this.participants = participants;
+
+        // here we remove empty repair mutations from the block for total, since
+        // we're not sending them mutations
+        int blockFor = maxBlockFor;
+        for (InetAddressAndPort participant: participants)
+        {
+            // remote dcs can sometimes get involved in dc-local reads. We want to repair
+            // them if they do, but they shouldn't interfere with blocking the client read.
+            if (!repairs.containsKey(participant) && shouldBlockOn(participant))
+                blockFor--;
+        }
+
+        // there are some cases where logically identical data can return different digests
+        // For read repair, this would result in ReadRepairHandler being called with a map of
+        // empty mutations. If we'd also speculated on either of the read stages, the number
+        // of empty mutations would be greater than blockFor, causing the latch ctor to throw
+        // an illegal argument exception due to a negative start value. So here we clamp it 0
+        latch = new CountDownLatch(Math.max(blockFor, 0));
+    }
+
+    @VisibleForTesting
+    long waitingOn()
+    {
+        return latch.getCount();
+    }
+
+    @VisibleForTesting
+    boolean isLocal(InetAddressAndPort endpoint)
+    {
+        return ConsistencyLevel.isLocal(endpoint);
+    }
+
+    private boolean shouldBlockOn(InetAddressAndPort endpoint)
+    {
+        return !consistency.isDatacenterLocal() || isLocal(endpoint);
+    }
+
+    @VisibleForTesting
+    void ack(InetAddressAndPort from)
+    {
+        if (shouldBlockOn(from))
+        {
+            pendingRepairs.remove(from);
+            latch.countDown();
+        }
+    }
+
+    @Override
+    public void response(MessageIn<Object> msg)
+    {
+        ack(msg.from);
+    }
+
+    @Override
+    public boolean isLatencyForSnitch()
+    {
+        return false;
+    }
+
+    private static PartitionUpdate extractUpdate(Mutation mutation)
+    {
+        return Iterables.getOnlyElement(mutation.getPartitionUpdates());
+    }
+
+    /**
+     * Combine the contents of any unacked repair into a single update
+     */
+    private PartitionUpdate mergeUnackedUpdates()
+    {
+        // recombinate the updates
+        List<PartitionUpdate> updates = Lists.newArrayList(Iterables.transform(pendingRepairs.values(), BlockingPartitionRepair::extractUpdate));
+        return updates.isEmpty() ? null : PartitionUpdate.merge(updates);
+    }
+
+    @VisibleForTesting
+    protected void sendRR(MessageOut<Mutation> message, InetAddressAndPort endpoint)
+    {
+        MessagingService.instance().sendRR(message, endpoint, this);
+    }
+
+    public void sendInitialRepairs()
+    {
+        mutationsSentTime = System.nanoTime();
+        for (Map.Entry<InetAddressAndPort, Mutation> entry: pendingRepairs.entrySet())
+        {
+            InetAddressAndPort destination = entry.getKey();
+            Mutation mutation = entry.getValue();
+            TableId tableId = extractUpdate(mutation).metadata().id;
+
+            Tracing.trace("Sending read-repair-mutation to {}", destination);
+            // use a separate verb here to avoid writing hints on timeouts
+            sendRR(mutation.createMessage(MessagingService.Verb.READ_REPAIR), destination);
+            ColumnFamilyStore.metricsFor(tableId).readRepairRequests.mark();
+
+            if (!shouldBlockOn(destination))
+                pendingRepairs.remove(destination);
+        }
+    }
+
+    public boolean awaitRepairs(long timeout, TimeUnit timeoutUnit)
+    {
+        long elapsed = System.nanoTime() - mutationsSentTime;
+        long remaining = timeoutUnit.toNanos(timeout) - elapsed;
+
+        try
+        {
+            return latch.await(remaining, TimeUnit.NANOSECONDS);
+        }
+        catch (InterruptedException e)
+        {
+            throw new AssertionError(e);
+        }
+    }
+
+    private static int msgVersionIdx(int version)
+    {
+        return version - MessagingService.minimum_version;
+    }
+
+    /**
+     * If it looks like we might not receive acks for all the repair mutations we sent out, combine all
+     * the unacked mutations and send them to the minority of nodes not involved in the read repair data
+     * read / write cycle. We will accept acks from them in lieu of acks from the initial mutations sent
+     * out, so long as we receive the same number of acks as repair mutations transmitted. This prevents
+     * misbehaving nodes from killing a quorum read, while continuing to guarantee monotonic quorum reads
+     */
+    public void maybeSendAdditionalWrites(long timeout, TimeUnit timeoutUnit)
+    {
+        if (awaitRepairs(timeout, timeoutUnit))
+            return;
+
+        Set<InetAddressAndPort> exclude = Sets.newHashSet(participants);
+        Iterable<InetAddressAndPort> candidates = Iterables.filter(getCandidateEndpoints(), e -> !exclude.contains(e));
+        if (Iterables.isEmpty(candidates))
+            return;
+
+        PartitionUpdate update = mergeUnackedUpdates();
+        if (update == null)
+            // final response was received between speculate
+            // timeout and call to get unacked mutation.
+            return;
+
+        ReadRepairMetrics.speculatedWrite.mark();
+
+        Mutation[] versionedMutations = new Mutation[msgVersionIdx(MessagingService.current_version) + 1];
+
+        for (InetAddressAndPort endpoint: candidates)
+        {
+            int versionIdx = msgVersionIdx(MessagingService.instance().getVersion(endpoint));
+
+            Mutation mutation = versionedMutations[versionIdx];
+
+            if (mutation == null)
+            {
+                mutation = BlockingReadRepairs.createRepairMutation(update, consistency, endpoint, true);
+                versionedMutations[versionIdx] = mutation;
+            }
+
+            if (mutation == null)
+            {
+                // the mutation is too large to send.
+                continue;
+            }
+
+            Tracing.trace("Sending speculative read-repair-mutation to {}", endpoint);
+            sendRR(mutation.createMessage(MessagingService.Verb.READ_REPAIR), endpoint);
+        }
+    }
+
+    @VisibleForTesting
+    protected Iterable<InetAddressAndPort> getCandidateEndpoints()
+    {
+        return BlockingReadRepairs.getCandidateEndpoints(keyspace, key.getToken(), consistency);
+    }
+
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java b/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java
index cd4d2a7..c5f1bea 100644
--- a/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java
+++ b/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepair.java
@@ -18,20 +18,17 @@
 
 package org.apache.cassandra.service.reads.repair;
 
-import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.Queue;
+import java.util.Set;
 import java.util.concurrent.ConcurrentLinkedQueue;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import java.util.function.Consumer;
 
-import javax.annotation.Nullable;
-
-import com.google.common.util.concurrent.AbstractFuture;
-import com.google.common.util.concurrent.FutureCallback;
-import com.google.common.util.concurrent.Futures;
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Sets;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -43,13 +40,15 @@ import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.Keyspace;
 import org.apache.cassandra.db.Mutation;
 import org.apache.cassandra.db.ReadCommand;
+import org.apache.cassandra.db.SinglePartitionReadCommand;
 import org.apache.cassandra.db.partitions.PartitionIterator;
 import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
+import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.exceptions.ReadTimeoutException;
+import org.apache.cassandra.locator.AbstractReplicationStrategy;
 import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.NetworkTopologyStrategy;
 import org.apache.cassandra.metrics.ReadRepairMetrics;
-import org.apache.cassandra.net.AsyncOneResponse;
-import org.apache.cassandra.net.MessageOut;
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.service.reads.DataResolver;
 import org.apache.cassandra.service.reads.DigestResolver;
@@ -60,17 +59,14 @@ import org.apache.cassandra.tracing.Tracing;
  * 'Classic' read repair. Doesn't allow the client read to return until
  *  updates have been written to nodes needing correction.
  */
-public class BlockingReadRepair implements ReadRepair, RepairListener
+public class BlockingReadRepair implements ReadRepair
 {
     private static final Logger logger = LoggerFactory.getLogger(BlockingReadRepair.class);
 
-    private static final boolean DROP_OVERSIZED_READ_REPAIR_MUTATIONS =
-        Boolean.getBoolean("cassandra.drop_oversized_readrepair_mutations");
-
     private final ReadCommand command;
-    private final List<InetAddressAndPort> endpoints;
     private final long queryStartNanoTime;
     private final ConsistencyLevel consistency;
+    private final ColumnFamilyStore cfs;
 
     private final Queue<BlockingPartitionRepair> repairs = new ConcurrentLinkedQueue<>();
 
@@ -81,170 +77,149 @@ public class BlockingReadRepair implements ReadRepair, RepairListener
         private final DataResolver dataResolver;
         private final ReadCallback readCallback;
         private final Consumer<PartitionIterator> resultConsumer;
+        private final List<InetAddressAndPort> initialContacts;
 
-        public DigestRepair(DataResolver dataResolver, ReadCallback readCallback, Consumer<PartitionIterator> resultConsumer)
+        public DigestRepair(DataResolver dataResolver, ReadCallback readCallback, Consumer<PartitionIterator> resultConsumer, List<InetAddressAndPort> initialContacts)
         {
             this.dataResolver = dataResolver;
             this.readCallback = readCallback;
             this.resultConsumer = resultConsumer;
+            this.initialContacts = initialContacts;
         }
     }
 
     public BlockingReadRepair(ReadCommand command,
-                              List<InetAddressAndPort> endpoints,
                               long queryStartNanoTime,
                               ConsistencyLevel consistency)
     {
         this.command = command;
-        this.endpoints = endpoints;
         this.queryStartNanoTime = queryStartNanoTime;
         this.consistency = consistency;
+        this.cfs = Keyspace.openAndGetStore(command.metadata());
     }
 
     public UnfilteredPartitionIterators.MergeListener getMergeListener(InetAddressAndPort[] endpoints)
     {
-        return new PartitionIteratorMergeListener(endpoints, command, this);
+        return new PartitionIteratorMergeListener(endpoints, command, consistency, this);
+    }
+
+    private int getMaxResponses()
+    {
+        AbstractReplicationStrategy strategy = cfs.keyspace.getReplicationStrategy();
+        if (consistency.isDatacenterLocal() && strategy instanceof NetworkTopologyStrategy)
+        {
+            NetworkTopologyStrategy nts = (NetworkTopologyStrategy) strategy;
+            return nts.getReplicationFactor(DatabaseDescriptor.getLocalDataCenter());
+        }
+        else
+        {
+            return strategy.getReplicationFactor();
+        }
     }
 
-    public static class BlockingPartitionRepair extends AbstractFuture<Object> implements RepairListener.PartitionRepair
+    // digestResolver isn't used here because we resend read requests to all participants
+    public void startRepair(DigestResolver digestResolver, List<InetAddressAndPort> allEndpoints, List<InetAddressAndPort> contactedEndpoints, Consumer<PartitionIterator> resultConsumer)
     {
+        ReadRepairMetrics.repairedBlocking.mark();
+
+        // Do a full data read to resolve the correct response (and repair node that need be)
+        Keyspace keyspace = Keyspace.open(command.metadata().keyspace);
+        DataResolver resolver = new DataResolver(keyspace, command, ConsistencyLevel.ALL, getMaxResponses(), queryStartNanoTime, this);
+        ReadCallback readCallback = new ReadCallback(resolver, ConsistencyLevel.ALL, consistency.blockFor(cfs.keyspace), command,
+                                                     keyspace, allEndpoints, queryStartNanoTime);
 
-        final List<AsyncOneResponse<?>> responses;
-        final ReadCommand command;
-        final ConsistencyLevel consistency;
+        digestRepair = new DigestRepair(resolver, readCallback, resultConsumer, contactedEndpoints);
 
-        public BlockingPartitionRepair(int expectedResponses, ReadCommand command, ConsistencyLevel consistency)
+        for (InetAddressAndPort endpoint : contactedEndpoints)
         {
-            this.responses = new ArrayList<>(expectedResponses);
-            this.command = command;
-            this.consistency = consistency;
+            Tracing.trace("Enqueuing full data read to {}", endpoint);
+            MessagingService.instance().sendRRWithFailure(command.createMessage(), endpoint, readCallback);
         }
+    }
 
-        private AsyncOneResponse sendRepairMutation(Mutation mutation, InetAddressAndPort destination)
-        {
-            DecoratedKey key = mutation.key();
-            Keyspace keyspace = Keyspace.open(mutation.getKeyspaceName());
-            int messagingVersion = MessagingService.instance().getVersion(destination);
+    public void awaitReads() throws ReadTimeoutException
+    {
+        DigestRepair repair = digestRepair;
+        if (repair == null)
+            return;
 
-            int    mutationSize = (int) Mutation.serializer.serializedSize(mutation, messagingVersion);
-            int maxMutationSize = DatabaseDescriptor.getMaxMutationSize();
+        repair.readCallback.awaitResults();
+        repair.resultConsumer.accept(digestRepair.dataResolver.resolve());
+    }
 
-            AsyncOneResponse callback = null;
+    private boolean shouldSpeculate()
+    {
+        ConsistencyLevel speculativeCL = consistency.isDatacenterLocal() ? ConsistencyLevel.LOCAL_QUORUM : ConsistencyLevel.QUORUM;
+        return  consistency != ConsistencyLevel.EACH_QUORUM
+               && consistency.satisfies(speculativeCL, cfs.keyspace)
+               && cfs.sampleLatencyNanos <= TimeUnit.MILLISECONDS.toNanos(command.getTimeout());
+    }
 
-            if (mutationSize <= maxMutationSize)
-            {
-                Tracing.trace("Sending read-repair-mutation to {}", destination);
-                // use a separate verb here to avoid writing hints on timeouts
-                MessageOut<Mutation> message = mutation.createMessage(MessagingService.Verb.READ_REPAIR);
-                callback = MessagingService.instance().sendRR(message, destination);
-                ColumnFamilyStore.metricsFor(command.metadata().id).readRepairRequests.mark();
-            }
-            else if (DROP_OVERSIZED_READ_REPAIR_MUTATIONS)
-            {
-                logger.debug("Encountered an oversized ({}/{}) read repair mutation for table {}, key {}, node {}",
-                             mutationSize,
-                             maxMutationSize,
-                             command.metadata(),
-                             command.metadata().partitionKeyType.getString(key.getKey()),
-                             destination);
-            }
-            else
-            {
-                logger.warn("Encountered an oversized ({}/{}) read repair mutation for table {}, key {}, node {}",
-                            mutationSize,
-                            maxMutationSize,
-                            command.metadata(),
-                            command.metadata().partitionKeyType.getString(key.getKey()),
-                            destination);
-
-                int blockFor = consistency.blockFor(keyspace);
-                Tracing.trace("Timed out while read-repairing after receiving all {} data and digest responses", blockFor);
-                throw new ReadTimeoutException(consistency, blockFor - 1, blockFor, true);
-            }
-            return callback;
-        }
+    public void maybeSendAdditionalReads()
+    {
+        Preconditions.checkState(command instanceof SinglePartitionReadCommand,
+                                 "maybeSendAdditionalReads can only be called for SinglePartitionReadCommand");
+        DigestRepair repair = digestRepair;
+        if (repair == null)
+            return;
 
-        public void reportMutation(InetAddressAndPort endpoint, Mutation mutation)
+        if (shouldSpeculate() && !repair.readCallback.await(cfs.sampleLatencyNanos, TimeUnit.NANOSECONDS))
         {
-            AsyncOneResponse<?> response = sendRepairMutation(mutation, endpoint);
+            Set<InetAddressAndPort> contacted = Sets.newHashSet(repair.initialContacts);
+            Token replicaToken = ((SinglePartitionReadCommand) command).partitionKey().getToken();
+            Iterable<InetAddressAndPort> candidates = BlockingReadRepairs.getCandidateEndpoints(cfs.keyspace, replicaToken, consistency);
+            boolean speculated = false;
+            for (InetAddressAndPort endpoint: Iterables.filter(candidates, e -> !contacted.contains(e)))
+            {
+                speculated = true;
+                Tracing.trace("Enqueuing speculative full data read to {}", endpoint);
+                MessagingService.instance().sendRR(command.createMessage(), endpoint, repair.readCallback);
+                break;
+            }
 
-            if (response != null)
-                responses.add(response);
+            if (speculated)
+                ReadRepairMetrics.speculatedRead.mark();
         }
+    }
 
-        public void finish()
+    @Override
+    public void maybeSendAdditionalWrites()
+    {
+        for (BlockingPartitionRepair repair: repairs)
         {
-            Futures.addCallback(Futures.allAsList(responses), new FutureCallback<List<Object>>()
-            {
-                public void onSuccess(@Nullable List<Object> result)
-                {
-                    set(result);
-                }
-
-                public void onFailure(Throwable t)
-                {
-                    setException(t);
-                }
-            });
+            repair.maybeSendAdditionalWrites(cfs.sampleLatencyNanos, TimeUnit.NANOSECONDS);
         }
     }
 
-    public void awaitRepairs(long timeout)
+    @Override
+    public void awaitWrites()
     {
-        try
+        boolean timedOut = false;
+        for (BlockingPartitionRepair repair: repairs)
         {
-            Futures.allAsList(repairs).get(timeout, TimeUnit.MILLISECONDS);
+            if (!repair.awaitRepairs(DatabaseDescriptor.getWriteRpcTimeout(), TimeUnit.MILLISECONDS))
+            {
+                timedOut = true;
+            }
         }
-        catch (TimeoutException ex)
+        if (timedOut)
         {
             // We got all responses, but timed out while repairing
-            Keyspace keyspace = Keyspace.open(command.metadata().keyspace);
-            int blockFor = consistency.blockFor(keyspace);
+            int blockFor = consistency.blockFor(cfs.keyspace);
             if (Tracing.isTracing())
                 Tracing.trace("Timed out while read-repairing after receiving all {} data and digest responses", blockFor);
             else
                 logger.debug("Timeout while read-repairing after receiving all {} data and digest responses", blockFor);
 
-            throw new ReadTimeoutException(consistency, blockFor - 1, blockFor, true);
-        }
-        catch (InterruptedException | ExecutionException e)
-        {
-            throw new RuntimeException(e);
-        }
-    }
-
-    public PartitionRepair startPartitionRepair()
-    {
-        BlockingPartitionRepair repair = new BlockingPartitionRepair(endpoints.size(), command, consistency);
-        repairs.add(repair);
-        return repair;
-    }
-
-    public void startRepair(DigestResolver digestResolver, List<InetAddressAndPort> allEndpoints, List<InetAddressAndPort> contactedEndpoints, Consumer<PartitionIterator> resultConsumer)
-    {
-        ReadRepairMetrics.repairedBlocking.mark();
-
-        // Do a full data read to resolve the correct response (and repair node that need be)
-        Keyspace keyspace = Keyspace.open(command.metadata().keyspace);
-        DataResolver resolver = new DataResolver(keyspace, command, ConsistencyLevel.ALL, allEndpoints.size(), queryStartNanoTime, this);
-        ReadCallback readCallback = new ReadCallback(resolver, ConsistencyLevel.ALL, contactedEndpoints.size(), command,
-                                                     keyspace, allEndpoints, queryStartNanoTime);
-
-        digestRepair = new DigestRepair(resolver, readCallback, resultConsumer);
-
-        for (InetAddressAndPort endpoint : contactedEndpoints)
-        {
-            Tracing.trace("Enqueuing full data read to {}", endpoint);
-            MessagingService.instance().sendRRWithFailure(command.createMessage(), endpoint, readCallback);
+            throw new ReadTimeoutException(consistency, blockFor-1, blockFor, true);
         }
     }
 
-    public void awaitRepair() throws ReadTimeoutException
+    @Override
+    public void repairPartition(DecoratedKey key, Map<InetAddressAndPort, Mutation> mutations, InetAddressAndPort[] destinations)
     {
-        if (digestRepair != null)
-        {
-            digestRepair.readCallback.awaitResults();
-            digestRepair.resultConsumer.accept(digestRepair.dataResolver.resolve());
-        }
+        BlockingPartitionRepair blockingRepair = new BlockingPartitionRepair(cfs.keyspace, key, consistency, mutations, consistency.blockFor(cfs.keyspace), destinations);
+        blockingRepair.sendInitialRepairs();
+        repairs.add(blockingRepair);
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepairs.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepairs.java b/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepairs.java
new file mode 100644
index 0000000..e5f7179
--- /dev/null
+++ b/src/java/org/apache/cassandra/service/reads/repair/BlockingReadRepairs.java
@@ -0,0 +1,114 @@
+/*
+ * 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.service.reads.repair;
+
+import java.util.List;
+
+import com.google.common.collect.Iterables;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.partitions.PartitionUpdate;
+import org.apache.cassandra.dht.Token;
+import org.apache.cassandra.exceptions.ReadTimeoutException;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.locator.NetworkTopologyStrategy;
+import org.apache.cassandra.net.MessagingService;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.StorageProxy;
+import org.apache.cassandra.tracing.Tracing;
+
+public class BlockingReadRepairs
+{
+    private static final Logger logger = LoggerFactory.getLogger(BlockingReadRepairs.class);
+
+    private static final boolean DROP_OVERSIZED_READ_REPAIR_MUTATIONS =
+        Boolean.getBoolean("cassandra.drop_oversized_readrepair_mutations");
+
+    /**
+     * Returns all of the endpoints that are replicas for the given key. If the consistency level is datacenter
+     * local, only the endpoints in the local dc will be returned.
+     */
+    static Iterable<InetAddressAndPort> getCandidateEndpoints(Keyspace keyspace, Token token, ConsistencyLevel consistency)
+    {
+        List<InetAddressAndPort> endpoints = StorageProxy.getLiveSortedEndpoints(keyspace, token);
+        return consistency.isDatacenterLocal() && keyspace.getReplicationStrategy() instanceof NetworkTopologyStrategy
+               ? Iterables.filter(endpoints, ConsistencyLevel::isLocal)
+               : endpoints;
+    }
+
+    /**
+     * Create a read repair mutation from the given update, if the mutation is not larger than the maximum
+     * mutation size, otherwise return null. Or, if we're configured to be strict, throw an exception.
+     */
+    public static Mutation createRepairMutation(PartitionUpdate update, ConsistencyLevel consistency, InetAddressAndPort destination, boolean suppressException)
+    {
+        if (update == null)
+            return null;
+
+        DecoratedKey key = update.partitionKey();
+        Mutation mutation = new Mutation(update);
+        Keyspace keyspace = Keyspace.open(mutation.getKeyspaceName());
+        TableMetadata metadata = update.metadata();
+
+        int messagingVersion = MessagingService.instance().getVersion(destination);
+
+        int    mutationSize = (int) Mutation.serializer.serializedSize(mutation, messagingVersion);
+        int maxMutationSize = DatabaseDescriptor.getMaxMutationSize();
+
+
+        if (mutationSize <= maxMutationSize)
+        {
+            return mutation;
+        }
+        else if (DROP_OVERSIZED_READ_REPAIR_MUTATIONS)
+        {
+            logger.debug("Encountered an oversized ({}/{}) read repair mutation for table {}, key {}, node {}",
+                         mutationSize,
+                         maxMutationSize,
+                         metadata,
+                         metadata.partitionKeyType.getString(key.getKey()),
+                         destination);
+            return null;
+        }
+        else
+        {
+            logger.warn("Encountered an oversized ({}/{}) read repair mutation for table {}, key {}, node {}",
+                        mutationSize,
+                        maxMutationSize,
+                        metadata,
+                        metadata.partitionKeyType.getString(key.getKey()),
+                        destination);
+
+            if (!suppressException)
+            {
+                int blockFor = consistency.blockFor(keyspace);
+                Tracing.trace("Timed out while read-repairing after receiving all {} data and digest responses", blockFor);
+                throw new ReadTimeoutException(consistency, blockFor - 1, blockFor, true);
+            }
+            return null;
+        }
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java b/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java
index 4436f3a..6e161a8 100644
--- a/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java
+++ b/src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java
@@ -19,8 +19,11 @@
 package org.apache.cassandra.service.reads.repair;
 
 import java.util.List;
+import java.util.Map;
 import java.util.function.Consumer;
 
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Mutation;
 import org.apache.cassandra.db.partitions.PartitionIterator;
 import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
 import org.apache.cassandra.exceptions.ReadTimeoutException;
@@ -43,7 +46,31 @@ public class NoopReadRepair implements ReadRepair
         resultConsumer.accept(digestResolver.getData());
     }
 
-    public void awaitRepair() throws ReadTimeoutException
+    public void awaitReads() throws ReadTimeoutException
     {
     }
+
+    @Override
+    public void maybeSendAdditionalReads()
+    {
+
+    }
+
+    @Override
+    public void maybeSendAdditionalWrites()
+    {
+
+    }
+
+    @Override
+    public void awaitWrites()
+    {
+
+    }
+
+    @Override
+    public void repairPartition(DecoratedKey key, Map<InetAddressAndPort, Mutation> mutations, InetAddressAndPort[] destinations)
+    {
+
+    }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/service/reads/repair/PartitionIteratorMergeListener.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/reads/repair/PartitionIteratorMergeListener.java b/src/java/org/apache/cassandra/service/reads/repair/PartitionIteratorMergeListener.java
index 4ccdcbf..6cf761a 100644
--- a/src/java/org/apache/cassandra/service/reads/repair/PartitionIteratorMergeListener.java
+++ b/src/java/org/apache/cassandra/service/reads/repair/PartitionIteratorMergeListener.java
@@ -20,8 +20,8 @@ package org.apache.cassandra.service.reads.repair;
 
 import java.util.List;
 
-import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.Columns;
+import org.apache.cassandra.db.ConsistencyLevel;
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.ReadCommand;
 import org.apache.cassandra.db.RegularAndStaticColumns;
@@ -34,18 +34,20 @@ public class PartitionIteratorMergeListener implements UnfilteredPartitionIterat
 {
     private final InetAddressAndPort[] sources;
     private final ReadCommand command;
-    private final RepairListener repairListener;
+    private final ConsistencyLevel consistency;
+    private final ReadRepair readRepair;
 
-    public PartitionIteratorMergeListener(InetAddressAndPort[] sources, ReadCommand command, RepairListener repairListener)
+    public PartitionIteratorMergeListener(InetAddressAndPort[] sources, ReadCommand command, ConsistencyLevel consistency, ReadRepair readRepair)
     {
         this.sources = sources;
         this.command = command;
-        this.repairListener = repairListener;
+        this.consistency = consistency;
+        this.readRepair = readRepair;
     }
 
     public UnfilteredRowIterators.MergeListener getRowMergeListener(DecoratedKey partitionKey, List<UnfilteredRowIterator> versions)
     {
-        return new RowIteratorMergeListener(partitionKey, columns(versions), isReversed(versions), sources, command, repairListener);
+        return new RowIteratorMergeListener(partitionKey, columns(versions), isReversed(versions), sources, command, consistency, readRepair);
     }
 
     private RegularAndStaticColumns columns(List<UnfilteredRowIterator> versions)
@@ -81,7 +83,6 @@ public class PartitionIteratorMergeListener implements UnfilteredPartitionIterat
 
     public void close()
     {
-        repairListener.awaitRepairs(DatabaseDescriptor.getWriteRpcTimeout());
     }
 }
 

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java b/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java
index 289875d..a1a9546 100644
--- a/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java
+++ b/src/java/org/apache/cassandra/service/reads/repair/ReadRepair.java
@@ -18,9 +18,12 @@
 package org.apache.cassandra.service.reads.repair;
 
 import java.util.List;
+import java.util.Map;
 import java.util.function.Consumer;
 
 import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Mutation;
 import org.apache.cassandra.db.ReadCommand;
 import org.apache.cassandra.db.partitions.PartitionIterator;
 import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
@@ -38,6 +41,10 @@ public interface ReadRepair
     /**
      * Called when the digests from the initial read don't match. Reads may block on the
      * repair started by this method.
+     * @param digestResolver supplied so we can get the original data response
+     * @param allEndpoints all available replicas for this read
+     * @param contactedEndpoints the replicas we actually sent requests to
+     * @param resultConsumer hook for the repair to set it's result on completion
      */
     public void startRepair(DigestResolver digestResolver,
                             List<InetAddressAndPort> allEndpoints,
@@ -45,12 +52,41 @@ public interface ReadRepair
                             Consumer<PartitionIterator> resultConsumer);
 
     /**
-     * Wait for any operations started by {@link ReadRepair#startRepair} to complete
+     * Block on the reads (or timeout) sent out in {@link ReadRepair#startRepair}
      */
-    public void awaitRepair() throws ReadTimeoutException;
+    public void awaitReads() throws ReadTimeoutException;
 
-    static ReadRepair create(ReadCommand command, List<InetAddressAndPort> endpoints, long queryStartNanoTime, ConsistencyLevel consistency)
+    /**
+     * if it looks like we might not receive data requests from everyone in time, send additional requests
+     * to additional replicas not contacted in the initial full data read. If the collection of nodes that
+     * end up responding in time end up agreeing on the data, and we don't consider the response from the
+     * disagreeing replica that triggered the read repair, that's ok, since the disagreeing data would not
+     * have been successfully written and won't be included in the response the the client, preserving the
+     * expectation of monotonic quorum reads
+     */
+    public void maybeSendAdditionalReads();
+
+    /**
+     * If it looks like we might not receive acks for all the repair mutations we sent out, combine all
+     * the unacked mutations and send them to the minority of nodes not involved in the read repair data
+     * read / write cycle. We will accept acks from them in lieu of acks from the initial mutations sent
+     * out, so long as we receive the same number of acks as repair mutations transmitted. This prevents
+     * misbehaving nodes from killing a quorum read, while continuing to guarantee monotonic quorum reads
+     */
+    public void maybeSendAdditionalWrites();
+
+    /**
+     * Hook for the merge listener to start repairs on individual partitions.
+     */
+    void repairPartition(DecoratedKey key, Map<InetAddressAndPort, Mutation> mutations, InetAddressAndPort[] destinations);
+
+    /**
+     * Block on any mutations (or timeout) we sent out to repair replicas in {@link ReadRepair#repairPartition}
+     */
+    public void awaitWrites();
+
+    static ReadRepair create(ReadCommand command, long queryStartNanoTime, ConsistencyLevel consistency)
     {
-        return new BlockingReadRepair(command, endpoints, queryStartNanoTime, consistency);
+        return new BlockingReadRepair(command, queryStartNanoTime, consistency);
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/service/reads/repair/RepairListener.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/reads/repair/RepairListener.java b/src/java/org/apache/cassandra/service/reads/repair/RepairListener.java
deleted file mode 100644
index 174c0e7..0000000
--- a/src/java/org/apache/cassandra/service/reads/repair/RepairListener.java
+++ /dev/null
@@ -1,34 +0,0 @@
-/*
- * 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.service.reads.repair;
-
-import org.apache.cassandra.db.Mutation;
-import org.apache.cassandra.locator.InetAddressAndPort;
-
-public interface RepairListener
-{
-    interface PartitionRepair
-    {
-        void reportMutation(InetAddressAndPort endpoint, Mutation mutation);
-        void finish();
-    }
-
-    PartitionRepair startPartitionRepair();
-    void awaitRepairs(long timeoutMillis);
-}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/src/java/org/apache/cassandra/service/reads/repair/RowIteratorMergeListener.java
----------------------------------------------------------------------
diff --git a/src/java/org/apache/cassandra/service/reads/repair/RowIteratorMergeListener.java b/src/java/org/apache/cassandra/service/reads/repair/RowIteratorMergeListener.java
index f11d264..cb6707d 100644
--- a/src/java/org/apache/cassandra/service/reads/repair/RowIteratorMergeListener.java
+++ b/src/java/org/apache/cassandra/service/reads/repair/RowIteratorMergeListener.java
@@ -19,9 +19,13 @@
 package org.apache.cassandra.service.reads.repair;
 
 import java.util.Arrays;
+import java.util.Map;
+
+import com.google.common.collect.Maps;
 
 import org.apache.cassandra.db.Clustering;
 import org.apache.cassandra.db.ClusteringBound;
+import org.apache.cassandra.db.ConsistencyLevel;
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.DeletionTime;
 import org.apache.cassandra.db.LivenessInfo;
@@ -49,6 +53,7 @@ public class RowIteratorMergeListener implements UnfilteredRowIterators.MergeLis
     private final boolean isReversed;
     private final InetAddressAndPort[] sources;
     private final ReadCommand command;
+    private final ConsistencyLevel consistency;
 
     private final PartitionUpdate.Builder[] repairs;
 
@@ -64,9 +69,9 @@ public class RowIteratorMergeListener implements UnfilteredRowIterators.MergeLis
     // For each source, record if there is an open range to send as repair, and from where.
     private final ClusteringBound[] markerToRepair;
 
-    private final RepairListener repairListener;
+    private final ReadRepair readRepair;
 
-    public RowIteratorMergeListener(DecoratedKey partitionKey, RegularAndStaticColumns columns, boolean isReversed, InetAddressAndPort[] sources, ReadCommand command, RepairListener repairListener)
+    public RowIteratorMergeListener(DecoratedKey partitionKey, RegularAndStaticColumns columns, boolean isReversed, InetAddressAndPort[] sources, ReadCommand command, ConsistencyLevel consistency, ReadRepair readRepair)
     {
         this.partitionKey = partitionKey;
         this.columns = columns;
@@ -77,7 +82,8 @@ public class RowIteratorMergeListener implements UnfilteredRowIterators.MergeLis
         sourceDeletionTime = new DeletionTime[sources.length];
         markerToRepair = new ClusteringBound[sources.length];
         this.command = command;
-        this.repairListener = repairListener;
+        this.consistency = consistency;
+        this.readRepair = readRepair;
 
         this.diffListener = new RowDiffListener()
         {
@@ -300,22 +306,25 @@ public class RowIteratorMergeListener implements UnfilteredRowIterators.MergeLis
 
     public void close()
     {
-        RepairListener.PartitionRepair repair = null;
+        Map<InetAddressAndPort, Mutation> mutations = null;
         for (int i = 0; i < repairs.length; i++)
         {
             if (repairs[i] == null)
                 continue;
 
-            if (repair == null)
-            {
-                repair = repairListener.startPartitionRepair();
-            }
-            repair.reportMutation(sources[i], new Mutation(repairs[i].build()));
+            Mutation mutation = BlockingReadRepairs.createRepairMutation(repairs[i].build(), consistency, sources[i], false);
+            if (mutation == null)
+                continue;
+
+            if (mutations == null)
+                mutations = Maps.newHashMapWithExpectedSize(sources.length);
+
+            mutations.put(sources[i], mutation);
         }
 
-        if (repair != null)
+        if (mutations != null)
         {
-            repair.finish();
+            readRepair.repairPartition(partitionKey, mutations, sources);
         }
     }
 }

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/test/unit/org/apache/cassandra/service/reads/DataResolverTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/service/reads/DataResolverTest.java b/test/unit/org/apache/cassandra/service/reads/DataResolverTest.java
index 4d4d398..1a5aa7a 100644
--- a/test/unit/org/apache/cassandra/service/reads/DataResolverTest.java
+++ b/test/unit/org/apache/cassandra/service/reads/DataResolverTest.java
@@ -22,16 +22,12 @@ import java.nio.ByteBuffer;
 import java.util.*;
 
 import com.google.common.collect.Iterators;
-import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import org.junit.*;
 
 import org.apache.cassandra.SchemaLoader;
 import org.apache.cassandra.Util;
 import org.apache.cassandra.locator.InetAddressAndPort;
-import org.apache.cassandra.service.reads.DataResolver;
-import org.apache.cassandra.service.reads.repair.BlockingReadRepair;
-import org.apache.cassandra.service.reads.repair.NoopReadRepair;
 import org.apache.cassandra.schema.TableMetadata;
 import org.apache.cassandra.schema.ColumnMetadata;
 import org.apache.cassandra.config.DatabaseDescriptor;
@@ -47,8 +43,6 @@ import org.apache.cassandra.db.partitions.*;
 import org.apache.cassandra.exceptions.ConfigurationException;
 import org.apache.cassandra.net.*;
 import org.apache.cassandra.schema.KeyspaceParams;
-import org.apache.cassandra.service.reads.repair.ReadRepair;
-import org.apache.cassandra.service.reads.repair.RepairListener;
 import org.apache.cassandra.service.reads.repair.TestableReadRepair;
 import org.apache.cassandra.utils.ByteBufferUtil;
 import org.apache.cassandra.utils.FBUtilities;
@@ -118,7 +112,7 @@ public class DataResolverTest
 
         nowInSec = FBUtilities.nowInSeconds();
         command = Util.cmd(cfs, dk).withNowInSeconds(nowInSec).build();
-        readRepair = new TestableReadRepair(command);
+        readRepair = new TestableReadRepair(command, ConsistencyLevel.QUORUM);
     }
 
     @Test
@@ -346,7 +340,7 @@ public class DataResolverTest
     @Test
     public void testResolveWithBothEmpty()
     {
-        TestableReadRepair readRepair = new TestableReadRepair(command);
+        TestableReadRepair readRepair = new TestableReadRepair(command, ConsistencyLevel.QUORUM);
         DataResolver resolver = new DataResolver(ks, command, ConsistencyLevel.ALL, 2, System.nanoTime(), readRepair);
         resolver.preprocess(readResponseMessage(peer(), EmptyIterators.unfilteredPartition(cfm)));
         resolver.preprocess(readResponseMessage(peer(), EmptyIterators.unfilteredPartition(cfm)));
@@ -715,7 +709,7 @@ public class DataResolverTest
     public void testResolveComplexDelete()
     {
         ReadCommand cmd = Util.cmd(cfs2, dk).withNowInSeconds(nowInSec).build();
-        TestableReadRepair readRepair = new TestableReadRepair(cmd);
+        TestableReadRepair readRepair = new TestableReadRepair(cmd, ConsistencyLevel.QUORUM);
         DataResolver resolver = new DataResolver(ks, cmd, ConsistencyLevel.ALL, 2, System.nanoTime(), readRepair);
 
         long[] ts = {100, 200};
@@ -767,7 +761,7 @@ public class DataResolverTest
     public void testResolveDeletedCollection()
     {
         ReadCommand cmd = Util.cmd(cfs2, dk).withNowInSeconds(nowInSec).build();
-        TestableReadRepair readRepair = new TestableReadRepair(cmd);
+        TestableReadRepair readRepair = new TestableReadRepair(cmd, ConsistencyLevel.QUORUM);
         DataResolver resolver = new DataResolver(ks, cmd, ConsistencyLevel.ALL, 2, System.nanoTime(), readRepair);
 
         long[] ts = {100, 200};
@@ -810,7 +804,7 @@ public class DataResolverTest
     public void testResolveNewCollection()
     {
         ReadCommand cmd = Util.cmd(cfs2, dk).withNowInSeconds(nowInSec).build();
-        TestableReadRepair readRepair = new TestableReadRepair(cmd);
+        TestableReadRepair readRepair = new TestableReadRepair(cmd, ConsistencyLevel.QUORUM);
         DataResolver resolver = new DataResolver(ks, cmd, ConsistencyLevel.ALL, 2, System.nanoTime(), readRepair);
 
         long[] ts = {100, 200};
@@ -859,7 +853,7 @@ public class DataResolverTest
     public void testResolveNewCollectionOverwritingDeleted()
     {
         ReadCommand cmd = Util.cmd(cfs2, dk).withNowInSeconds(nowInSec).build();
-        TestableReadRepair readRepair = new TestableReadRepair(cmd);
+        TestableReadRepair readRepair = new TestableReadRepair(cmd, ConsistencyLevel.QUORUM);
         DataResolver resolver = new DataResolver(ks, cmd, ConsistencyLevel.ALL, 2, System.nanoTime(), readRepair);
 
         long[] ts = {100, 200};

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/test/unit/org/apache/cassandra/service/reads/repair/ReadRepairTest.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/service/reads/repair/ReadRepairTest.java b/test/unit/org/apache/cassandra/service/reads/repair/ReadRepairTest.java
new file mode 100644
index 0000000..75d6e83
--- /dev/null
+++ b/test/unit/org/apache/cassandra/service/reads/repair/ReadRepairTest.java
@@ -0,0 +1,361 @@
+/*
+ * 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.service.reads.repair;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+
+import com.google.common.collect.ImmutableList;
+import com.google.common.collect.Iterables;
+import com.google.common.collect.Lists;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.cql3.ColumnIdentifier;
+import org.apache.cassandra.cql3.statements.schema.CreateTableStatement;
+import org.apache.cassandra.db.Clustering;
+import org.apache.cassandra.db.ColumnFamilyStore;
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.DecoratedKey;
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.db.Mutation;
+import org.apache.cassandra.db.partitions.PartitionUpdate;
+import org.apache.cassandra.db.rows.BTreeRow;
+import org.apache.cassandra.db.rows.BufferCell;
+import org.apache.cassandra.db.rows.Cell;
+import org.apache.cassandra.db.rows.Row;
+import org.apache.cassandra.locator.InetAddressAndPort;
+import org.apache.cassandra.net.MessageOut;
+import org.apache.cassandra.schema.KeyspaceMetadata;
+import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.schema.MigrationManager;
+import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.schema.Tables;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+public class ReadRepairTest
+{
+    static Keyspace ks;
+    static ColumnFamilyStore cfs;
+    static TableMetadata cfm;
+    static InetAddressAndPort target1;
+    static InetAddressAndPort target2;
+    static InetAddressAndPort target3;
+    static List<InetAddressAndPort> targets;
+
+    private static class InstrumentedReadRepairHandler extends BlockingPartitionRepair
+    {
+        public InstrumentedReadRepairHandler(Keyspace keyspace, DecoratedKey key, ConsistencyLevel consistency, Map<InetAddressAndPort, Mutation> repairs, int maxBlockFor, InetAddressAndPort[] participants)
+        {
+            super(keyspace, key, consistency, repairs, maxBlockFor, participants);
+        }
+
+        Map<InetAddressAndPort, Mutation> mutationsSent = new HashMap<>();
+
+        protected void sendRR(MessageOut<Mutation> message, InetAddressAndPort endpoint)
+        {
+            mutationsSent.put(endpoint, message.payload);
+        }
+
+        List<InetAddressAndPort> candidates = targets;
+
+        protected List<InetAddressAndPort> getCandidateEndpoints()
+        {
+            return candidates;
+        }
+
+        @Override
+        protected boolean isLocal(InetAddressAndPort endpoint)
+        {
+            return targets.contains(endpoint);
+        }
+    }
+
+    static long now = TimeUnit.NANOSECONDS.toMicros(System.nanoTime());
+    static DecoratedKey key;
+    static Cell cell1;
+    static Cell cell2;
+    static Cell cell3;
+    static Mutation resolved;
+
+    private static void assertRowsEqual(Row expected, Row actual)
+    {
+        try
+        {
+            Assert.assertEquals(expected == null, actual == null);
+            if (expected == null)
+                return;
+            Assert.assertEquals(expected.clustering(), actual.clustering());
+            Assert.assertEquals(expected.deletion(), actual.deletion());
+            Assert.assertArrayEquals(Iterables.toArray(expected.cells(), Cell.class), Iterables.toArray(expected.cells(), Cell.class));
+        } catch (Throwable t)
+        {
+            throw new AssertionError(String.format("Row comparison failed, expected %s got %s", expected, actual), t);
+        }
+    }
+
+    @BeforeClass
+    public static void setUpClass() throws Throwable
+    {
+        SchemaLoader.loadSchema();
+        String ksName = "ks";
+
+        cfm = CreateTableStatement.parse("CREATE TABLE tbl (k int primary key, v text)", ksName).build();
+        KeyspaceMetadata ksm = KeyspaceMetadata.create(ksName, KeyspaceParams.simple(3), Tables.of(cfm));
+        MigrationManager.announceNewKeyspace(ksm, false);
+
+        ks = Keyspace.open(ksName);
+        cfs = ks.getColumnFamilyStore("tbl");
+
+        cfs.sampleLatencyNanos = 0;
+
+        target1 = InetAddressAndPort.getByName("127.0.0.255");
+        target2 = InetAddressAndPort.getByName("127.0.0.254");
+        target3 = InetAddressAndPort.getByName("127.0.0.253");
+
+        targets = ImmutableList.of(target1, target2, target3);
+
+        // default test values
+        key  = dk(5);
+        cell1 = cell("v", "val1", now);
+        cell2 = cell("v", "val2", now);
+        cell3 = cell("v", "val3", now);
+        resolved = mutation(cell1, cell2);
+    }
+
+    private static DecoratedKey dk(int v)
+    {
+        return DatabaseDescriptor.getPartitioner().decorateKey(ByteBufferUtil.bytes(v));
+    }
+
+    private static Cell cell(String name, String value, long timestamp)
+    {
+        return BufferCell.live(cfm.getColumn(ColumnIdentifier.getInterned(name, false)), timestamp, ByteBufferUtil.bytes(value));
+    }
+
+    private static Mutation mutation(Cell... cells)
+    {
+        Row.Builder builder = BTreeRow.unsortedBuilder(0);
+        builder.newRow(Clustering.EMPTY);
+        for (Cell cell: cells)
+        {
+            builder.addCell(cell);
+        }
+        return new Mutation(PartitionUpdate.singleRowUpdate(cfm, key, builder.build()));
+    }
+
+    private static InstrumentedReadRepairHandler createRepairHandler(Map<InetAddressAndPort, Mutation> repairs, int maxBlockFor, Collection<InetAddressAndPort> participants)
+    {
+        InetAddressAndPort[] participantArray = new InetAddressAndPort[participants.size()];
+        participants.toArray(participantArray);
+        return new InstrumentedReadRepairHandler(ks, key, ConsistencyLevel.LOCAL_QUORUM, repairs, maxBlockFor, participantArray);
+    }
+
+    private static InstrumentedReadRepairHandler createRepairHandler(Map<InetAddressAndPort, Mutation> repairs, int maxBlockFor)
+    {
+        return createRepairHandler(repairs, maxBlockFor, repairs.keySet());
+    }
+
+    @Test
+    public void consistencyLevelTest() throws Exception
+    {
+        Assert.assertTrue(ConsistencyLevel.QUORUM.satisfies(ConsistencyLevel.QUORUM, ks));
+        Assert.assertTrue(ConsistencyLevel.THREE.satisfies(ConsistencyLevel.QUORUM, ks));
+        Assert.assertTrue(ConsistencyLevel.TWO.satisfies(ConsistencyLevel.QUORUM, ks));
+        Assert.assertFalse(ConsistencyLevel.ONE.satisfies(ConsistencyLevel.QUORUM, ks));
+        Assert.assertFalse(ConsistencyLevel.ANY.satisfies(ConsistencyLevel.QUORUM, ks));
+    }
+
+    private static void assertMutationEqual(Mutation expected, Mutation actual)
+    {
+        Assert.assertEquals(expected.getKeyspaceName(), actual.getKeyspaceName());
+        Assert.assertEquals(expected.key(), actual.key());
+        Assert.assertEquals(expected.key(), actual.key());
+        PartitionUpdate expectedUpdate = Iterables.getOnlyElement(expected.getPartitionUpdates());
+        PartitionUpdate actualUpdate = Iterables.getOnlyElement(actual.getPartitionUpdates());
+        assertRowsEqual(Iterables.getOnlyElement(expectedUpdate), Iterables.getOnlyElement(actualUpdate));
+    }
+
+    @Test
+    public void additionalMutationRequired() throws Exception
+    {
+
+        Mutation repair1 = mutation(cell2);
+        Mutation repair2 = mutation(cell1);
+
+        // check that the correct repairs are calculated
+        Map<InetAddressAndPort, Mutation> repairs = new HashMap<>();
+        repairs.put(target1, repair1);
+        repairs.put(target2, repair2);
+
+
+        InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2);
+
+        Assert.assertTrue(handler.mutationsSent.isEmpty());
+
+        // check that the correct mutations are sent
+        handler.sendInitialRepairs();
+        Assert.assertEquals(2, handler.mutationsSent.size());
+        assertMutationEqual(repair1, handler.mutationsSent.get(target1));
+        assertMutationEqual(repair2, handler.mutationsSent.get(target2));
+
+        // check that a combined mutation is speculatively sent to the 3rd target
+        handler.mutationsSent.clear();
+        handler.maybeSendAdditionalWrites(0, TimeUnit.NANOSECONDS);
+        Assert.assertEquals(1, handler.mutationsSent.size());
+        assertMutationEqual(resolved, handler.mutationsSent.get(target3));
+
+        // check repairs stop blocking after receiving 2 acks
+        Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS));
+        handler.ack(target1);
+        Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS));
+        handler.ack(target3);
+        Assert.assertTrue(handler.awaitRepairs(0, TimeUnit.NANOSECONDS));
+
+    }
+
+    /**
+     * If we've received enough acks, we shouldn't send any additional mutations
+     */
+    @Test
+    public void noAdditionalMutationRequired() throws Exception
+    {
+        Map<InetAddressAndPort, Mutation> repairs = new HashMap<>();
+        repairs.put(target1, mutation(cell2));
+        repairs.put(target2, mutation(cell1));
+
+        InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2);
+        handler.sendInitialRepairs();
+        handler.ack(target1);
+        handler.ack(target2);
+
+        // both replicas have acked, we shouldn't send anything else out
+        handler.mutationsSent.clear();
+        handler.maybeSendAdditionalWrites(0, TimeUnit.NANOSECONDS);
+        Assert.assertTrue(handler.mutationsSent.isEmpty());
+    }
+
+    /**
+     * If there are no additional nodes we can send mutations to, we... shouldn't
+     */
+    @Test
+    public void noAdditionalMutationPossible() throws Exception
+    {
+        Map<InetAddressAndPort, Mutation> repairs = new HashMap<>();
+        repairs.put(target1, mutation(cell2));
+        repairs.put(target2, mutation(cell1));
+
+        InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2);
+        handler.sendInitialRepairs();
+
+        // we've already sent mutations to all candidates, so we shouldn't send any more
+        handler.candidates = Lists.newArrayList(target1, target2);
+        handler.mutationsSent.clear();
+        handler.maybeSendAdditionalWrites(0, TimeUnit.NANOSECONDS);
+        Assert.assertTrue(handler.mutationsSent.isEmpty());
+    }
+
+    /**
+     * If we didn't send a repair to a replica because there wasn't a diff with the
+     * resolved column family, we shouldn't send it a speculative mutation
+     */
+    @Test
+    public void mutationsArentSentToInSyncNodes() throws Exception
+    {
+        Mutation repair1 = mutation(cell2);
+
+        Map<InetAddressAndPort, Mutation> repairs = new HashMap<>();
+        repairs.put(target1, repair1);
+        Collection<InetAddressAndPort> participants = Lists.newArrayList(target1, target2);
+
+        // check that the correct initial mutations are sent out
+        InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2, participants);
+        handler.sendInitialRepairs();
+        Assert.assertEquals(1, handler.mutationsSent.size());
+        Assert.assertTrue(handler.mutationsSent.containsKey(target1));
+
+        // check that speculative mutations aren't sent to target2
+        handler.mutationsSent.clear();
+        handler.maybeSendAdditionalWrites(0, TimeUnit.NANOSECONDS);
+        Assert.assertEquals(1, handler.mutationsSent.size());
+        Assert.assertTrue(handler.mutationsSent.containsKey(target3));
+    }
+
+    @Test
+    public void onlyBlockOnQuorum()
+    {
+        Map<InetAddressAndPort, Mutation> repairs = new HashMap<>();
+        repairs.put(target1, mutation(cell1));
+        repairs.put(target2, mutation(cell2));
+        repairs.put(target3, mutation(cell3));
+        Assert.assertEquals(3, repairs.size());
+
+        InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2);
+        handler.sendInitialRepairs();
+
+        Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS));
+        handler.ack(target1);
+        Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS));
+
+        // here we should stop blocking, even though we've sent 3 repairs
+        handler.ack(target2);
+        Assert.assertTrue(handler.awaitRepairs(0, TimeUnit.NANOSECONDS));
+
+    }
+
+    /**
+     * For dc local consistency levels, noop mutations and responses from remote dcs should not affect effective blockFor
+     */
+    @Test
+    public void remoteDCTest() throws Exception
+    {
+        Map<InetAddressAndPort, Mutation> repairs = new HashMap<>();
+        repairs.put(target1, mutation(cell1));
+
+
+        InetAddressAndPort remote1 = InetAddressAndPort.getByName("10.0.0.1");
+        InetAddressAndPort remote2 = InetAddressAndPort.getByName("10.0.0.2");
+        repairs.put(remote1, mutation(cell1));
+
+        Collection<InetAddressAndPort> participants = Lists.newArrayList(target1, target2, remote1, remote2);
+
+        InstrumentedReadRepairHandler handler = createRepairHandler(repairs, 2, participants);
+        handler.sendInitialRepairs();
+        Assert.assertEquals(2, handler.mutationsSent.size());
+        Assert.assertTrue(handler.mutationsSent.containsKey(target1));
+        Assert.assertTrue(handler.mutationsSent.containsKey(remote1));
+
+        Assert.assertEquals(1, handler.waitingOn());
+        Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS));
+
+        handler.ack(remote1);
+        Assert.assertEquals(1, handler.waitingOn());
+        Assert.assertFalse(handler.awaitRepairs(0, TimeUnit.NANOSECONDS));
+
+        handler.ack(target1);
+        Assert.assertEquals(0, handler.waitingOn());
+        Assert.assertTrue(handler.awaitRepairs(0, TimeUnit.NANOSECONDS));
+    }
+}

http://git-wip-us.apache.org/repos/asf/cassandra/blob/644676b0/test/unit/org/apache/cassandra/service/reads/repair/TestableReadRepair.java
----------------------------------------------------------------------
diff --git a/test/unit/org/apache/cassandra/service/reads/repair/TestableReadRepair.java b/test/unit/org/apache/cassandra/service/reads/repair/TestableReadRepair.java
index 5664c9b..f97980b 100644
--- a/test/unit/org/apache/cassandra/service/reads/repair/TestableReadRepair.java
+++ b/test/unit/org/apache/cassandra/service/reads/repair/TestableReadRepair.java
@@ -23,6 +23,8 @@ import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
 
+import org.apache.cassandra.db.ConsistencyLevel;
+import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.Mutation;
 import org.apache.cassandra.db.ReadCommand;
 import org.apache.cassandra.db.partitions.PartitionIterator;
@@ -31,60 +33,59 @@ import org.apache.cassandra.exceptions.ReadTimeoutException;
 import org.apache.cassandra.locator.InetAddressAndPort;
 import org.apache.cassandra.service.reads.DigestResolver;
 
-public class TestableReadRepair implements ReadRepair, RepairListener
+public class TestableReadRepair implements ReadRepair
 {
     public final Map<InetAddressAndPort, Mutation> sent = new HashMap<>();
 
     private final ReadCommand command;
+    private final ConsistencyLevel consistency;
 
-    public TestableReadRepair(ReadCommand command)
+    public TestableReadRepair(ReadCommand command, ConsistencyLevel consistency)
     {
         this.command = command;
+        this.consistency = consistency;
     }
 
-    private class TestablePartitionRepair implements RepairListener.PartitionRepair
+    @Override
+    public UnfilteredPartitionIterators.MergeListener getMergeListener(InetAddressAndPort[] endpoints)
     {
-        @Override
-        public void reportMutation(InetAddressAndPort endpoint, Mutation mutation)
-        {
-            sent.put(endpoint, mutation);
-        }
+        return new PartitionIteratorMergeListener(endpoints, command, consistency, this);
+    }
 
-        @Override
-        public void finish()
-        {
+    @Override
+    public void startRepair(DigestResolver digestResolver, List<InetAddressAndPort> allEndpoints, List<InetAddressAndPort> contactedEndpoints, Consumer<PartitionIterator> resultConsumer)
+    {
 
-        }
     }
 
     @Override
-    public UnfilteredPartitionIterators.MergeListener getMergeListener(InetAddressAndPort[] endpoints)
+    public void awaitReads() throws ReadTimeoutException
     {
-        return new PartitionIteratorMergeListener(endpoints, command, this);
+
     }
 
     @Override
-    public void startRepair(DigestResolver digestResolver, List<InetAddressAndPort> allEndpoints, List<InetAddressAndPort> contactedEndpoints, Consumer<PartitionIterator> resultConsumer)
+    public void maybeSendAdditionalReads()
     {
 
     }
 
     @Override
-    public void awaitRepair() throws ReadTimeoutException
+    public void maybeSendAdditionalWrites()
     {
 
     }
 
     @Override
-    public PartitionRepair startPartitionRepair()
+    public void awaitWrites()
     {
-        return new TestablePartitionRepair();
+
     }
 
     @Override
-    public void awaitRepairs(long timeoutMillis)
+    public void repairPartition(DecoratedKey key, Map<InetAddressAndPort, Mutation> mutations, InetAddressAndPort[] destinations)
     {
-
+        sent.putAll(mutations);
     }
 
     public Mutation getForEndpoint(InetAddressAndPort endpoint)


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@cassandra.apache.org
For additional commands, e-mail: commits-help@cassandra.apache.org


Mime
View raw message