Merge branch 'cassandra-3.11' into trunk

CHANGES.txt

@@ -7,6 +7,7 @@ Merged from 3.11:
  * stop_paranoid disk failure policy is ignored on CorruptSSTableException after node is up (CASSANDRA-15191)
  * Frozen RawTuple is not annotated with frozen in the toString method (CASSANDRA-15857)
 Merged from 3.0:
+ * Operational improvements and hardening for replica filtering protection (CASSANDRA-15907)
  * Fix empty/null json string representation (CASSANDRA-15896)
 Merged from 2.2:
  * Fix CQL parsing of collections when the column type is reversed (CASSANDRA-15814)

build.xml

@@ -563,8 +563,7 @@
           <dependency groupId="org.mockito" artifactId="mockito-core" version="3.2.4" />
           <dependency groupId="org.quicktheories" artifactId="quicktheories" version="0.25" />
           <dependency groupId="com.google.code.java-allocation-instrumenter" artifactId="java-allocation-instrumenter" version="${allocation-instrumenter.version}" />
-          <dependency groupId="org.apache.cassandra" artifactId="dtest-api" version="0.0.3" />
-
+          <dependency groupId="org.apache.cassandra" artifactId="dtest-api" version="0.0.4" />
           <dependency groupId="org.apache.rat" artifactId="apache-rat" version="0.10">
              <exclusion groupId="commons-lang" artifactId="commons-lang"/>
           </dependency>

conf/cassandra.yaml

@@ -1226,6 +1226,26 @@ transparent_data_encryption_options:
 tombstone_warn_threshold: 1000
 tombstone_failure_threshold: 100000
 
+# Filtering and secondary index queries at read consistency levels above ONE/LOCAL_ONE use a
+# mechanism called replica filtering protection to ensure that results from stale replicas do
+# not violate consistency. (See CASSANDRA-8272 and CASSANDRA-15907 for more details.) This
+# mechanism materializes replica results by partition on-heap at the coordinator. The more possibly
+# stale results returned by the replicas, the more rows materialized during the query.
+replica_filtering_protection:
+    # These thresholds exist to limit the damage severely out-of-date replicas can cause during these
+    # queries. They limit the number of rows from all replicas individual index and filtering queries
+    # can materialize on-heap to return correct results at the desired read consistency level.
+    #
+    # "cached_replica_rows_warn_threshold" is the per-query threshold at which a warning will be logged.
+    # "cached_replica_rows_fail_threshold" is the per-query threshold at which the query will fail.
+    #
+    # These thresholds may also be adjusted at runtime using the StorageService mbean.
+    #
+    # If the failure threshold is breached, it is likely that either the current page/fetch size
+    # is too large or one or more replicas is severely out-of-sync and in need of repair.
+    cached_rows_warn_threshold: 2000
+    cached_rows_fail_threshold: 32000
+
 # Log WARN on any multiple-partition batch size exceeding this value. 5kb per batch by default.
 # Caution should be taken on increasing the size of this threshold as it can lead to node instability.
 batch_size_warn_threshold_in_kb: 5

doc/source/operating/metrics.rst

@@ -79,76 +79,80 @@ Reported name format:
     **all** tables and keyspaces on the node.
 
 
-======================================= ============== ===========
-Name                                    Type           Description
-======================================= ============== ===========
-MemtableOnHeapSize                      Gauge<Long>    Total amount of data stored in the memtable that resides **on**-heap, including column related overhead and partitions overwritten.
-MemtableOffHeapSize                     Gauge<Long>    Total amount of data stored in the memtable that resides **off**-heap, including column related overhead and partitions overwritten.
-MemtableLiveDataSize                    Gauge<Long>    Total amount of live data stored in the memtable, excluding any data structure overhead.
-AllMemtablesOnHeapSize                  Gauge<Long>    Total amount of data stored in the memtables (2i and pending flush memtables included) that resides **on**-heap.
-AllMemtablesOffHeapSize                 Gauge<Long>    Total amount of data stored in the memtables (2i and pending flush memtables included) that resides **off**-heap.
-AllMemtablesLiveDataSize                Gauge<Long>    Total amount of live data stored in the memtables (2i and pending flush memtables included) that resides off-heap, excluding any data structure overhead.
-MemtableColumnsCount                    Gauge<Long>    Total number of columns present in the memtable.
-MemtableSwitchCount                     Counter        Number of times flush has resulted in the memtable being switched out.
-CompressionRatio                        Gauge<Double>  Current compression ratio for all SSTables.
-EstimatedPartitionSizeHistogram         Gauge<long[]>  Histogram of estimated partition size (in bytes).
-EstimatedPartitionCount                 Gauge<Long>    Approximate number of keys in table.
-EstimatedColumnCountHistogram           Gauge<long[]>  Histogram of estimated number of columns.
-SSTablesPerReadHistogram                Histogram      Histogram of the number of sstable data files accessed per single partition read. SSTables skipped due to Bloom Filters, min-max key or partition index lookup are not taken into acoount.
-ReadLatency                             Latency        Local read latency for this table.
-RangeLatency                            Latency        Local range scan latency for this table.
-WriteLatency                            Latency        Local write latency for this table.
-CoordinatorReadLatency                  Timer          Coordinator read latency for this table.
-CoordinatorWriteLatency                 Timer          Coordinator write latency for this table.
-CoordinatorScanLatency                  Timer          Coordinator range scan latency for this table.
-PendingFlushes                          Counter        Estimated number of flush tasks pending for this table.
-BytesFlushed                            Counter        Total number of bytes flushed since server [re]start.
-CompactionBytesWritten                  Counter        Total number of bytes written by compaction since server [re]start.
-PendingCompactions                      Gauge<Integer> Estimate of number of pending compactions for this table.
-LiveSSTableCount                        Gauge<Integer> Number of SSTables on disk for this table.
-LiveDiskSpaceUsed                       Counter        Disk space used by SSTables belonging to this table (in bytes).
-TotalDiskSpaceUsed                      Counter        Total disk space used by SSTables belonging to this table, including obsolete ones waiting to be GC'd.
-MinPartitionSize                        Gauge<Long>    Size of the smallest compacted partition (in bytes).
-MaxPartitionSize                        Gauge<Long>    Size of the largest compacted partition (in bytes).
-MeanPartitionSize                       Gauge<Long>    Size of the average compacted partition (in bytes).
-BloomFilterFalsePositives               Gauge<Long>    Number of false positives on table's bloom filter.
-BloomFilterFalseRatio                   Gauge<Double>  False positive ratio of table's bloom filter.
-BloomFilterDiskSpaceUsed                Gauge<Long>    Disk space used by bloom filter (in bytes).
-BloomFilterOffHeapMemoryUsed            Gauge<Long>    Off-heap memory used by bloom filter.
-IndexSummaryOffHeapMemoryUsed           Gauge<Long>    Off-heap memory used by index summary.
-CompressionMetadataOffHeapMemoryUsed    Gauge<Long>    Off-heap memory used by compression meta data.
-KeyCacheHitRate                         Gauge<Double>  Key cache hit rate for this table.
-TombstoneScannedHistogram               Histogram      Histogram of tombstones scanned in queries on this table.
-LiveScannedHistogram                    Histogram      Histogram of live cells scanned in queries on this table.
-ColUpdateTimeDeltaHistogram             Histogram      Histogram of column update time delta on this table.
-ViewLockAcquireTime                     Timer          Time taken acquiring a partition lock for materialized view updates on this table.
-ViewReadTime                            Timer          Time taken during the local read of a materialized view update.
-TrueSnapshotsSize                       Gauge<Long>    Disk space used by snapshots of this table including all SSTable components.
-RowCacheHitOutOfRange                   Counter        Number of table row cache hits that do not satisfy the query filter, thus went to disk.
-RowCacheHit                             Counter        Number of table row cache hits.
-RowCacheMiss                            Counter        Number of table row cache misses.
-CasPrepare                              Latency        Latency of paxos prepare round.
-CasPropose                              Latency        Latency of paxos propose round.
-CasCommit                               Latency        Latency of paxos commit round.
-PercentRepaired                         Gauge<Double>  Percent of table data that is repaired on disk.
-BytesRepaired                           Gauge<Long>    Size of table data repaired on disk
-BytesUnrepaired                         Gauge<Long>    Size of table data unrepaired on disk
-BytesPendingRepair                      Gauge<Long>    Size of table data isolated for an ongoing incremental repair
-SpeculativeRetries                      Counter        Number of times speculative retries were sent for this table.
-SpeculativeFailedRetries                Counter        Number of speculative retries that failed to prevent a timeout
-SpeculativeInsufficientReplicas         Counter        Number of speculative retries that couldn't be attempted due to lack of replicas
-SpeculativeSampleLatencyNanos           Gauge<Long>    Number of nanoseconds to wait before speculation is attempted. Value may be statically configured or updated periodically based on coordinator latency.
-WaitingOnFreeMemtableSpace              Histogram      Histogram of time spent waiting for free memtable space, either on- or off-heap.
-DroppedMutations                        Counter        Number of dropped mutations on this table.
-AnticompactionTime                      Timer          Time spent anticompacting before a consistent repair.
-ValidationTime                          Timer          Time spent doing validation compaction during repair.
-SyncTime                                Timer          Time spent doing streaming during repair.
-BytesValidated                          Histogram      Histogram over the amount of bytes read during validation.
-PartitionsValidated                     Histogram      Histogram over the number of partitions read during validation.
-BytesAnticompacted                      Counter        How many bytes we anticompacted.
-BytesMutatedAnticompaction              Counter        How many bytes we avoided anticompacting because the sstable was fully contained in the repaired range.
-MutatedAnticompactionGauge              Gauge<Double>  Ratio of bytes mutated vs total bytes repaired.
-======================================= ============== ===========
+=============================================== ============== ===========
+Name                                            Type           Description
+=============================================== ============== ===========
+MemtableOnHeapSize                              Gauge<Long>    Total amount of data stored in the memtable that resides **on**-heap, including column related overhead and partitions overwritten.
+MemtableOffHeapSize                             Gauge<Long>    Total amount of data stored in the memtable that resides **off**-heap, including column related overhead and partitions overwritten.
+MemtableLiveDataSize                            Gauge<Long>    Total amount of live data stored in the memtable, excluding any data structure overhead.
+AllMemtablesOnHeapSize                          Gauge<Long>    Total amount of data stored in the memtables (2i and pending flush memtables included) that resides **on**-heap.
+AllMemtablesOffHeapSize                         Gauge<Long>    Total amount of data stored in the memtables (2i and pending flush memtables included) that resides **off**-heap.
+AllMemtablesLiveDataSize                        Gauge<Long>    Total amount of live data stored in the memtables (2i and pending flush memtables included) that resides off-heap, excluding any data structure overhead.
+MemtableColumnsCount                            Gauge<Long>    Total number of columns present in the memtable.
+MemtableSwitchCount                             Counter        Number of times flush has resulted in the memtable being switched out.
+CompressionRatio                                Gauge<Double>  Current compression ratio for all SSTables.
+EstimatedPartitionSizeHistogram                 Gauge<long[]>  Histogram of estimated partition size (in bytes).
+EstimatedPartitionCount                         Gauge<Long>    Approximate number of keys in table.
+EstimatedColumnCountHistogram                   Gauge<long[]>  Histogram of estimated number of columns.
+SSTablesPerReadHistogram                        Histogram      Histogram of the number of sstable data files accessed per single partition read. SSTables skipped due to Bloom Filters, min-max key or partition index lookup are not taken into acoount.
+ReadLatency                                     Latency        Local read latency for this table.
+RangeLatency                                    Latency        Local range scan latency for this table.
+WriteLatency                                    Latency        Local write latency for this table.
+CoordinatorReadLatency                          Timer          Coordinator read latency for this table.
+CoordinatorWriteLatency                         Timer          Coordinator write latency for this table.
+CoordinatorScanLatency                          Timer          Coordinator range scan latency for this table.
+PendingFlushes                                  Counter        Estimated number of flush tasks pending for this table.
+BytesFlushed                                    Counter        Total number of bytes flushed since server [re]start.
+CompactionBytesWritten                          Counter        Total number of bytes written by compaction since server [re]start.
+PendingCompactions                              Gauge<Integer> Estimate of number of pending compactions for this table.
+LiveSSTableCount                                Gauge<Integer> Number of SSTables on disk for this table.
+LiveDiskSpaceUsed                               Counter        Disk space used by SSTables belonging to this table (in bytes).
+TotalDiskSpaceUsed                              Counter        Total disk space used by SSTables belonging to this table, including obsolete ones waiting to be GC'd.
+MinPartitionSize                                Gauge<Long>    Size of the smallest compacted partition (in bytes).
+MaxPartitionSize                                Gauge<Long>    Size of the largest compacted partition (in bytes).
+MeanPartitionSize                               Gauge<Long>    Size of the average compacted partition (in bytes).
+BloomFilterFalsePositives                       Gauge<Long>    Number of false positives on table's bloom filter.
+BloomFilterFalseRatio                           Gauge<Double>  False positive ratio of table's bloom filter.
+BloomFilterDiskSpaceUsed                        Gauge<Long>    Disk space used by bloom filter (in bytes).
+BloomFilterOffHeapMemoryUsed                    Gauge<Long>    Off-heap memory used by bloom filter.
+IndexSummaryOffHeapMemoryUsed                   Gauge<Long>    Off-heap memory used by index summary.
+CompressionMetadataOffHeapMemoryUsed            Gauge<Long>    Off-heap memory used by compression meta data.
+KeyCacheHitRate                                 Gauge<Double>  Key cache hit rate for this table.
+TombstoneScannedHistogram                       Histogram      Histogram of tombstones scanned in queries on this table.
+LiveScannedHistogram                            Histogram      Histogram of live cells scanned in queries on this table.
+ColUpdateTimeDeltaHistogram                     Histogram      Histogram of column update time delta on this table.
+ViewLockAcquireTime                             Timer          Time taken acquiring a partition lock for materialized view updates on this table.
+ViewReadTime                                    Timer          Time taken during the local read of a materialized view update.
+TrueSnapshotsSize                               Gauge<Long>    Disk space used by snapshots of this table including all SSTable components.
+RowCacheHitOutOfRange                           Counter        Number of table row cache hits that do not satisfy the query filter, thus went to disk.
+RowCacheHit                                     Counter        Number of table row cache hits.
+RowCacheMiss                                    Counter        Number of table row cache misses.
+CasPrepare                                      Latency        Latency of paxos prepare round.
+CasPropose                                      Latency        Latency of paxos propose round.
+CasCommit                                       Latency        Latency of paxos commit round.
+PercentRepaired                                 Gauge<Double>  Percent of table data that is repaired on disk.
+BytesRepaired                                   Gauge<Long>    Size of table data repaired on disk
+BytesUnrepaired                                 Gauge<Long>    Size of table data unrepaired on disk
+BytesPendingRepair                              Gauge<Long>    Size of table data isolated for an ongoing incremental repair
+SpeculativeRetries                              Counter        Number of times speculative retries were sent for this table.
+SpeculativeFailedRetries                        Counter        Number of speculative retries that failed to prevent a timeout
+SpeculativeInsufficientReplicas                 Counter        Number of speculative retries that couldn't be attempted due to lack of replicas
+SpeculativeSampleLatencyNanos                   Gauge<Long>    Number of nanoseconds to wait before speculation is attempted. Value may be statically configured or updated periodically based on coordinator latency.
+WaitingOnFreeMemtableSpace                      Histogram      Histogram of time spent waiting for free memtable space, either on- or off-heap.
+DroppedMutations                                Counter        Number of dropped mutations on this table.
+AnticompactionTime                              Timer          Time spent anticompacting before a consistent repair.
+ValidationTime                                  Timer          Time spent doing validation compaction during repair.
+SyncTime                                        Timer          Time spent doing streaming during repair.
+BytesValidated                                  Histogram      Histogram over the amount of bytes read during validation.
+PartitionsValidated                             Histogram      Histogram over the number of partitions read during validation.
+BytesAnticompacted                              Counter        How many bytes we anticompacted.
+BytesMutatedAnticompaction                      Counter        How many bytes we avoided anticompacting because the sstable was fully contained in the repaired range.
+MutatedAnticompactionGauge                      Gauge<Double>  Ratio of bytes mutated vs total bytes repaired.
+ReadRepairRequests                              Meter          Throughput for mutations generated by read-repair.
+ShortReadProtectionRequests                     Meter          Throughput for requests to get extra rows during short read protection.
+ReplicaFilteringProtectionRequests              Meter          Throughput for row completion requests during replica filtering protection.
+ReplicaFilteringProtectionRowsCachedPerQuery    Histogram      Histogram of the number of rows cached per query when replica filtering protection is engaged.
+============================================    ============== ===========
 
 Keyspace Metrics
 ^^^^^^^^^^^^^^^^

src/java/org/apache/cassandra/config/Config.java

@@ -330,6 +330,8 @@ public class Config
     public volatile int tombstone_warn_threshold = 1000;
     public volatile int tombstone_failure_threshold = 100000;
 
+    public final ReplicaFilteringProtectionOptions replica_filtering_protection = new ReplicaFilteringProtectionOptions();
+
     public volatile Long index_summary_capacity_in_mb;
     public volatile int index_summary_resize_interval_in_minutes = 60;
 

src/java/org/apache/cassandra/config/DatabaseDescriptor.java

@@ -1863,6 +1863,26 @@ public class DatabaseDescriptor
         conf.tombstone_failure_threshold = threshold;
     }
 
+    public static int getCachedReplicaRowsWarnThreshold()
+    {
+        return conf.replica_filtering_protection.cached_rows_warn_threshold;
+    }
+
+    public static void setCachedReplicaRowsWarnThreshold(int threshold)
+    {
+        conf.replica_filtering_protection.cached_rows_warn_threshold = threshold;
+    }
+
+    public static int getCachedReplicaRowsFailThreshold()
+    {
+        return conf.replica_filtering_protection.cached_rows_fail_threshold;
+    }
+
+    public static void setCachedReplicaRowsFailThreshold(int threshold)
+    {
+        conf.replica_filtering_protection.cached_rows_fail_threshold = threshold;
+    }
+
     /**
      * size of commitlog segments to allocate
      */

src/java/org/apache/cassandra/config/ReplicaFilteringProtectionOptions.java

@@ -0,0 +1,28 @@
+/*
+ * 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.config;
+
+public class ReplicaFilteringProtectionOptions
+{
+    public static final int DEFAULT_WARN_THRESHOLD = 2000;
+    public static final int DEFAULT_FAIL_THRESHOLD = 32000;
+
+    public volatile int cached_rows_warn_threshold = DEFAULT_WARN_THRESHOLD;
+    public volatile int cached_rows_fail_threshold = DEFAULT_FAIL_THRESHOLD;
+}

src/java/org/apache/cassandra/db/partitions/PartitionIterators.java

@@ -95,6 +95,21 @@ public abstract class PartitionIterators
         }
     }
 
+    /**
+     * Consumes all rows in the next partition of the provided partition iterator.
+     */
+    public static void consumeNext(PartitionIterator iterator)
+    {
+        if (iterator.hasNext())
+        {
+            try (RowIterator partition = iterator.next())
+            {
+                while (partition.hasNext())
+                    partition.next();
+            }
+        }
+    }
+
     /**
      * Wraps the provided iterator so it logs the returned rows for debugging purposes.
      * <p>
@@ -114,6 +129,38 @@ public abstract class PartitionIterators
         return Transformation.apply(iterator, new Logger());
     }
 
+    /**
+     * Wraps the provided iterator to run a specified action on close. Note that the action will be
+     * run even if closure of the provided iterator throws an exception.
+     */
+    public static PartitionIterator doOnClose(PartitionIterator delegate, Runnable action)
+    {
+        return new PartitionIterator()
+        {
+            public void close()
+            {
+                try
+                {
+                    delegate.close();
+                }
+                finally
+                {
+                    action.run();
+                }
+            }
+
+            public boolean hasNext()
+            {
+                return delegate.hasNext();
+            }
+
+            public RowIterator next()
+            {
+                return delegate.next();
+            }
+        };
+    }
+
     private static class SingletonPartitionIterator extends AbstractIterator<RowIterator> implements PartitionIterator
     {
         private final RowIterator iterator;

src/java/org/apache/cassandra/metrics/TableMetrics.java

@@ -328,7 +328,16 @@ public class TableMetrics
 
     public final Meter readRepairRequests;
     public final Meter shortReadProtectionRequests;
-    public final Meter replicaSideFilteringProtectionRequests;
+    
+    public final Meter replicaFilteringProtectionRequests;
+    
+    /**
+     * This histogram records the maximum number of rows {@link org.apache.cassandra.service.ReplicaFilteringProtection}
+     * caches at a point in time per query. With no replica divergence, this is equivalent to the maximum number of
+     * cached rows in a single partition during a query. It can be helpful when choosing appropriate values for the
+     * replica_filtering_protection thresholds in cassandra.yaml. 
+     */
+    public final Histogram rfpRowsCachedPerQuery;
 
     public final EnumMap<SamplerType, Sampler<?>> samplers;
     /**
@@ -942,7 +951,8 @@ public class TableMetrics
 
         readRepairRequests = createTableMeter("ReadRepairRequests");
         shortReadProtectionRequests = createTableMeter("ShortReadProtectionRequests");
-        replicaSideFilteringProtectionRequests = createTableMeter("ReplicaSideFilteringProtectionRequests");
+        replicaFilteringProtectionRequests = createTableMeter("ReplicaFilteringProtectionRequests");
+        rfpRowsCachedPerQuery = createHistogram("ReplicaFilteringProtectionRowsCachedPerQuery", true);
 
         confirmedRepairedInconsistencies = createTableMeter("RepairedDataInconsistenciesConfirmed", cfs.keyspace.metric.confirmedRepairedInconsistencies);
         unconfirmedRepairedInconsistencies = createTableMeter("RepairedDataInconsistenciesUnconfirmed", cfs.keyspace.metric.unconfirmedRepairedInconsistencies);
@@ -1060,6 +1070,13 @@ public class TableMetrics
         register(name, alias, tableMeter);
         return tableMeter;
     }
+    
+    private Histogram createHistogram(String name, boolean considerZeroes)
+    {
+        Histogram histogram = Metrics.histogram(factory.createMetricName(name), aliasFactory.createMetricName(name), considerZeroes);
+        register(name, name, histogram);
+        return histogram;
+    }
 
     /**
      * Computes the compression ratio for the specified SSTables

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

@@ -5387,6 +5387,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
     public void setTombstoneWarnThreshold(int threshold)
     {
         DatabaseDescriptor.setTombstoneWarnThreshold(threshold);
+        logger.info("updated tombstone_warn_threshold to {}", threshold);
     }
 
     public int getTombstoneFailureThreshold()
@@ -5397,6 +5398,29 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
     public void setTombstoneFailureThreshold(int threshold)
     {
         DatabaseDescriptor.setTombstoneFailureThreshold(threshold);
+        logger.info("updated tombstone_failure_threshold to {}", threshold);
+    }
+
+    public int getCachedReplicaRowsWarnThreshold()
+    {
+        return DatabaseDescriptor.getCachedReplicaRowsWarnThreshold();
+    }
+
+    public void setCachedReplicaRowsWarnThreshold(int threshold)
+    {
+        DatabaseDescriptor.setCachedReplicaRowsWarnThreshold(threshold);
+        logger.info("updated replica_filtering_protection.cached_rows_warn_threshold to {}", threshold);
+    }
+
+    public int getCachedReplicaRowsFailThreshold()
+    {
+        return DatabaseDescriptor.getCachedReplicaRowsFailThreshold();
+    }
+
+    public void setCachedReplicaRowsFailThreshold(int threshold)
+    {
+        DatabaseDescriptor.setCachedReplicaRowsFailThreshold(threshold);
+        logger.info("updated replica_filtering_protection.cached_rows_fail_threshold to {}", threshold);
     }
 
     public int getColumnIndexCacheSize()
@@ -5418,7 +5442,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
     public void setBatchSizeFailureThreshold(int threshold)
     {
         DatabaseDescriptor.setBatchSizeFailThresholdInKB(threshold);
-        logger.info("Updated batch_size_fail_threshold_in_kb to {}", threshold);
+        logger.info("updated batch_size_fail_threshold_in_kb to {}", threshold);
     }
 
     public int getBatchSizeWarnThreshold()
@@ -5470,7 +5494,7 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
     public void setHintedHandoffThrottleInKB(int throttleInKB)
     {
         DatabaseDescriptor.setHintedHandoffThrottleInKB(throttleInKB);
-        logger.info("Updated hinted_handoff_throttle_in_kb to {}", throttleInKB);
+        logger.info("updated hinted_handoff_throttle_in_kb to {}", throttleInKB);
     }
 
     @Override

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

@@ -697,6 +697,18 @@ public interface StorageServiceMBean extends NotificationEmitter
     /** Sets the threshold for abandoning queries with many tombstones */
     public void setTombstoneFailureThreshold(int tombstoneDebugThreshold);
 
+    /** Returns the number of rows cached at the coordinator before filtering/index queries log a warning. */
+    public int getCachedReplicaRowsWarnThreshold();
+
+    /** Sets the number of rows cached at the coordinator before filtering/index queries log a warning. */
+    public void setCachedReplicaRowsWarnThreshold(int threshold);
+
+    /** Returns the number of rows cached at the coordinator before filtering/index queries fail outright. */
+    public int getCachedReplicaRowsFailThreshold();
+
+    /** Sets the number of rows cached at the coordinator before filtering/index queries fail outright. */
+    public void setCachedReplicaRowsFailThreshold(int threshold);
+
     /** Returns the threshold for skipping the column index when caching partition info **/
     public int getColumnIndexCacheSize();
     /** Sets the threshold for skipping the column index when caching partition info **/

src/java/org/apache/cassandra/service/reads/DataResolver.java

@@ -25,6 +25,7 @@ import java.util.function.UnaryOperator;
 
 import com.google.common.base.Joiner;
 
+import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.cql3.statements.schema.IndexTarget;
 import org.apache.cassandra.db.DecoratedKey;
 import org.apache.cassandra.db.DeletionTime;
@@ -171,6 +172,7 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
 
         return context.needShortReadProtection()
                ? ShortReadProtection.extend(context.replicas.get(i),
+                                            () -> responses.clearUnsafe(i),
                                             originalResponse,
                                             command,
                                             context.mergedResultCounter,
@@ -199,20 +201,20 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
     {
         // Protecting against inconsistent replica filtering (some replica returning a row that is outdated but that
         // wouldn't be removed by normal reconciliation because up-to-date replica have filtered the up-to-date version
-        // of that row) works in 3 steps:
-        //   1) we read the full response just to collect rows that may be outdated (the ones we got from some
-        //      replica but didn't got any response for other; it could be those other replica have filtered a more
-        //      up-to-date result). In doing so, we do not count any of such "potentially outdated" row towards the
-        //      query limit. This simulate the worst case scenario where all those "potentially outdated" rows are
-        //      indeed outdated, and thus make sure we are guaranteed to read enough results (thanks to short read
-        //      protection).
-        //   2) we query all the replica/rows we need to rule out whether those "potentially outdated" rows are outdated
-        //      or not.
-        //   3) we re-read cached copies of each replica response using the "normal" read path merge with read-repair,
-        //      but where for each replica we use their original response _plus_ the additional rows queried in the
-        //      previous step (and apply the command#rowFilter() on the full result). Since the first phase has
-        //      pessimistically collected enough results for the case where all potentially outdated results are indeed
-        //      outdated, we shouldn't need further short-read protection requests during this phase.
+        // of that row) involves 3 main elements:
+        //   1) We combine short-read protection and a merge listener that identifies potentially "out-of-date"
+        //      rows to create an iterator that is guaranteed to produce enough valid row results to satisfy the query
+        //      limit if enough actually exist. A row is considered out-of-date if its merged from is non-empty and we
+        //      receive not response from at least one replica. In this case, it is possible that filtering at the
+        //      "silent" replica has produced a more up-to-date result.
+        //   2) This iterator is passed to the standard resolution process with read-repair, but is first wrapped in a
+        //      response provider that lazily "completes" potentially out-of-date rows by directly querying them on the
+        //      replicas that were previously silent. As this iterator is consumed, it caches valid data for potentially
+        //      out-of-date rows, and this cached data is merged with the fetched data as rows are requested. If there
+        //      is no replica divergence, only rows in the partition being evalutated will be cached (then released
+        //      when the partition is consumed).
+        //   3) After a "complete" row is materialized, it must pass the row filter supplied by the original query
+        //      before it counts against the limit.
 
         // We need separate contexts, as each context has his own counter
         ResolveContext firstPhaseContext = new ResolveContext(replicas);
@@ -221,26 +223,22 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
                                                                              command,
                                                                              replicaPlan().consistencyLevel(),
                                                                              queryStartNanoTime,
-                                                                             firstPhaseContext.replicas);
+                                                                             firstPhaseContext.replicas,
+                                                                             DatabaseDescriptor.getCachedReplicaRowsWarnThreshold(),
+                                                                             DatabaseDescriptor.getCachedReplicaRowsFailThreshold());
+
         PartitionIterator firstPhasePartitions = resolveInternal(firstPhaseContext,
                                                                  rfp.mergeController(),
                                                                  i -> shortReadProtectedResponse(i, firstPhaseContext),
                                                                  UnaryOperator.identity());
 
-        // Consume the first phase partitions to populate the replica filtering protection with both those materialized
-        // partitions and the primary keys to be fetched.
-        PartitionIterators.consume(firstPhasePartitions);
-        firstPhasePartitions.close();
+        PartitionIterator completedPartitions = resolveWithReadRepair(secondPhaseContext,
+                                                                      i -> rfp.queryProtectedPartitions(firstPhasePartitions, i),
+                                                                      results -> command.rowFilter().filter(results, command.metadata(), command.nowInSec()),
+                                                                      repairedDataTracker);
 
-        // After reading the entire query results the protection helper should have cached all the partitions so we can
-        // clear the responses accumulator for the sake of memory usage, given that the second phase might take long if
-        // it needs to query replicas.
-        responses.clearUnsafe();
-
-        return resolveWithReadRepair(secondPhaseContext,
-                                     rfp::queryProtectedPartitions,
-                                     results -> command.rowFilter().filter(results, command.metadata(), command.nowInSec()),
-                                     repairedDataTracker);
+        // Ensure that the RFP instance has a chance to record metrics when the iterator closes.
+        return PartitionIterators.doOnClose(completedPartitions, firstPhasePartitions::close);
     }
 
     @SuppressWarnings("resource")
@@ -269,7 +267,8 @@ public class DataResolver<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<
          */
 
         UnfilteredPartitionIterator merged = UnfilteredPartitionIterators.merge(results, mergeListener);
-        FilteredPartitions filtered = FilteredPartitions.filter(merged, new Filter(command.nowInSec(), command.metadata().enforceStrictLiveness()));
+        Filter filter = new Filter(command.nowInSec(), command.metadata().enforceStrictLiveness());
+        FilteredPartitions filtered = FilteredPartitions.filter(merged, filter);
         PartitionIterator counted = Transformation.apply(preCountFilter.apply(filtered), context.mergedResultCounter);
         return Transformation.apply(counted, new EmptyPartitionsDiscarder());
     }

src/java/org/apache/cassandra/service/reads/ReplicaFilteringProtection.java

@@ -18,14 +18,14 @@
 
 package org.apache.cassandra.service.reads;
 
+import java.util.ArrayDeque;
 import java.util.ArrayList;
 import java.util.Arrays;
-import java.util.Iterator;
 import java.util.List;
 import java.util.NavigableSet;
-import java.util.SortedMap;
-import java.util.TreeMap;
-import java.util.stream.Collectors;
+import java.util.concurrent.TimeUnit;
+import java.util.Queue;
+import java.util.function.Function;
 
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -45,6 +45,8 @@ import org.apache.cassandra.db.filter.ClusteringIndexFilter;
 import org.apache.cassandra.db.filter.ClusteringIndexNamesFilter;
 import org.apache.cassandra.db.filter.DataLimits;
 import org.apache.cassandra.db.filter.RowFilter;
+import org.apache.cassandra.db.partitions.PartitionIterator;
+import org.apache.cassandra.db.partitions.PartitionIterators;
 import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
 import org.apache.cassandra.db.partitions.UnfilteredPartitionIterators;
 import org.apache.cassandra.db.rows.EncodingStats;
@@ -54,18 +56,22 @@ import org.apache.cassandra.db.rows.Rows;
 import org.apache.cassandra.db.rows.Unfiltered;
 import org.apache.cassandra.db.rows.UnfilteredRowIterator;
 import org.apache.cassandra.db.rows.UnfilteredRowIterators;
+import org.apache.cassandra.exceptions.OverloadedException;
 import org.apache.cassandra.exceptions.ReadTimeoutException;
 import org.apache.cassandra.exceptions.UnavailableException;
 import org.apache.cassandra.locator.Endpoints;
+import org.apache.cassandra.locator.EndpointsForToken;
 import org.apache.cassandra.locator.Replica;
 import org.apache.cassandra.locator.ReplicaPlan;
 import org.apache.cassandra.locator.ReplicaPlans;
 import org.apache.cassandra.metrics.TableMetrics;
 import org.apache.cassandra.net.MessagingService;
 import org.apache.cassandra.schema.TableMetadata;
+import org.apache.cassandra.service.ClientWarn;
 import org.apache.cassandra.service.StorageProxy;
 import org.apache.cassandra.service.reads.repair.NoopReadRepair;
 import org.apache.cassandra.tracing.Tracing;
+import org.apache.cassandra.utils.NoSpamLogger;
 import org.apache.cassandra.utils.btree.BTreeSet;
 
 /**
@@ -77,11 +83,15 @@ import org.apache.cassandra.utils.btree.BTreeSet;
  * the rows in a replica response that don't have a corresponding row in other replica responses, and requests them by
  * primary key to the "silent" replicas in a second fetch round.
  * <p>
- * See CASSANDRA-8272 and CASSANDRA-8273 for further details.
+ * See CASSANDRA-8272, CASSANDRA-8273, and CASSANDRA-15907 for further details.
  */
 class ReplicaFilteringProtection<E extends Endpoints<E>>
 {
     private static final Logger logger = LoggerFactory.getLogger(ReplicaFilteringProtection.class);
+    private static final NoSpamLogger oneMinuteLogger = NoSpamLogger.getLogger(logger, 1, TimeUnit.MINUTES);
+
+    private static final Function<UnfilteredRowIterator, EncodingStats> NULL_TO_NO_STATS =
+        rowIterator -> rowIterator == null ? EncodingStats.NO_STATS : rowIterator.stats();
 
     private final Keyspace keyspace;
     private final ReadCommand command;
@@ -90,117 +100,53 @@ class ReplicaFilteringProtection<E extends Endpoints<E>>
     private final E sources;
     private final TableMetrics tableMetrics;
 
-    /**
-     * Per-source primary keys of the rows that might be outdated so they need to be fetched.
-     * For outdated static rows we use an empty builder to signal it has to be queried.
-     */
-    private final List<SortedMap<DecoratedKey, BTreeSet.Builder<Clustering>>> rowsToFetch;
+    private final int cachedRowsWarnThreshold;
+    private final int cachedRowsFailThreshold;
+
+    /** Tracks whether or not we've already hit the warning threshold while evaluating a partition. */
+    private boolean hitWarningThreshold = false;
+
+    private int currentRowsCached = 0; // tracks the current number of cached rows
+    private int maxRowsCached = 0; // tracks the high watermark for the number of cached rows
 
     /**
-     * Per-source list of all the partitions seen by the merge listener, to be merged with the extra fetched rows.
+     * Per-source list of the pending partitions seen by the merge listener, to be merged with the extra fetched rows.
      */
-    private final List<List<PartitionBuilder>> originalPartitions;
+    private final List<Queue<PartitionBuilder>> originalPartitions;
 
     ReplicaFilteringProtection(Keyspace keyspace,
                                ReadCommand command,
                                ConsistencyLevel consistency,
                                long queryStartNanoTime,
-                               E sources)
+                               E sources,
+                               int cachedRowsWarnThreshold,
+                               int cachedRowsFailThreshold)
     {
         this.keyspace = keyspace;
         this.command = command;
         this.consistency = consistency;
         this.queryStartNanoTime = queryStartNanoTime;
         this.sources = sources;
-        this.rowsToFetch = new ArrayList<>(sources.size());
         this.originalPartitions = new ArrayList<>(sources.size());
 
-        for (Replica ignored : sources)
+        for (int i = 0; i < sources.size(); i++)
         {
-            rowsToFetch.add(new TreeMap<>());
-            originalPartitions.add(new ArrayList<>());
+            originalPartitions.add(new ArrayDeque<>());
         }
 
         tableMetrics = ColumnFamilyStore.metricsFor(command.metadata().id);
+
+        this.cachedRowsWarnThreshold = cachedRowsWarnThreshold;
+        this.cachedRowsFailThreshold = cachedRowsFailThreshold;
     }
 
-    private BTreeSet.Builder<Clustering> getOrCreateToFetch(int source, DecoratedKey partitionKey)
+    private UnfilteredPartitionIterator executeReadCommand(ReadCommand cmd, Replica source, ReplicaPlan.Shared<EndpointsForToken, ReplicaPlan.ForTokenRead> replicaPlan)
     {
-        return rowsToFetch.get(source).computeIfAbsent(partitionKey, k -> BTreeSet.builder(command.metadata().comparator));
-    }
+        @SuppressWarnings("unchecked")
+        DataResolver<EndpointsForToken, ReplicaPlan.ForTokenRead> resolver =
+            new DataResolver<>(cmd, replicaPlan, (NoopReadRepair<EndpointsForToken, ReplicaPlan.ForTokenRead>) NoopReadRepair.instance, queryStartNanoTime);
 
-    /**
-     * Returns the protected results for the specified replica. These are generated fetching the extra rows and merging
-     * them with the cached original filtered results for that replica.
-     *
-     * @param source the source
-     * @return the protected results for the specified replica
-     */
-    UnfilteredPartitionIterator queryProtectedPartitions(int source)
-    {
-        UnfilteredPartitionIterator original = makeIterator(originalPartitions.get(source));
-        SortedMap<DecoratedKey, BTreeSet.Builder<Clustering>> toFetch = rowsToFetch.get(source);
-
-        if (toFetch.isEmpty())
-            return original;
-
-        // TODO: this would be more efficient if we had multi-key queries internally
-        List<UnfilteredPartitionIterator> fetched = toFetch.keySet()
-                                                           .stream()
-                                                           .map(k -> querySourceOnKey(source, k))
-                                                           .collect(Collectors.toList());
-
-        return UnfilteredPartitionIterators.merge(Arrays.asList(original, UnfilteredPartitionIterators.concat(fetched)), null);
-    }
-
-    private UnfilteredPartitionIterator querySourceOnKey(int i, DecoratedKey key)
-    {
-        BTreeSet.Builder<Clustering> builder = rowsToFetch.get(i).get(key);
-        assert builder != null; // We're calling this on the result of rowsToFetch.get(i).keySet()
-
-        Replica source = sources.get(i);
-        NavigableSet<Clustering> clusterings = builder.build();
-        tableMetrics.replicaSideFilteringProtectionRequests.mark();
-        if (logger.isTraceEnabled())
-            logger.trace("Requesting rows {} in partition {} from {} for replica-side filtering protection",
-                         clusterings, key, source);
-        Tracing.trace("Requesting {} rows in partition {} from {} for replica-side filtering protection",
-                      clusterings.size(), key, source);
-
-        // build the read command taking into account that we could be requesting only in the static row
-        DataLimits limits = clusterings.isEmpty() ? DataLimits.cqlLimits(1) : DataLimits.NONE;
-        ClusteringIndexFilter filter = new ClusteringIndexNamesFilter(clusterings, command.isReversed());
-        SinglePartitionReadCommand cmd = SinglePartitionReadCommand.create(command.metadata(),
-                                                                           command.nowInSec(),
-                                                                           command.columnFilter(),
-                                                                           RowFilter.NONE,
-                                                                           limits,
-                                                                           key,
-                                                                           filter);
-
-        ReplicaPlan.ForTokenRead replicaPlan = ReplicaPlans.forSingleReplicaRead(keyspace, key.getToken(), source);
-        ReplicaPlan.SharedForTokenRead sharedReplicaPlan = ReplicaPlan.shared(replicaPlan);
-        try
-        {
-            return executeReadCommand(cmd, source, sharedReplicaPlan);
-        }
-        catch (ReadTimeoutException e)
-        {
-            int blockFor = consistency.blockFor(keyspace);
-            throw new ReadTimeoutException(consistency, blockFor - 1, blockFor, true);
-        }
-        catch (UnavailableException e)
-        {
-            int blockFor = consistency.blockFor(keyspace);
-            throw UnavailableException.create(consistency, blockFor, blockFor - 1);
-        }
-    }
-
-    private <E extends Endpoints<E>, P extends ReplicaPlan.ForRead<E>>
-    UnfilteredPartitionIterator executeReadCommand(ReadCommand cmd, Replica source, ReplicaPlan.Shared<E, P> replicaPlan)
-    {
-        DataResolver<E, P> resolver = new DataResolver<>(cmd, replicaPlan, (NoopReadRepair<E, P>)NoopReadRepair.instance, queryStartNanoTime);
-        ReadCallback<E, P> handler = new ReadCallback<>(resolver, cmd, replicaPlan, queryStartNanoTime);
+        ReadCallback<EndpointsForToken, ReplicaPlan.ForTokenRead> handler = new ReadCallback<>(resolver, cmd, replicaPlan, queryStartNanoTime);
 
         if (source.isSelf())
         {
@@ -227,83 +173,124 @@ class ReplicaFilteringProtection<E extends Endpoints<E>>
      * <p>
      * The listener will track both the accepted data and the primary keys of the rows that are considered as outdated.
      * That way, once the query results would have been merged using this listener, further calls to
-     * {@link #queryProtectedPartitions(int)} will use the collected data to return a copy of the
+     * {@link #queryProtectedPartitions(PartitionIterator, int)} will use the collected data to return a copy of the
      * data originally collected from the specified replica, completed with the potentially outdated rows.
      */
     UnfilteredPartitionIterators.MergeListener mergeController()
     {
-        return (partitionKey, versions) -> {
-
-            PartitionBuilder[] builders = new PartitionBuilder[sources.size()];
-
-            for (int i = 0; i < sources.size(); i++)
-                builders[i] = new PartitionBuilder(command, partitionKey, columns(versions), stats(versions));
-
-            return new UnfilteredRowIterators.MergeListener()
+        return new UnfilteredPartitionIterators.MergeListener()
+        {
+            @Override
+            public void close()
             {
-                @Override
-                public void onMergedPartitionLevelDeletion(DeletionTime mergedDeletion, DeletionTime[] versions)
+                // If we hit the failure threshold before consuming a single partition, record the current rows cached.
+                tableMetrics.rfpRowsCachedPerQuery.update(Math.max(currentRowsCached, maxRowsCached));
+            }
+
+            @Override
+            public UnfilteredRowIterators.MergeListener getRowMergeListener(DecoratedKey partitionKey, List<UnfilteredRowIterator> versions)
+            {
+                List<PartitionBuilder> builders = new ArrayList<>(sources.size());
+                RegularAndStaticColumns columns = columns(versions);
+                EncodingStats stats = EncodingStats.merge(versions, NULL_TO_NO_STATS);
+
+                for (int i = 0; i < sources.size(); i++)
+                    builders.add(i, new PartitionBuilder(partitionKey, sources.get(i), columns, stats));
+
+                return new UnfilteredRowIterators.MergeListener()
                 {
-                    // cache the deletion time versions to be able to regenerate the original row iterator
-                    for (int i = 0; i < versions.length; i++)
-                        builders[i].setDeletionTime(versions[i]);
-                }
-
-                @Override
-                public Row onMergedRows(Row merged, Row[] versions)
-                {
-                    // cache the row versions to be able to regenerate the original row iterator
-                    for (int i = 0; i < versions.length; i++)
-                        builders[i].addRow(versions[i]);
-
-                    if (merged.isEmpty())
-                        return merged;
-
-                    boolean isPotentiallyOutdated = false;
-                    boolean isStatic = merged.isStatic();
-                    for (int i = 0; i < versions.length; i++)
+                    @Override
+                    public void onMergedPartitionLevelDeletion(DeletionTime mergedDeletion, DeletionTime[] versions)
                     {
-                        Row version = versions[i];
-                        if (version == null || (isStatic && version.isEmpty()))
-                        {
-                            isPotentiallyOutdated = true;
-                            BTreeSet.Builder<Clustering> toFetch = getOrCreateToFetch(i, partitionKey);
-                            // Note that for static, we shouldn't add the clustering to the clustering set (the
-                            // ClusteringIndexNamesFilter we'll build from this later does not expect it), but the fact
-                            // we created a builder in the first place will act as a marker that the static row must be
-                            // fetched, even if no other rows are added for this partition.
-                            if (!isStatic)
-                                toFetch.add(merged.clustering());
-                        }
+                        // cache the deletion time versions to be able to regenerate the original row iterator
+                        for (int i = 0; i < versions.length; i++)
+                            builders.get(i).setDeletionTime(versions[i]);
                     }
 
-                    // If the row is potentially outdated (because some replica didn't send anything and so it _may_ be
-                    // an outdated result that is only present because other replica have filtered the up-to-date result
-                    // out), then we skip the row. In other words, the results of the initial merging of results by this
-                    // protection assume the worst case scenario where every row that might be outdated actually is.
-                    // This ensures that during this first phase (collecting additional row to fetch) we are guaranteed
-                    // to look at enough data to ultimately fulfill the query limit.
-                    return isPotentiallyOutdated ? null : merged;
-                }
+                    @Override
+                    public Row onMergedRows(Row merged, Row[] versions)
+                    {
+                        // cache the row versions to be able to regenerate the original row iterator
+                        for (int i = 0; i < versions.length; i++)
+                            builders.get(i).addRow(versions[i]);
 
-                @Override
-                public void onMergedRangeTombstoneMarkers(RangeTombstoneMarker merged, RangeTombstoneMarker[] versions)
-                {
-                    // cache the marker versions to be able to regenerate the original row iterator
-                    for (int i = 0; i < versions.length; i++)
-                        builders[i].addRangeTombstoneMarker(versions[i]);
-                }
+                        if (merged.isEmpty())
+                            return merged;
 
-                @Override
-                public void close()
-                {
-                    for (int i = 0; i < sources.size(); i++)
-                        originalPartitions.get(i).add(builders[i]);
-                }
-            };
+                        boolean isPotentiallyOutdated = false;
+                        boolean isStatic = merged.isStatic();
+                        for (int i = 0; i < versions.length; i++)
+                        {
+                            Row version = versions[i];
+                            if (version == null || (isStatic && version.isEmpty()))
+                            {
+                                isPotentiallyOutdated = true;
+                                builders.get(i).addToFetch(merged);
+                            }
+                        }
+
+                        // If the row is potentially outdated (because some replica didn't send anything and so it _may_ be
+                        // an outdated result that is only present because other replica have filtered the up-to-date result
+                        // out), then we skip the row. In other words, the results of the initial merging of results by this
+                        // protection assume the worst case scenario where every row that might be outdated actually is.
+                        // This ensures that during this first phase (collecting additional row to fetch) we are guaranteed
+                        // to look at enough data to ultimately fulfill the query limit.
+                        return isPotentiallyOutdated ? null : merged;
+                    }
+
+                    @Override
+                    public void onMergedRangeTombstoneMarkers(RangeTombstoneMarker merged, RangeTombstoneMarker[] versions)
+                    {
+                        // cache the marker versions to be able to regenerate the original row iterator
+                        for (int i = 0; i < versions.length; i++)
+                            builders.get(i).addRangeTombstoneMarker(versions[i]);
+                    }
+
+                    @Override
+                    public void close()
+                    {
+                        for (int i = 0; i < sources.size(); i++)
+                            originalPartitions.get(i).add(builders.get(i));
+                    }
+                };
+            }
         };
     }
 
+    private void incrementCachedRows()
+    {
+        currentRowsCached++;
+
+        if (currentRowsCached == cachedRowsFailThreshold + 1)
+        {
+            String message = String.format("Replica filtering protection has cached over %d rows during query %s. " +
+                                           "(See 'cached_replica_rows_fail_threshold' in cassandra.yaml.)",
+                                           cachedRowsFailThreshold, command.toCQLString());
+
+            logger.error(message);
+            Tracing.trace(message);
+            throw new OverloadedException(message);
+        }
+        else if (currentRowsCached == cachedRowsWarnThreshold + 1 && !hitWarningThreshold)
+        {
+            hitWarningThreshold = true;
+
+            String message = String.format("Replica filtering protection has cached over %d rows during query %s. " +
+                                           "(See 'cached_replica_rows_warn_threshold' in cassandra.yaml.)",
+                                           cachedRowsWarnThreshold, command.toCQLString());
+
+            ClientWarn.instance.warn(message);
+            oneMinuteLogger.warn(message);
+            Tracing.trace(message);
+        }
+    }
+
+    private void releaseCachedRows(int count)
+    {
+        maxRowsCached = Math.max(maxRowsCached, currentRowsCached);
+        currentRowsCached -= count;
+    }
+
     private static RegularAndStaticColumns columns(List<UnfilteredRowIterator> versions)
     {
         Columns statics = Columns.NONE;
@@ -320,24 +307,19 @@ class ReplicaFilteringProtection<E extends Endpoints<E>>
         return new RegularAndStaticColumns(statics, regulars);
     }
 
-    private static EncodingStats stats(List<UnfilteredRowIterator> iterators)
-    {
-        EncodingStats stats = EncodingStats.NO_STATS;
-        for (UnfilteredRowIterator iter : iterators)
-        {
-            if (iter == null)
-                continue;
-
-            stats = stats.mergeWith(iter.stats());
-        }
-        return stats;
-    }
-
-    private UnfilteredPartitionIterator makeIterator(List<PartitionBuilder> builders)
+    /**
+     * Returns the protected results for the specified replica. These are generated fetching the extra rows and merging
+     * them with the cached original filtered results for that replica.
+     *
+     * @param merged the first iteration partitions, that should have been read used with the {@link #mergeController()}
+     * @param source the source
+     * @return the protected results for the specified replica
+     */
+    UnfilteredPartitionIterator queryProtectedPartitions(PartitionIterator merged, int source)
     {
         return new UnfilteredPartitionIterator()
         {
-            final Iterator<PartitionBuilder> iterator = builders.iterator();
+            final Queue<PartitionBuilder> partitions = originalPartitions.get(source);
 
             @Override
             public TableMetadata metadata()
@@ -346,43 +328,49 @@ class ReplicaFilteringProtection<E extends Endpoints<E>>
             }
 
             @Override
-            public void close()
-            {
-                // nothing to do here
-            }
+            public void close() { }
 
             @Override
             public boolean hasNext()
             {
-                return iterator.hasNext();
+                // If there are no cached partition builders for this source, advance the first phase iterator, which
+                // will force the RFP merge listener to load at least the next protected partition. Note that this may
+                // load more than one partition if any divergence between replicas is discovered by the merge listener.
+                if (partitions.isEmpty())
+                {
+                    PartitionIterators.consumeNext(merged);
+                }
+
+                return !partitions.isEmpty();
             }
 
             @Override
             public UnfilteredRowIterator next()
             {
-                return iterator.next().build();
+                PartitionBuilder builder = partitions.poll();
+                assert builder != null;
+                return builder.protectedPartition();
             }
         };
     }
 
-    private static class PartitionBuilder
+    private class PartitionBuilder
     {
-        private final ReadCommand command;
-        private final DecoratedKey partitionKey;
+        private final DecoratedKey key;
+        private final Replica source;
         private final RegularAndStaticColumns columns;
         private final EncodingStats stats;
 
         private DeletionTime deletionTime;
         private Row staticRow = Rows.EMPTY_STATIC_ROW;
-        private final List<Unfiltered> contents = new ArrayList<>();
+        private final Queue<Unfiltered> contents = new ArrayDeque<>();
+        private BTreeSet.Builder<Clustering> toFetch;
+        private int partitionRowsCached;
 
-        private PartitionBuilder(ReadCommand command,
-                                 DecoratedKey partitionKey,
-                                 RegularAndStaticColumns columns,
-                                 EncodingStats stats)
+        private PartitionBuilder(DecoratedKey key, Replica source, RegularAndStaticColumns columns, EncodingStats stats)
         {
-            this.command = command;
-            this.partitionKey = partitionKey;
+            this.key = key;
+            this.source = source;
             this.columns = columns;
             this.stats = stats;
         }
@@ -394,6 +382,12 @@ class ReplicaFilteringProtection<E extends Endpoints<E>>
 
         private void addRow(Row row)
         {
+            partitionRowsCached++;
+
+            incrementCachedRows();
+
+            // Note that even null rows are counted against the row caching limit. The assumption is that
+            // a subsequent protection query will later fetch the row onto the heap anyway.
             if (row == null)
                 return;
 
@@ -409,12 +403,23 @@ class ReplicaFilteringProtection<E extends Endpoints<E>>
                 contents.add(marker);
         }
 
-        private UnfilteredRowIterator build()
+        private void addToFetch(Row row)
+        {
+            if (toFetch == null)
+                toFetch = BTreeSet.builder(command.metadata().comparator);
+
+            // Note that for static, we shouldn't add the clustering to the clustering set (the
+            // ClusteringIndexNamesFilter we'll build from this later does not expect it), but the fact
+            // we created a builder in the first place will act as a marker that the static row must be
+            // fetched, even if no other rows are added for this partition.
+            if (!row.isStatic())
+                toFetch.add(row.clustering());
+        }
+
+        private UnfilteredRowIterator originalPartition()
         {
             return new UnfilteredRowIterator()
             {
-                final Iterator<Unfiltered> iterator = contents.iterator();
-
                 @Override
                 public DeletionTime partitionLevelDeletion()
                 {
@@ -448,7 +453,7 @@ class ReplicaFilteringProtection<E extends Endpoints<E>>
                 @Override
                 public DecoratedKey partitionKey()
                 {
-                    return partitionKey;
+                    return key;
                 }
 
                 @Override
@@ -460,21 +465,86 @@ class ReplicaFilteringProtection<E extends Endpoints<E>>
                 @Override
                 public void close()
                 {
-                    // nothing to do here
+                    releaseCachedRows(partitionRowsCached);
                 }
 
                 @Override
                 public boolean hasNext()
                 {
-                    return iterator.hasNext();
+                    return !contents.isEmpty();
                 }
 
                 @Override
                 public Unfiltered next()
                 {
-                    return iterator.next();
+                    return contents.poll();
                 }
             };
         }
+
+        private UnfilteredRowIterator protectedPartition()
+        {
+            UnfilteredRowIterator original = originalPartition();
+
+            if (toFetch != null)
+            {
+                try (UnfilteredPartitionIterator partitions = fetchFromSource())
+                {
+                    if (partitions.hasNext())
+                    {
+                        try (UnfilteredRowIterator fetchedRows = partitions.next())
+                        {
+                            return UnfilteredRowIterators.merge(Arrays.asList(original, fetchedRows));
+                        }
+                    }
+                }
+            }
+
+            return original;
+        }
+
+        private UnfilteredPartitionIterator fetchFromSource()
+        {
+            assert toFetch != null;
+
+            NavigableSet<Clustering> clusterings = toFetch.build();
+            tableMetrics.replicaFilteringProtectionRequests.mark();
+
+            if (logger.isTraceEnabled())
+                logger.trace("Requesting rows {} in partition {} from {} for replica filtering protection",
+                             clusterings, key, source);
+
+            Tracing.trace("Requesting {} rows in partition {} from {} for replica filtering protection",
+                          clusterings.size(), key, source);
+
+            // build the read command taking into account that we could be requesting only in the static row
+            DataLimits limits = clusterings.isEmpty() ? DataLimits.cqlLimits(1) : DataLimits.NONE;
+            ClusteringIndexFilter filter = new ClusteringIndexNamesFilter(clusterings, command.isReversed());
+            SinglePartitionReadCommand cmd = SinglePartitionReadCommand.create(command.metadata(),
+                                                                               command.nowInSec(),
+                                                                               command.columnFilter(),
+                                                                               RowFilter.NONE,
+                                                                               limits,
+                                                                               key,
+                                                                               filter);
+
+            ReplicaPlan.ForTokenRead replicaPlan = ReplicaPlans.forSingleReplicaRead(keyspace, key.getToken(), source);
+            ReplicaPlan.SharedForTokenRead sharedReplicaPlan = ReplicaPlan.shared(replicaPlan);
+
+            try
+            {
+                return executeReadCommand(cmd, source, sharedReplicaPlan);
+            }
+            catch (ReadTimeoutException e)
+            {
+                int blockFor = consistency.blockFor(keyspace);
+                throw new ReadTimeoutException(consistency, blockFor - 1, blockFor, true);
+            }
+            catch (UnavailableException e)
+            {
+                int blockFor = consistency.blockFor(keyspace);
+                throw UnavailableException.create(consistency, blockFor, blockFor - 1);
+            }
+        }
     }
 }

src/java/org/apache/cassandra/service/reads/ShortReadPartitionsProtection.java

@@ -53,6 +53,8 @@ public class ShortReadPartitionsProtection extends Transformation<UnfilteredRowI
     private final ReadCommand command;
     private final Replica source;
 
+    private final Runnable preFetchCallback; // called immediately before fetching more contents
+
     private final DataLimits.Counter singleResultCounter; // unmerged per-source counter
     private final DataLimits.Counter mergedResultCounter; // merged end-result counter
 
@@ -62,13 +64,16 @@ public class ShortReadPartitionsProtection extends Transformation<UnfilteredRowI
 
     private final long queryStartNanoTime;
 
-    public ShortReadPartitionsProtection(ReadCommand command, Replica source,
+    public ShortReadPartitionsProtection(ReadCommand command,
+                                         Replica source,
+                                         Runnable preFetchCallback,
                                          DataLimits.Counter singleResultCounter,
                                          DataLimits.Counter mergedResultCounter,
                                          long queryStartNanoTime)
     {
         this.command = command;
         this.source = source;
+        this.preFetchCallback = preFetchCallback;
         this.singleResultCounter = singleResultCounter;
         this.mergedResultCounter = mergedResultCounter;
         this.queryStartNanoTime = queryStartNanoTime;
@@ -147,6 +152,9 @@ public class ShortReadPartitionsProtection extends Transformation<UnfilteredRowI
         Tracing.trace("Requesting {} extra rows from {} for short read protection", toQuery, source);
         logger.info("Requesting {} extra rows from {} for short read protection", toQuery, source);
 
+        // If we've arrived here, all responses have been consumed, and we're about to request more.
+        preFetchCallback.run();
+
         return makeAndExecuteFetchAdditionalPartitionReadCommand(toQuery);
     }
 

src/java/org/apache/cassandra/service/reads/ShortReadProtection.java

@@ -42,16 +42,25 @@ import org.apache.cassandra.locator.Replica;
 public class ShortReadProtection
 {
     @SuppressWarnings("resource")
-    public static UnfilteredPartitionIterator extend(Replica source, UnfilteredPartitionIterator partitions,
-                                                     ReadCommand command, DataLimits.Counter mergedResultCounter,
-                                                     long queryStartNanoTime, boolean enforceStrictLiveness)
+    public static UnfilteredPartitionIterator extend(Replica source,
+                                                     Runnable preFetchCallback,
+                                                     UnfilteredPartitionIterator partitions,
+                                                     ReadCommand command,
+                                                     DataLimits.Counter mergedResultCounter,
+                                                     long queryStartNanoTime,
+                                                     boolean enforceStrictLiveness)
     {
         DataLimits.Counter singleResultCounter = command.limits().newCounter(command.nowInSec(),
                                                                              false,
                                                                              command.selectsFullPartition(),
                                                                              enforceStrictLiveness).onlyCount();
 
-        ShortReadPartitionsProtection protection = new ShortReadPartitionsProtection(command, source, singleResultCounter, mergedResultCounter, queryStartNanoTime);
+        ShortReadPartitionsProtection protection = new ShortReadPartitionsProtection(command,
+                                                                                     source,
+                                                                                     preFetchCallback,
+                                                                                     singleResultCounter,
+                                                                                     mergedResultCounter,
+                                                                                     queryStartNanoTime);
 
         /*
          * The order of extention and transformations is important here. Extending with more partitions has to happen

src/java/org/apache/cassandra/service/reads/repair/NoopReadRepair.java

@@ -34,8 +34,7 @@ import org.apache.cassandra.service.reads.DigestResolver;
 /**
  * Bypasses the read repair path for short read protection and testing
  */
-public class NoopReadRepair<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<E>>
-        implements ReadRepair<E, P>
+public class NoopReadRepair<E extends Endpoints<E>, P extends ReplicaPlan.ForRead<E>> implements ReadRepair<E, P>
 {
     public static final NoopReadRepair instance = new NoopReadRepair();
 

src/java/org/apache/cassandra/transport/Message.java

@@ -299,7 +299,7 @@ public abstract class Message
             return tracingId;
         }
 
-        Message setWarnings(List<String> warnings)
+        public Message setWarnings(List<String> warnings)
         {
             this.warnings = warnings;
             return this;

src/java/org/apache/cassandra/utils/concurrent/Accumulator.java

@@ -20,7 +20,6 @@ package org.apache.cassandra.utils.concurrent;
 
 import java.util.AbstractCollection;
 import java.util.Collection;
-import java.util.Arrays;
 import java.util.Iterator;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 
@@ -159,14 +158,12 @@ public class Accumulator<E>
     }
 
     /**
-     * Removes all of the elements from this accumulator.
+     * Removes element at the speficied index from this accumulator.
      *
      * This method is not thread-safe when used concurrently with {@link #add(Object)}.
      */
-    public void clearUnsafe()
+    public void clearUnsafe(int i)
     {
-        nextIndexUpdater.set(this, 0);
-        presentCountUpdater.set(this, 0);
-        Arrays.fill(values, null);
+        values[i] = null;
     }
 }

test/distributed/org/apache/cassandra/distributed/impl/Coordinator.java

@@ -39,6 +39,7 @@ import org.apache.cassandra.distributed.api.QueryResult;
 import org.apache.cassandra.distributed.api.QueryResults;
 import org.apache.cassandra.distributed.api.SimpleQueryResult;
 import org.apache.cassandra.service.ClientState;
+import org.apache.cassandra.service.ClientWarn;
 import org.apache.cassandra.service.QueryState;
 import org.apache.cassandra.service.pager.QueryPager;
 import org.apache.cassandra.transport.ProtocolVersion;
@@ -91,6 +92,11 @@ public class Coordinator implements ICoordinator
             boundBBValues.add(ByteBufferUtil.objectToBytes(boundValue));
 
         prepared.validate(QueryState.forInternalCalls().getClientState());
+
+        // Start capturing warnings on this thread. Note that this will implicitly clear out any previous 
+        // warnings as it sets a new State instance on the ThreadLocal.
+        ClientWarn.instance.captureWarnings();
+        
         ResultMessage res = prepared.execute(QueryState.forInternalCalls(),
                                              QueryOptions.create(toCassandraCL(consistencyLevel),
                                                                  boundBBValues,
@@ -102,6 +108,10 @@ public class Coordinator implements ICoordinator
                                                                  null),
                                              System.nanoTime());
 
+        // Collect warnings reported during the query.
+        if (res != null)
+            res.setWarnings(ClientWarn.instance.getWarnings());
+
         return RowUtil.toQueryResult(res);
     }
 

test/distributed/org/apache/cassandra/distributed/impl/RowUtil.java

@@ -19,6 +19,7 @@
 package org.apache.cassandra.distributed.impl;
 
 import java.nio.ByteBuffer;
+import java.util.Collections;
 import java.util.Iterator;
 import java.util.List;
 
@@ -41,7 +42,11 @@ public class RowUtil
             ResultMessage.Rows rows = (ResultMessage.Rows) res;
             String[] names = getColumnNames(rows.result.metadata.names);
             Object[][] results = RowUtil.toObjects(rows);
-            return new SimpleQueryResult(names, results);
+            
+            // Warnings may be null here, due to ClientWarn#getWarnings() handling of empty warning lists.
+            List<String> warnings = res.getWarnings();
+
+            return new SimpleQueryResult(names, results, warnings == null ? Collections.emptyList() : warnings);
         }
         else
         {

test/distributed/org/apache/cassandra/distributed/test/ReplicaFilteringProtectionTest.java

@@ -0,0 +1,244 @@
+/*
+ * 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.distributed.test;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.db.Keyspace;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.IInvokableInstance;
+import org.apache.cassandra.distributed.api.SimpleQueryResult;
+import org.apache.cassandra.exceptions.OverloadedException;
+import org.apache.cassandra.service.StorageService;
+
+import static org.apache.cassandra.config.ReplicaFilteringProtectionOptions.DEFAULT_FAIL_THRESHOLD;
+import static org.apache.cassandra.config.ReplicaFilteringProtectionOptions.DEFAULT_WARN_THRESHOLD;
+import static org.apache.cassandra.distributed.api.ConsistencyLevel.ALL;
+import static org.apache.cassandra.distributed.shared.AssertUtils.assertRows;
+import static org.apache.cassandra.distributed.shared.AssertUtils.row;
+import static org.junit.Assert.assertEquals;
+
+/**
+ * Exercises the functionality of {@link org.apache.cassandra.service.reads.ReplicaFilteringProtection}, the
+ * mechanism that ensures distributed index and filtering queries at read consistency levels > ONE/LOCAL_ONE
+ * avoid stale replica results.
+ */
+public class ReplicaFilteringProtectionTest extends TestBaseImpl
+{
+    private static final int REPLICAS = 2;
+    private static final int ROWS = 3;
+
+    private static Cluster cluster;
+
+    @BeforeClass
+    public static void setup() throws IOException
+    {
+        cluster = init(Cluster.build()
+                              .withNodes(REPLICAS)
+                              .withConfig(config -> config.set("hinted_handoff_enabled", false)
+                                                          .set("commitlog_sync", "batch")
+                                                          .set("num_tokens", 1)).start());
+
+        // Make sure we start w/ the correct defaults:
+        cluster.get(1).runOnInstance(() -> assertEquals(DEFAULT_WARN_THRESHOLD, StorageService.instance.getCachedReplicaRowsWarnThreshold()));
+        cluster.get(1).runOnInstance(() -> assertEquals(DEFAULT_FAIL_THRESHOLD, StorageService.instance.getCachedReplicaRowsFailThreshold()));
+    }
+
+    @AfterClass
+    public static void teardown()
+    {
+        cluster.close();
+    }
+
+    @Test
+    public void testMissedUpdatesBelowCachingWarnThreshold()
+    {
+        String tableName = "missed_updates_no_warning";
+        cluster.schemaChange(withKeyspace("CREATE TABLE %s." + tableName + " (k int PRIMARY KEY, v text)"));
+
+        // The warning threshold provided is one more than the total number of rows returned
+        // to the coordinator from all replicas and therefore should not be triggered.
+        testMissedUpdates(tableName, REPLICAS * ROWS, Integer.MAX_VALUE, false);
+    }
+
+    @Test
+    public void testMissedUpdatesAboveCachingWarnThreshold()
+    {
+        String tableName = "missed_updates_cache_warn";
+        cluster.schemaChange(withKeyspace("CREATE TABLE %s." + tableName + " (k int PRIMARY KEY, v text)"));
+
+        // The warning threshold provided is one less than the total number of rows returned
+        // to the coordinator from all replicas and therefore should be triggered but not fail the query.
+        testMissedUpdates(tableName, REPLICAS * ROWS - 1, Integer.MAX_VALUE, true);
+    }
+
+    @Test
+    public void testMissedUpdatesAroundCachingFailThreshold()
+    {
+        String tableName = "missed_updates_cache_fail";
+        cluster.schemaChange(withKeyspace("CREATE TABLE %s." + tableName + " (k int PRIMARY KEY, v text)"));
+
+        // The failure threshold provided is exactly the total number of rows returned
+        // to the coordinator from all replicas and therefore should just warn.
+        testMissedUpdates(tableName, 1, REPLICAS * ROWS, true);
+
+        try
+        {
+            // The failure threshold provided is one less than the total number of rows returned
+            // to the coordinator from all replicas and therefore should fail the query.
+            testMissedUpdates(tableName, 1, REPLICAS * ROWS - 1, true);
+        }
+        catch (RuntimeException e)
+        {
+            assertEquals(e.getClass().getName(), OverloadedException.class.getName());
+        }
+    }
+
+    private void testMissedUpdates(String tableName, int warnThreshold, int failThreshold, boolean shouldWarn)
+    {
+        cluster.get(1).runOnInstance(() -> StorageService.instance.setCachedReplicaRowsWarnThreshold(warnThreshold));
+        cluster.get(1).runOnInstance(() -> StorageService.instance.setCachedReplicaRowsFailThreshold(failThreshold));
+
+        String fullTableName = KEYSPACE + '.' + tableName;
+
+        // Case 1: Insert and query rows at ALL to verify base line.
+        for (int i = 0; i < ROWS; i++)
+        {
+            cluster.coordinator(1).execute("INSERT INTO " + fullTableName + "(k, v) VALUES (?, 'old')", ALL, i);
+        }
+
+        long histogramSampleCount = rowsCachedPerQueryCount(cluster.get(1), tableName);
+
+        String query = "SELECT * FROM " + fullTableName + " WHERE v = ? LIMIT ? ALLOW FILTERING";
+
+        Object[][] initialRows = cluster.coordinator(1).execute(query, ALL, "old", ROWS);
+        assertRows(initialRows, row(1, "old"), row(0, "old"), row(2, "old"));
+
+        // Make sure only one sample was recorded for the query.
+        assertEquals(histogramSampleCount + 1, rowsCachedPerQueryCount(cluster.get(1), tableName));
+
+        // Case 2: Update all rows on only one replica, leaving the entire dataset of the remaining replica out-of-date.
+        updateAllRowsOn(1, fullTableName, "new");
+
+        // The replica that missed the results creates a mismatch at every row, and we therefore cache a version
+        // of that row for all replicas.
+        SimpleQueryResult oldResult = cluster.coordinator(1).executeWithResult(query, ALL, "old", ROWS);
+        assertRows(oldResult.toObjectArrays());
+        verifyWarningState(shouldWarn, oldResult);
+
+        // We should have made 3 row "completion" requests.
+        assertEquals(ROWS, protectionQueryCount(cluster.get(1), tableName));
+
+        // In all cases above, the queries should be caching 1 row per partition per replica, but
+        // 6 for the whole query, given every row is potentially stale.
+        assertEquals(ROWS * REPLICAS, maxRowsCachedPerQuery(cluster.get(1), tableName));
+
+        // Make sure only one more sample was recorded for the query.
+        assertEquals(histogramSampleCount + 2, rowsCachedPerQueryCount(cluster.get(1), tableName));
+
+        // Case 3: Observe the effects of blocking read-repair.
+
+        // The previous query peforms a blocking read-repair, which removes replica divergence. This
+        // will only warn, therefore, if the warning threshold is actually below the number of replicas.
+        // (i.e. The row cache counter is decremented/reset as each partition is consumed.)
+        SimpleQueryResult newResult = cluster.coordinator(1).executeWithResult(query, ALL, "new", ROWS);
+        Object[][] newRows = newResult.toObjectArrays();
+        assertRows(newRows, row(1, "new"), row(0, "new"), row(2, "new"));
+
+        verifyWarningState(warnThreshold < REPLICAS, newResult);
+
+        // We still sould only have made 3 row "completion" requests, with no replica divergence in the last query.
+        assertEquals(ROWS, protectionQueryCount(cluster.get(1), tableName));
+
+        // With no replica divergence, we only cache a single partition at a time across 2 replicas.
+        assertEquals(REPLICAS, minRowsCachedPerQuery(cluster.get(1), tableName));
+
+        // Make sure only one more sample was recorded for the query.
+        assertEquals(histogramSampleCount + 3, rowsCachedPerQueryCount(cluster.get(1), tableName));
+
+        // Case 4: Introduce another mismatch by updating all rows on only one replica.
+
+        updateAllRowsOn(1, fullTableName, "future");
+
+        // Another mismatch is introduced, and we once again cache a version of each row during resolution.
+        SimpleQueryResult futureResult = cluster.coordinator(1).executeWithResult(query, ALL, "future", ROWS);
+        Object[][] futureRows = futureResult.toObjectArrays();
+        assertRows(futureRows, row(1, "future"), row(0, "future"), row(2, "future"));
+
+        verifyWarningState(shouldWarn, futureResult);
+
+        // We sould have made 3 more row "completion" requests.
+        assertEquals(ROWS * 2, protectionQueryCount(cluster.get(1), tableName));
+
+        // In all cases above, the queries should be caching 1 row per partition, but 6 for the
+        // whole query, given every row is potentially stale.
+        assertEquals(ROWS * REPLICAS, maxRowsCachedPerQuery(cluster.get(1), tableName));
+
+        // Make sure only one more sample was recorded for the query.
+        assertEquals(histogramSampleCount + 4, rowsCachedPerQueryCount(cluster.get(1), tableName));
+    }
+
+    private void updateAllRowsOn(int node, String table, String value)
+    {
+        for (int i = 0; i < ROWS; i++)
+        {
+            cluster.get(node).executeInternal("UPDATE " + table + " SET v = ? WHERE k = ?", value, i);
+        }
+    }
+
+    private void verifyWarningState(boolean shouldWarn, SimpleQueryResult futureResult)
+    {
+        List<String> futureWarnings = futureResult.warnings();
+        assertEquals(shouldWarn, futureWarnings.stream().anyMatch(w -> w.contains("cached_replica_rows_warn_threshold")));
+        assertEquals(shouldWarn ? 1 : 0, futureWarnings.size());
+    }
+
+    private long protectionQueryCount(IInvokableInstance instance, String tableName)
+    {
+        return instance.callOnInstance(() -> Keyspace.open(KEYSPACE)
+                                                     .getColumnFamilyStore(tableName)
+                                                     .metric.replicaFilteringProtectionRequests.getCount());
+    }
+
+    private long maxRowsCachedPerQuery(IInvokableInstance instance, String tableName)
+    {
+        return instance.callOnInstance(() -> Keyspace.open(KEYSPACE)
+                                                     .getColumnFamilyStore(tableName)
+                                                     .metric.rfpRowsCachedPerQuery.getSnapshot().getMax());
+    }
+
+    private long minRowsCachedPerQuery(IInvokableInstance instance, String tableName)
+    {
+        return instance.callOnInstance(() -> Keyspace.open(KEYSPACE)
+                                                     .getColumnFamilyStore(tableName)
+                                                     .metric.rfpRowsCachedPerQuery.getSnapshot().getMin());
+    }
+
+    private long rowsCachedPerQueryCount(IInvokableInstance instance, String tableName)
+    {
+        return instance.callOnInstance(() -> Keyspace.open(KEYSPACE)
+                                                     .getColumnFamilyStore(tableName)
+                                                     .metric.rfpRowsCachedPerQuery.getCount());
+    }
+}

test/unit/org/apache/cassandra/utils/concurrent/AccumulatorTest.java

@@ -29,7 +29,7 @@ public class AccumulatorTest
     @Test
     public void testAddMoreThanCapacity()
     {
-        Accumulator<Integer> accu = new Accumulator(4);
+        Accumulator<Integer> accu = new Accumulator<>(4);
 
         accu.add(1);
         accu.add(2);
@@ -50,7 +50,7 @@ public class AccumulatorTest
     @Test
     public void testIsEmptyAndSize()
     {
-        Accumulator<Integer> accu = new Accumulator(4);
+        Accumulator<Integer> accu = new Accumulator<>(4);
 
         assertTrue(accu.isEmpty());
         assertEquals(0, accu.size());
@@ -58,20 +58,20 @@ public class AccumulatorTest
         accu.add(1);
         accu.add(2);
 
-        assertTrue(!accu.isEmpty());
+        assertFalse(accu.isEmpty());
         assertEquals(2, accu.size());
 
         accu.add(3);
         accu.add(4);
 
-        assertTrue(!accu.isEmpty());
+        assertFalse(accu.isEmpty());
         assertEquals(4, accu.size());
     }
 
     @Test
     public void testGetAndIterator()
     {
-        Accumulator<String> accu = new Accumulator(4);
+        Accumulator<String> accu = new Accumulator<>(4);
 
         accu.add("3");
         accu.add("2");
@@ -99,32 +99,32 @@ public class AccumulatorTest
     @Test
     public void testClearUnsafe()
     {
-        Accumulator<String> accu = new Accumulator<>(3);
+        Accumulator<String> accu = new Accumulator<>(5);
 
         accu.add("1");
         accu.add("2");
         accu.add("3");
 
-        accu.clearUnsafe();
+        accu.clearUnsafe(1);
 
-        assertEquals(0, accu.size());
-        assertFalse(accu.snapshot().iterator().hasNext());
-        assertOutOfBonds(accu, 0);
+        assertEquals(3, accu.size());
+        assertTrue(accu.snapshot().iterator().hasNext());
 
         accu.add("4");
         accu.add("5");
 
-        assertEquals(2, accu.size());
+        assertEquals(5, accu.size());
 
-        assertEquals("4", accu.get(0));
-        assertEquals("5", accu.get(1));
-        assertOutOfBonds(accu, 2);
+        assertEquals("4", accu.get(3));
+        assertEquals("5", accu.get(4));
+        assertOutOfBonds(accu, 5);
 
         Iterator<String> iter = accu.snapshot().iterator();
         assertTrue(iter.hasNext());
-        assertEquals("4", iter.next());
-        assertEquals("5", iter.next());
-        assertFalse(iter.hasNext());
+        assertEquals("1", iter.next());
+        assertNull(iter.next());
+        assertTrue(iter.hasNext());
+        assertEquals("3", iter.next());
     }
 
     private static void assertOutOfBonds(Accumulator<String> accumulator, int index)