Abort in-progress queries that time out

patch by Stefania; reviewed by aweisberg for CASSANDRA-7392

CHANGES.txt

@@ -1,4 +1,5 @@
 3.2
+ * Abort in-progress queries that time out (CASSANDRA-7392)
  * Add transparent data encryption core classes (CASSANDRA-9945)
 
 

src/java/org/apache/cassandra/concurrent/ScheduledExecutors.java

@@ -22,6 +22,11 @@ package org.apache.cassandra.concurrent;
  */
 public class ScheduledExecutors
 {
+    /**
+     * This pool is used for periodic fast (sub-microsecond) tasks.
+     */
+    public static final DebuggableScheduledThreadPoolExecutor scheduledFastTasks = new DebuggableScheduledThreadPoolExecutor("ScheduledFastTasks");
+
     /**
      * This pool is used for periodic short (sub-second) tasks.
      */

src/java/org/apache/cassandra/cql3/UntypedResultSet.java

@@ -187,7 +187,8 @@ public abstract class UntypedResultSet implements Iterable<UntypedResultSet.Row>
                         if (pager.isExhausted())
                             return endOfData();
 
-                        try (ReadOrderGroup orderGroup = pager.startOrderGroup(); PartitionIterator iter = pager.fetchPageInternal(pageSize, orderGroup))
+                        try (ReadExecutionController executionController = pager.executionController();
+                             PartitionIterator iter = pager.fetchPageInternal(pageSize, executionController))
                         {
                             currentPage = select.process(iter, nowInSec).rows.iterator();
                         }

src/java/org/apache/cassandra/cql3/statements/ModificationStatement.java

@@ -360,7 +360,8 @@ public abstract class ModificationStatement implements CQLStatement
 
         if (local)
         {
-            try (ReadOrderGroup orderGroup = group.startOrderGroup(); PartitionIterator iter = group.executeInternal(orderGroup))
+            try (ReadExecutionController executionController = group.executionController();
+                 PartitionIterator iter = group.executeInternal(executionController))
             {
                 return asMaterializedMap(iter);
             }
@@ -575,7 +576,8 @@ public abstract class ModificationStatement implements CQLStatement
 
         SinglePartitionReadCommand<?> readCommand = request.readCommand(FBUtilities.nowInSeconds());
         FilteredPartition current;
-        try (ReadOrderGroup orderGroup = readCommand.startOrderGroup(); PartitionIterator iter = readCommand.executeInternal(orderGroup))
+        try (ReadExecutionController executionController = readCommand.executionController();
+             PartitionIterator iter = readCommand.executeInternal(executionController))
         {
             current = FilteredPartition.create(PartitionIterators.getOnlyElement(iter, readCommand));
         }

src/java/org/apache/cassandra/cql3/statements/SelectStatement.java

@@ -253,9 +253,9 @@ public class SelectStatement implements CQLStatement
             this.pager = pager;
         }
 
-        public static Pager forInternalQuery(QueryPager pager, ReadOrderGroup orderGroup)
+        public static Pager forInternalQuery(QueryPager pager, ReadExecutionController executionController)
         {
-            return new InternalPager(pager, orderGroup);
+            return new InternalPager(pager, executionController);
         }
 
         public static Pager forDistributedQuery(QueryPager pager, ConsistencyLevel consistency, ClientState clientState)
@@ -295,17 +295,17 @@ public class SelectStatement implements CQLStatement
 
         public static class InternalPager extends Pager
         {
-            private final ReadOrderGroup orderGroup;
+            private final ReadExecutionController executionController;
 
-            private InternalPager(QueryPager pager, ReadOrderGroup orderGroup)
+            private InternalPager(QueryPager pager, ReadExecutionController executionController)
             {
                 super(pager);
-                this.orderGroup = orderGroup;
+                this.executionController = executionController;
             }
 
             public PartitionIterator fetchPage(int pageSize)
             {
-                return pager.fetchPageInternal(pageSize, orderGroup);
+                return pager.fetchPageInternal(pageSize, executionController);
             }
         }
     }
@@ -378,11 +378,11 @@ public class SelectStatement implements CQLStatement
         ReadQuery query = getQuery(options, nowInSec);
         int pageSize = getPageSize(options);
 
-        try (ReadOrderGroup orderGroup = query.startOrderGroup())
+        try (ReadExecutionController executionController = query.executionController())
         {
             if (pageSize <= 0 || query.limits().count() <= pageSize)
             {
-                try (PartitionIterator data = query.executeInternal(orderGroup))
+                try (PartitionIterator data = query.executeInternal(executionController))
                 {
                     return processResults(data, options, nowInSec);
                 }
@@ -390,7 +390,7 @@ public class SelectStatement implements CQLStatement
             else
             {
                 QueryPager pager = query.getPager(options.getPagingState(), options.getProtocolVersion());
-                return execute(Pager.forInternalQuery(pager, orderGroup), options, pageSize, nowInSec);
+                return execute(Pager.forInternalQuery(pager, executionController), options, pageSize, nowInSec);
             }
         }
     }

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

@@ -175,7 +175,7 @@ public class PartitionRangeReadCommand extends ReadCommand
         metric.rangeLatency.addNano(latencyNanos);
     }
 
-    protected UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadOrderGroup orderGroup)
+    protected UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadExecutionController executionController)
     {
         ColumnFamilyStore.ViewFragment view = cfs.select(View.select(SSTableSet.LIVE, dataRange().keyRange()));
         Tracing.trace("Executing seq scan across {} sstables for {}", view.sstables.size(), dataRange().keyRange().getString(metadata().getKeyValidator()));

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

@@ -28,6 +28,7 @@ import org.slf4j.LoggerFactory;
 import org.apache.cassandra.config.*;
 import org.apache.cassandra.cql3.Operator;
 import org.apache.cassandra.db.filter.*;
+import org.apache.cassandra.db.monitoring.MonitorableImpl;
 import org.apache.cassandra.db.partitions.*;
 import org.apache.cassandra.db.rows.*;
 import org.apache.cassandra.dht.AbstractBounds;
@@ -43,6 +44,7 @@ import org.apache.cassandra.schema.UnknownIndexException;
 import org.apache.cassandra.service.ClientWarn;
 import org.apache.cassandra.tracing.Tracing;
 import org.apache.cassandra.utils.ByteBufferUtil;
+import org.apache.cassandra.utils.FBUtilities;
 import org.apache.cassandra.utils.Pair;
 
 /**
@@ -51,10 +53,10 @@ import org.apache.cassandra.utils.Pair;
  * <p>
  * This contains all the informations needed to do a local read.
  */
-public abstract class ReadCommand implements ReadQuery
+public abstract class ReadCommand extends MonitorableImpl implements ReadQuery
 {
+    private static final int TEST_ITERATION_DELAY_MILLIS = Integer.valueOf(System.getProperty("cassandra.test.read_iteration_delay_ms", "0"));
     protected static final Logger logger = LoggerFactory.getLogger(ReadCommand.class);
-
     public static final IVersionedSerializer<ReadCommand> serializer = new Serializer();
     // For RANGE_SLICE verb: will either dispatch on 'serializer' for 3.0 or 'legacyRangeSliceCommandSerializer' for earlier version.
     // Can be removed (and replaced by 'serializer') once we drop pre-3.0 backward compatibility.
@@ -276,7 +278,7 @@ public abstract class ReadCommand implements ReadQuery
      */
     public abstract ReadCommand copy();
 
-    protected abstract UnfilteredPartitionIterator queryStorage(ColumnFamilyStore cfs, ReadOrderGroup orderGroup);
+    protected abstract UnfilteredPartitionIterator queryStorage(ColumnFamilyStore cfs, ReadExecutionController executionController);
 
     protected abstract int oldestUnrepairedTombstone();
 
@@ -321,13 +323,13 @@ public abstract class ReadCommand implements ReadQuery
     /**
      * Executes this command on the local host.
      *
-     * @param orderGroup the operation group spanning this command
+     * @param executionController the execution controller spanning this command
      *
      * @return an iterator over the result of executing this command locally.
      */
     @SuppressWarnings("resource") // The result iterator is closed upon exceptions (we know it's fine to potentially not close the intermediary
                                   // iterators created inside the try as long as we do close the original resultIterator), or by closing the result.
-    public UnfilteredPartitionIterator executeLocally(ReadOrderGroup orderGroup)
+    public UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController)
     {
         long startTimeNanos = System.nanoTime();
 
@@ -339,11 +341,12 @@ public abstract class ReadCommand implements ReadQuery
             Tracing.trace("Executing read on {}.{} using index {}", cfs.metadata.ksName, cfs.metadata.cfName, index.getIndexName());
 
         UnfilteredPartitionIterator resultIterator = searcher == null
-                                         ? queryStorage(cfs, orderGroup)
-                                         : searcher.search(orderGroup);
+                                         ? queryStorage(cfs, executionController)
+                                         : searcher.search(executionController);
 
         try
         {
+            resultIterator = withStateTracking(resultIterator);
             resultIterator = withMetricsRecording(withoutPurgeableTombstones(resultIterator, cfs), cfs.metric, startTimeNanos);
 
             // If we've used a 2ndary index, we know the result already satisfy the primary expression used, so
@@ -367,14 +370,14 @@ public abstract class ReadCommand implements ReadQuery
 
     protected abstract void recordLatency(TableMetrics metric, long latencyNanos);
 
-    public PartitionIterator executeInternal(ReadOrderGroup orderGroup)
+    public PartitionIterator executeInternal(ReadExecutionController controller)
     {
-        return UnfilteredPartitionIterators.filter(executeLocally(orderGroup), nowInSec());
+        return UnfilteredPartitionIterators.filter(executeLocally(controller), nowInSec());
     }
 
-    public ReadOrderGroup startOrderGroup()
+    public ReadExecutionController executionController()
     {
-        return ReadOrderGroup.forCommand(this);
+        return ReadExecutionController.forCommand(this);
     }
 
     /**
@@ -470,6 +473,48 @@ public abstract class ReadCommand implements ReadQuery
         };
     }
 
+    protected UnfilteredPartitionIterator withStateTracking(UnfilteredPartitionIterator iter)
+    {
+        return new WrappingUnfilteredPartitionIterator(iter)
+        {
+            @Override
+            public UnfilteredRowIterator computeNext(UnfilteredRowIterator iter)
+            {
+                if (isAborted())
+                    return null;
+
+                if (TEST_ITERATION_DELAY_MILLIS > 0)
+                    maybeDelayForTesting();
+
+                return iter;
+            }
+        };
+    }
+
+    protected UnfilteredRowIterator withStateTracking(UnfilteredRowIterator iter)
+    {
+        return new WrappingUnfilteredRowIterator(iter)
+        {
+            @Override
+            public boolean hasNext()
+            {
+                if (isAborted())
+                    return false;
+
+                if (TEST_ITERATION_DELAY_MILLIS > 0)
+                    maybeDelayForTesting();
+
+                return super.hasNext();
+            }
+        };
+    }
+
+    private void maybeDelayForTesting()
+    {
+        if (!metadata.ksName.startsWith("system"))
+            FBUtilities.sleepQuietly(TEST_ITERATION_DELAY_MILLIS);
+    }
+
     /**
      * Creates a message for this command.
      */
@@ -512,6 +557,12 @@ public abstract class ReadCommand implements ReadQuery
         return sb.toString();
     }
 
+    // Monitorable interface
+    public String name()
+    {
+        return toCQLString();
+    }
+
     private static class Serializer implements IVersionedSerializer<ReadCommand>
     {
         private static int digestFlag(boolean isDigest)

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

@@ -41,15 +41,24 @@ public class ReadCommandVerbHandler implements IVerbHandler<ReadCommand>
         }
 
         ReadCommand command = message.payload;
+        command.setMonitoringTime(message.constructionTime, message.getTimeout());
+
         ReadResponse response;
-        try (ReadOrderGroup opGroup = command.startOrderGroup(); UnfilteredPartitionIterator iterator = command.executeLocally(opGroup))
+        try (ReadExecutionController executionController = command.executionController();
+             UnfilteredPartitionIterator iterator = command.executeLocally(executionController))
         {
             response = command.createResponse(iterator, command.columnFilter());
         }
 
-        MessageOut<ReadResponse> reply = new MessageOut<>(MessagingService.Verb.REQUEST_RESPONSE, response, serializer());
+        if (!command.complete())
+        {
+            Tracing.trace("Discarding partial response to {} (timed out)", message.from);
+            MessagingService.instance().incrementDroppedMessages(message);
+            return;
+        }
 
         Tracing.trace("Enqueuing response to {}", message.from);
+        MessageOut<ReadResponse> reply = new MessageOut<>(MessagingService.Verb.REQUEST_RESPONSE, response, ReadResponse.serializer);
         MessagingService.instance().sendReply(reply, id, message.from);
     }
 }

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

@@ -20,7 +20,7 @@ package org.apache.cassandra.db;
 import org.apache.cassandra.index.Index;
 import org.apache.cassandra.utils.concurrent.OpOrder;
 
-public class ReadOrderGroup implements AutoCloseable
+public class ReadExecutionController implements AutoCloseable
 {
     // For every reads
     private final OpOrder.Group baseOp;
@@ -29,7 +29,7 @@ public class ReadOrderGroup implements AutoCloseable
     private final OpOrder.Group indexOp;
     private final OpOrder.Group writeOp;
 
-    private ReadOrderGroup(OpOrder.Group baseOp, OpOrder.Group indexOp, OpOrder.Group writeOp)
+    private ReadExecutionController(OpOrder.Group baseOp, OpOrder.Group indexOp, OpOrder.Group writeOp)
     {
         this.baseOp = baseOp;
         this.indexOp = indexOp;
@@ -51,19 +51,24 @@ public class ReadOrderGroup implements AutoCloseable
         return writeOp;
     }
 
-    public static ReadOrderGroup emptyGroup()
+    public static ReadExecutionController empty()
     {
-        return new ReadOrderGroup(null, null, null);
+        return new ReadExecutionController(null, null, null);
     }
 
-    public static ReadOrderGroup forCommand(ReadCommand command)
+    public static ReadExecutionController forReadOp(OpOrder.Group readOp)
+    {
+        return new ReadExecutionController(readOp, null, null);
+    }
+
+    public static ReadExecutionController forCommand(ReadCommand command)
     {
         ColumnFamilyStore baseCfs = Keyspace.openAndGetStore(command.metadata());
         ColumnFamilyStore indexCfs = maybeGetIndexCfs(baseCfs, command);
 
         if (indexCfs == null)
         {
-            return new ReadOrderGroup(baseCfs.readOrdering.start(), null, null);
+            return new ReadExecutionController(baseCfs.readOrdering.start(), null, null);
         }
         else
         {
@@ -76,7 +81,7 @@ public class ReadOrderGroup implements AutoCloseable
                 // TODO: this should perhaps not open and maintain a writeOp for the full duration, but instead only *try* to delete stale entries, without blocking if there's no room
                 // as it stands, we open a writeOp and keep it open for the duration to ensure that should this CF get flushed to make room we don't block the reclamation of any room being made
                 writeOp = baseCfs.keyspace.writeOrder.start();
-                return new ReadOrderGroup(baseOp, indexOp, writeOp);
+                return new ReadExecutionController(baseOp, indexOp, writeOp);
             }
             catch (RuntimeException e)
             {

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

@@ -35,9 +35,9 @@ public interface ReadQuery
 {
     ReadQuery EMPTY = new ReadQuery()
     {
-        public ReadOrderGroup startOrderGroup()
+        public ReadExecutionController executionController()
         {
-            return ReadOrderGroup.emptyGroup();
+            return ReadExecutionController.empty();
         }
 
         public PartitionIterator execute(ConsistencyLevel consistency, ClientState clientState) throws RequestExecutionException
@@ -45,7 +45,7 @@ public interface ReadQuery
             return PartitionIterators.EMPTY;
         }
 
-        public PartitionIterator executeInternal(ReadOrderGroup orderGroup)
+        public PartitionIterator executeInternal(ReadExecutionController controller)
         {
             return PartitionIterators.EMPTY;
         }
@@ -86,9 +86,9 @@ public interface ReadQuery
      * The returned object <b>must</b> be closed on all path and it is thus strongly advised to
      * use it in a try-with-ressource construction.
      *
-     * @return a newly started order group for this {@code ReadQuery}.
+     * @return a newly started execution controller for this {@code ReadQuery}.
      */
-    public ReadOrderGroup startOrderGroup();
+    public ReadExecutionController executionController();
 
     /**
      * Executes the query at the provided consistency level.
@@ -104,10 +104,10 @@ public interface ReadQuery
     /**
      * Execute the query for internal queries (that is, it basically executes the query locally).
      *
-     * @param orderGroup the {@code ReadOrderGroup} protecting the read.
+     * @param controller the {@code ReadExecutionController} protecting the read.
      * @return the result of the query.
      */
-    public PartitionIterator executeInternal(ReadOrderGroup orderGroup);
+    public PartitionIterator executeInternal(ReadExecutionController controller);
 
     /**
      * Returns a pager for the query.

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

@@ -181,7 +181,7 @@ public class SinglePartitionNamesCommand extends SinglePartitionReadCommand<Clus
             }
         }
 
-        return result.unfilteredIterator(columnFilter(), Slices.ALL, clusteringIndexFilter().isReversed());
+        return withStateTracking(result.unfilteredIterator(columnFilter(), Slices.ALL, clusteringIndexFilter().isReversed()));
     }
 
     private ImmutableBTreePartition add(UnfilteredRowIterator iter, ImmutableBTreePartition result, boolean isRepaired)

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

@@ -254,12 +254,12 @@ public abstract class SinglePartitionReadCommand<F extends ClusteringIndexFilter
         metric.readLatency.addNano(latencyNanos);
     }
 
-    protected UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadOrderGroup orderGroup)
+    protected UnfilteredPartitionIterator queryStorage(final ColumnFamilyStore cfs, ReadExecutionController executionController)
     {
         @SuppressWarnings("resource") // we close the created iterator through closing the result of this method (and SingletonUnfilteredPartitionIterator ctor cannot fail)
         UnfilteredRowIterator partition = cfs.isRowCacheEnabled()
-                                        ? getThroughCache(cfs, orderGroup.baseReadOpOrderGroup())
-                                        : queryMemtableAndDisk(cfs, orderGroup.baseReadOpOrderGroup());
+                                        ? getThroughCache(cfs, executionController)
+                                        : queryMemtableAndDisk(cfs, executionController);
         return new SingletonUnfilteredPartitionIterator(partition, isForThrift());
     }
 
@@ -272,7 +272,7 @@ public abstract class SinglePartitionReadCommand<F extends ClusteringIndexFilter
      * If the partition is is not cached, we figure out what filter is "biggest", read
      * that from disk, then filter the result and either cache that or return it.
      */
-    private UnfilteredRowIterator getThroughCache(ColumnFamilyStore cfs, OpOrder.Group readOp)
+    private UnfilteredRowIterator getThroughCache(ColumnFamilyStore cfs, ReadExecutionController executionController)
     {
         assert !cfs.isIndex(); // CASSANDRA-5732
         assert cfs.isRowCacheEnabled() : String.format("Row cache is not enabled on table [%s]", cfs.name);
@@ -290,7 +290,7 @@ public abstract class SinglePartitionReadCommand<F extends ClusteringIndexFilter
                 // Some other read is trying to cache the value, just do a normal non-caching read
                 Tracing.trace("Row cache miss (race)");
                 cfs.metric.rowCacheMiss.inc();
-                return queryMemtableAndDisk(cfs, readOp);
+                return queryMemtableAndDisk(cfs, executionController);
             }
 
             CachedPartition cachedPartition = (CachedPartition)cached;
@@ -303,7 +303,7 @@ public abstract class SinglePartitionReadCommand<F extends ClusteringIndexFilter
 
             cfs.metric.rowCacheHitOutOfRange.inc();
             Tracing.trace("Ignoring row cache as cached value could not satisfy query");
-            return queryMemtableAndDisk(cfs, readOp);
+            return queryMemtableAndDisk(cfs, executionController);
         }
 
         cfs.metric.rowCacheMiss.inc();
@@ -328,7 +328,7 @@ public abstract class SinglePartitionReadCommand<F extends ClusteringIndexFilter
             {
                 int rowsToCache = metadata().params.caching.rowsPerPartitionToCache();
                 @SuppressWarnings("resource") // we close on exception or upon closing the result of this method
-                UnfilteredRowIterator iter = SinglePartitionReadCommand.fullPartitionRead(metadata(), nowInSec(), partitionKey()).queryMemtableAndDisk(cfs, readOp);
+                UnfilteredRowIterator iter = SinglePartitionReadCommand.fullPartitionRead(metadata(), nowInSec(), partitionKey()).queryMemtableAndDisk(cfs, executionController);
                 try
                 {
                     // We want to cache only rowsToCache rows
@@ -368,7 +368,7 @@ public abstract class SinglePartitionReadCommand<F extends ClusteringIndexFilter
         }
 
         Tracing.trace("Fetching data but not populating cache as query does not query from the start of the partition");
-        return queryMemtableAndDisk(cfs, readOp);
+        return queryMemtableAndDisk(cfs, executionController);
     }
 
     /**
@@ -387,6 +387,11 @@ public abstract class SinglePartitionReadCommand<F extends ClusteringIndexFilter
      * to enforce that that it is required as parameter, even though it's not explicitlly used by the method.
      */
     public UnfilteredRowIterator queryMemtableAndDisk(ColumnFamilyStore cfs, OpOrder.Group readOp)
+    {
+        return queryMemtableAndDisk(cfs, ReadExecutionController.forReadOp(readOp));
+    }
+
+    public UnfilteredRowIterator queryMemtableAndDisk(ColumnFamilyStore cfs, ReadExecutionController executionController)
     {
         Tracing.trace("Executing single-partition query on {}", cfs.name);
 
@@ -487,18 +492,18 @@ public abstract class SinglePartitionReadCommand<F extends ClusteringIndexFilter
             return commands.get(0).metadata();
         }
 
-        public ReadOrderGroup startOrderGroup()
+        public ReadExecutionController executionController()
         {
             // Note that the only difference between the command in a group must be the partition key on which
             // they applied. So as far as ReadOrderGroup is concerned, we can use any of the commands to start one.
-            return commands.get(0).startOrderGroup();
+            return commands.get(0).executionController();
         }
 
-        public PartitionIterator executeInternal(ReadOrderGroup orderGroup)
+        public PartitionIterator executeInternal(ReadExecutionController controller)
         {
             List<PartitionIterator> partitions = new ArrayList<>(commands.size());
             for (SinglePartitionReadCommand cmd : commands)
-                partitions.add(cmd.executeInternal(orderGroup));
+                partitions.add(cmd.executeInternal(controller));
 
             // Because we only have enforce the limit per command, we need to enforce it globally.
             return limits.filter(PartitionIterators.concat(partitions), nowInSec);

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

@@ -247,7 +247,7 @@ public class SinglePartitionSliceCommand extends SinglePartitionReadCommand<Clus
                 cfs.metric.samplers.get(TableMetrics.Sampler.READS).addSample(key.getKey(), key.hashCode(), 1);
             }
 
-            return merged;
+            return withStateTracking(merged);
         }
         catch (RuntimeException | Error e)
         {

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

@@ -738,7 +738,7 @@ public abstract class Slices implements Iterable<Slice>
 
             public boolean isEQ()
             {
-                return startValue.equals(endValue);
+                return Objects.equals(startValue, endValue);
             }
         }
     }

src/java/org/apache/cassandra/db/filter/ClusteringIndexNamesFilter.java

@@ -217,7 +217,7 @@ public class ClusteringIndexNamesFilter extends AbstractClusteringIndexFilter
 
     public String toCQLString(CFMetaData metadata)
     {
-        if (clusterings.isEmpty())
+        if (metadata.clusteringColumns().isEmpty() || clusterings.size() <= 1)
             return "";
 
         StringBuilder sb = new StringBuilder();

src/java/org/apache/cassandra/db/filter/ColumnFilter.java

@@ -317,9 +317,12 @@ public class ColumnFilter
             return "<none>";
 
         StringBuilder sb = new StringBuilder();
-        appendColumnDef(sb, defs.next());
         while (defs.hasNext())
-            appendColumnDef(sb.append(", "), defs.next());
+        {
+            appendColumnDef(sb, defs.next());
+            if (defs.hasNext())
+                sb.append(", ");
+        }
         return sb.toString();
     }
 

src/java/org/apache/cassandra/db/monitoring/ApproximateTime.java

@@ -0,0 +1,61 @@
+/*
+ * 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.db.monitoring;
+
+import java.util.concurrent.TimeUnit;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.concurrent.ScheduledExecutors;
+import org.apache.cassandra.config.Config;
+
+/**
+ * This is an approximation of System.currentTimeInMillis(). It updates its
+ * time value at periodic intervals of CHECK_INTERVAL_MS milliseconds
+ * (currently 10 milliseconds by default). It can be used as a faster alternative
+ * to System.currentTimeInMillis() every time an imprecision of a few milliseconds
+ * can be accepted.
+ */
+public class ApproximateTime
+{
+    private static final Logger logger = LoggerFactory.getLogger(ApproximateTime.class);
+    private static final int CHECK_INTERVAL_MS = Math.max(5, Integer.valueOf(System.getProperty(Config.PROPERTY_PREFIX + "approximate_time_precision_ms", "10")));
+
+    private static volatile long time = System.currentTimeMillis();
+    static
+    {
+        logger.info("Scheduling approximate time-check task with a precision of {} milliseconds", CHECK_INTERVAL_MS);
+        ScheduledExecutors.scheduledFastTasks.scheduleWithFixedDelay(() -> time = System.currentTimeMillis(),
+                                                                     CHECK_INTERVAL_MS,
+                                                                     CHECK_INTERVAL_MS,
+                                                                     TimeUnit.MILLISECONDS);
+    }
+
+    public static long currentTimeMillis()
+    {
+        return time;
+    }
+
+    public static long precision()
+    {
+        return 2 * CHECK_INTERVAL_MS;
+    }
+
+}

src/java/org/apache/cassandra/db/monitoring/ConstructionTime.java

@@ -0,0 +1,41 @@
+/*
+ * 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.db.monitoring;
+
+public final class ConstructionTime
+{
+    public final long timestamp;
+    public final boolean isCrossNode;
+
+    public ConstructionTime()
+    {
+        this(ApproximateTime.currentTimeMillis());
+    }
+
+    public ConstructionTime(long timestamp)
+    {
+        this(timestamp, false);
+    }
+
+    public ConstructionTime(long timestamp, boolean isCrossNode)
+    {
+        this.timestamp = timestamp;
+        this.isCrossNode = isCrossNode;
+    }
+}

src/java/org/apache/cassandra/db/monitoring/Monitorable.java

@@ -0,0 +1,33 @@
+/*
+ * 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.db.monitoring;
+
+public interface Monitorable
+{
+    String name();
+    ConstructionTime constructionTime();
+    long timeout();
+
+    boolean isInProgress();
+    boolean isAborted();
+    boolean isCompleted();
+
+    boolean abort();
+    boolean complete();
+}

src/java/org/apache/cassandra/db/monitoring/MonitorableImpl.java

@@ -0,0 +1,104 @@
+/*
+ * 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.db.monitoring;
+
+public abstract class MonitorableImpl implements Monitorable
+{
+    private MonitoringState state;
+    private ConstructionTime constructionTime;
+    private long timeout;
+
+    protected MonitorableImpl()
+    {
+        this.state = MonitoringState.IN_PROGRESS;
+    }
+
+    /**
+     * This setter is ugly but the construction chain to ReadCommand
+     * is too complex, it would require passing new parameters to all serializers
+     * or specializing the serializers to accept these message properties.
+     */
+    public void setMonitoringTime(ConstructionTime constructionTime, long timeout)
+    {
+        this.constructionTime = constructionTime;
+        this.timeout = timeout;
+    }
+
+    public ConstructionTime constructionTime()
+    {
+        return constructionTime;
+    }
+
+    public long timeout()
+    {
+        return timeout;
+    }
+
+    public boolean isInProgress()
+    {
+        check();
+        return state == MonitoringState.IN_PROGRESS;
+    }
+
+    public boolean isAborted()
+    {
+        check();
+        return state == MonitoringState.ABORTED;
+    }
+
+    public boolean isCompleted()
+    {
+        check();
+        return state == MonitoringState.COMPLETED;
+    }
+
+    public boolean abort()
+    {
+        if (state == MonitoringState.IN_PROGRESS)
+        {
+            if (constructionTime != null)
+                MonitoringTask.addFailedOperation(this, ApproximateTime.currentTimeMillis());
+            state = MonitoringState.ABORTED;
+            return true;
+        }
+
+        return state == MonitoringState.ABORTED;
+    }
+
+    public boolean complete()
+    {
+        if (state == MonitoringState.IN_PROGRESS)
+        {
+            state = MonitoringState.COMPLETED;
+            return true;
+        }
+
+        return state == MonitoringState.COMPLETED;
+    }
+
+    private void check()
+    {
+        if (constructionTime == null || state != MonitoringState.IN_PROGRESS)
+            return;
+
+        long elapsed = ApproximateTime.currentTimeMillis() - constructionTime.timestamp;
+        if (elapsed >= timeout)
+            abort();
+    }
+}

src/java/org/apache/cassandra/db/monitoring/MonitoringState.java

@@ -0,0 +1,26 @@
+/*
+ * 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.db.monitoring;
+
+public enum MonitoringState
+{
+    IN_PROGRESS,
+    ABORTED,
+    COMPLETED
+}

src/java/org/apache/cassandra/db/monitoring/MonitoringTask.java

@@ -0,0 +1,264 @@
+/*
+ * 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.db.monitoring;
+
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ArrayBlockingQueue;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicLong;
+
+import com.google.common.annotations.VisibleForTesting;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.cassandra.concurrent.ScheduledExecutors;
+import org.apache.cassandra.config.Config;
+
+import static java.lang.System.getProperty;
+
+/**
+ * A task for monitoring in progress operations, currently only read queries, and aborting them if they time out.
+ * We also log timed out operations, see CASSANDRA-7392.
+ */
+public class MonitoringTask
+{
+    private static final String LINE_SEPARATOR = getProperty("line.separator");
+    private static final Logger logger = LoggerFactory.getLogger(MonitoringTask.class);
+
+    /**
+     * Defines the interval for reporting any operations that have timed out.
+     */
+    private static final int REPORT_INTERVAL_MS = Math.max(0, Integer.valueOf(System.getProperty(Config.PROPERTY_PREFIX + "monitoring_report_interval_ms", "5000")));
+
+    /**
+     * Defines the maximum number of unique timed out queries that will be reported in the logs.
+     * Use a negative number to remove any limit.
+     */
+    private static final int MAX_OPERATIONS = Integer.valueOf(System.getProperty(Config.PROPERTY_PREFIX + "monitoring_max_operations", "50"));
+
+    @VisibleForTesting
+    static MonitoringTask instance = make(REPORT_INTERVAL_MS, MAX_OPERATIONS);
+
+    private final int maxOperations;
+    private final ScheduledFuture<?> reportingTask;
+    private final BlockingQueue<FailedOperation> operationsQueue;
+    private final AtomicLong numDroppedOperations;
+    private long lastLogTime;
+
+    @VisibleForTesting
+    static MonitoringTask make(int reportIntervalMillis, int maxTimedoutOperations)
+    {
+        if (instance != null)
+        {
+            instance.cancel();
+            instance = null;
+        }
+
+        return new MonitoringTask(reportIntervalMillis, maxTimedoutOperations);
+    }
+
+    private MonitoringTask(int reportIntervalMillis, int maxOperations)
+    {
+        this.maxOperations = maxOperations;
+        this.operationsQueue = maxOperations > 0 ? new ArrayBlockingQueue<>(maxOperations) : new LinkedBlockingQueue<>();
+        this.numDroppedOperations = new AtomicLong();
+        this.lastLogTime = ApproximateTime.currentTimeMillis();
+
+        logger.info("Scheduling monitoring task with report interval of {} ms, max operations {}", reportIntervalMillis, maxOperations);
+        this.reportingTask = ScheduledExecutors.scheduledTasks.scheduleWithFixedDelay(() -> logFailedOperations(ApproximateTime.currentTimeMillis()),
+                                                                                     reportIntervalMillis,
+                                                                                     reportIntervalMillis,
+                                                                                     TimeUnit.MILLISECONDS);
+    }
+
+    public void cancel()
+    {
+        reportingTask.cancel(false);
+    }
+
+    public static void addFailedOperation(Monitorable operation, long now)
+    {
+        instance.innerAddFailedOperation(operation, now);
+    }
+
+    private void innerAddFailedOperation(Monitorable operation, long now)
+    {
+        if (maxOperations == 0)
+            return; // logging of failed operations disabled
+
+        if (!operationsQueue.offer(new FailedOperation(operation, now)))
+            numDroppedOperations.incrementAndGet();
+    }
+
+    @VisibleForTesting
+    FailedOperations aggregateFailedOperations()
+    {
+        Map<String, FailedOperation> operations = new HashMap<>();
+
+        FailedOperation failedOperation;
+        while((failedOperation = operationsQueue.poll()) != null)
+        {
+            FailedOperation existing = operations.get(failedOperation.name());
+            if (existing != null)
+                existing.addTimeout(failedOperation);
+            else
+                operations.put(failedOperation.name(), failedOperation);
+        }
+
+        return new FailedOperations(operations, numDroppedOperations.getAndSet(0L));
+    }
+
+    @VisibleForTesting
+    List<String> getFailedOperations()
+    {
+        FailedOperations failedOperations = aggregateFailedOperations();
+        String ret = failedOperations.getLogMessage();
+        lastLogTime = ApproximateTime.currentTimeMillis();
+        return ret.isEmpty() ? Collections.emptyList() : Arrays.asList(ret.split("\n"));
+    }
+
+    @VisibleForTesting
+    void logFailedOperations(long now)
+    {
+        FailedOperations failedOperations = aggregateFailedOperations();
+        if (!failedOperations.isEmpty())
+        {
+            long elapsed = now - lastLogTime;
+            logger.warn("{} operations timed out in the last {} msecs, operation list available at debug log level",
+                        failedOperations.num(),
+                        elapsed);
+
+            if (logger.isDebugEnabled())
+                logger.debug("{} operations timed out in the last {} msecs:{}{}",
+                            failedOperations.num(),
+                            elapsed,
+                            LINE_SEPARATOR,
+                            failedOperations.getLogMessage());
+        }
+
+        lastLogTime = now;
+    }
+
+    private static final class FailedOperations
+    {
+        public final Map<String, FailedOperation> operations;
+        public final long numDropped;
+
+        FailedOperations(Map<String, FailedOperation> operations, long numDropped)
+        {
+            this.operations = operations;
+            this.numDropped = numDropped;
+        }
+
+        public boolean isEmpty()
+        {
+            return operations.isEmpty() && numDropped == 0;
+        }
+
+        public long num()
+        {
+            return operations.size() + numDropped;
+        }
+
+        public String getLogMessage()
+        {
+            if (isEmpty())
+                return "";
+
+            final StringBuilder ret = new StringBuilder();
+            operations.values().forEach(o -> addOperation(ret, o));
+
+            if (numDropped > 0)
+                ret.append(LINE_SEPARATOR)
+                   .append("... (")
+                   .append(numDropped)
+                   .append(" were dropped)");
+
+            return ret.toString();
+        }
+
+        private static void addOperation(StringBuilder ret, FailedOperation operation)
+        {
+            if (ret.length() > 0)
+                ret.append(LINE_SEPARATOR);
+
+            ret.append(operation.getLogMessage());
+        }
+    }
+
+    private final static class FailedOperation
+    {
+        public final Monitorable operation;
+        public int numTimeouts;
+        public long totalTime;
+        public long maxTime;
+        public long minTime;
+        private String name;
+
+        FailedOperation(Monitorable operation, long failedAt)
+        {
+            this.operation = operation;
+            numTimeouts = 1;
+            totalTime = failedAt - operation.constructionTime().timestamp;
+            minTime = totalTime;
+            maxTime = totalTime;
+        }
+
+        public String name()
+        {
+            if (name == null)
+                name = operation.name();
+            return name;
+        }
+
+        void addTimeout(FailedOperation operation)
+        {
+            numTimeouts++;
+            totalTime += operation.totalTime;
+            maxTime = Math.max(maxTime, operation.maxTime);
+            minTime = Math.min(minTime, operation.minTime);
+        }
+
+        public String getLogMessage()
+        {
+            if (numTimeouts == 1)
+                return String.format("%s: total time %d msec - timeout %d %s",
+                                     name(),
+                                     totalTime,
+                                     operation.timeout(),
+                                     operation.constructionTime().isCrossNode ? "msec/cross-node" : "msec");
+            else
+                return String.format("%s (timed out %d times): total time avg/min/max %d/%d/%d msec - timeout %d %s",
+                                     name(),
+                                     numTimeouts,
+                                     totalTime / numTimeouts,
+                                     minTime,
+                                     maxTime,
+                                     operation.timeout(),
+                                     operation.constructionTime().isCrossNode ? "msec/cross-node" : "msec");
+        }
+    }
+}

src/java/org/apache/cassandra/db/view/View.java

@@ -480,8 +480,8 @@ public class View
                 // Add all of the rows which were recovered from the query to the row set
                 while (!pager.isExhausted())
                 {
-                    try (ReadOrderGroup orderGroup = pager.startOrderGroup();
-                         PartitionIterator iter = pager.fetchPageInternal(128, orderGroup))
+                    try (ReadExecutionController executionController = pager.executionController();
+                         PartitionIterator iter = pager.fetchPageInternal(128, executionController))
                     {
                         if (!iter.hasNext())
                             break;
@@ -538,8 +538,8 @@ public class View
 
         while (!pager.isExhausted())
         {
-            try (ReadOrderGroup orderGroup = pager.startOrderGroup();
-                 PartitionIterator iter = pager.fetchPageInternal(128, orderGroup))
+            try (ReadExecutionController executionController = pager.executionController();
+                 PartitionIterator iter = pager.fetchPageInternal(128, executionController))
             {
                 while (iter.hasNext())
                 {

src/java/org/apache/cassandra/db/view/ViewBuilder.java

@@ -80,8 +80,8 @@ public class ViewBuilder extends CompactionInfo.Holder
 
         while (!pager.isExhausted())
         {
-           try (ReadOrderGroup orderGroup = pager.startOrderGroup();
-                PartitionIterator partitionIterator = pager.fetchPageInternal(128, orderGroup))
+           try (ReadExecutionController executionController = pager.executionController();
+                PartitionIterator partitionIterator = pager.fetchPageInternal(128, executionController))
            {
                if (!partitionIterator.hasNext())
                    return;

src/java/org/apache/cassandra/index/Index.java

@@ -427,6 +427,6 @@ public interface Index
          * @param orderGroup the collection of OpOrder.Groups which the ReadCommand is being performed under.
          * @return partitions from the base table matching the criteria of the search.
          */
-        public UnfilteredPartitionIterator search(ReadOrderGroup orderGroup);
+        public UnfilteredPartitionIterator search(ReadExecutionController executionController);
     }
 }

src/java/org/apache/cassandra/index/internal/CassandraIndexSearcher.java

@@ -36,14 +36,14 @@ public abstract class CassandraIndexSearcher implements Index.Searcher
 
     @SuppressWarnings("resource") // Both the OpOrder and 'indexIter' are closed on exception, or through the closing of the result
     // of this method.
-    public UnfilteredPartitionIterator search(ReadOrderGroup orderGroup)
+    public UnfilteredPartitionIterator search(ReadExecutionController executionController)
     {
         // the value of the index expression is the partition key in the index table
         DecoratedKey indexKey = index.getBackingTable().get().decorateKey(expression.getIndexValue());
-        UnfilteredRowIterator indexIter = queryIndex(indexKey, command, orderGroup);
+        UnfilteredRowIterator indexIter = queryIndex(indexKey, command, executionController);
         try
         {
-            return queryDataFromIndex(indexKey, UnfilteredRowIterators.filter(indexIter, command.nowInSec()), command, orderGroup);
+            return queryDataFromIndex(indexKey, UnfilteredRowIterators.filter(indexIter, command.nowInSec()), command, executionController);
         }
         catch (RuntimeException | Error e)
         {
@@ -52,13 +52,13 @@ public abstract class CassandraIndexSearcher implements Index.Searcher
         }
     }
 
-    private UnfilteredRowIterator queryIndex(DecoratedKey indexKey, ReadCommand command, ReadOrderGroup orderGroup)
+    private UnfilteredRowIterator queryIndex(DecoratedKey indexKey, ReadCommand command, ReadExecutionController executionController)
     {
         ClusteringIndexFilter filter = makeIndexFilter(command);
         ColumnFamilyStore indexCfs = index.getBackingTable().get();
         CFMetaData indexCfm = indexCfs.metadata;
         return SinglePartitionReadCommand.create(indexCfm, command.nowInSec(), indexKey, ColumnFilter.all(indexCfm), filter)
-                                         .queryMemtableAndDisk(indexCfs, orderGroup.indexReadOpOrderGroup());
+                                         .queryMemtableAndDisk(indexCfs, executionController.indexReadOpOrderGroup());
     }
 
     private ClusteringIndexFilter makeIndexFilter(ReadCommand command)
@@ -168,5 +168,5 @@ public abstract class CassandraIndexSearcher implements Index.Searcher
     protected abstract UnfilteredPartitionIterator queryDataFromIndex(DecoratedKey indexKey,
                                                                       RowIterator indexHits,
                                                                       ReadCommand command,
-                                                                      ReadOrderGroup orderGroup);
+                                                                      ReadExecutionController executionController);
 }

src/java/org/apache/cassandra/index/internal/composites/CompositesSearcher.java

@@ -57,7 +57,7 @@ public class CompositesSearcher extends CassandraIndexSearcher
     protected UnfilteredPartitionIterator queryDataFromIndex(final DecoratedKey indexKey,
                                                              final RowIterator indexHits,
                                                              final ReadCommand command,
-                                                             final ReadOrderGroup orderGroup)
+                                                             final ReadExecutionController executionController)
     {
         assert indexHits.staticRow() == Rows.EMPTY_STATIC_ROW;
 
@@ -143,10 +143,10 @@ public class CompositesSearcher extends CassandraIndexSearcher
                 @SuppressWarnings("resource") // We close right away if empty, and if it's assign to next it will be called either
                                               // by the next caller of next, or through closing this iterator is this come before.
                 UnfilteredRowIterator dataIter = filterStaleEntries(dataCmd.queryMemtableAndDisk(index.baseCfs,
-                                                                                                 orderGroup.baseReadOpOrderGroup()),
+                                                                                                 executionController.baseReadOpOrderGroup()),
                                                                     indexKey.getKey(),
                                                                     entries,
-                                                                    orderGroup.writeOpOrderGroup(),
+                                                                    executionController.writeOpOrderGroup(),
                                                                     command.nowInSec());
 
 

src/java/org/apache/cassandra/index/internal/keys/KeysSearcher.java

@@ -47,7 +47,7 @@ public class KeysSearcher extends CassandraIndexSearcher
     protected UnfilteredPartitionIterator queryDataFromIndex(final DecoratedKey indexKey,
                                                              final RowIterator indexHits,
                                                              final ReadCommand command,
-                                                             final ReadOrderGroup orderGroup)
+                                                             final ReadExecutionController executionController)
     {
         assert indexHits.staticRow() == Rows.EMPTY_STATIC_ROW;
 
@@ -100,10 +100,10 @@ public class KeysSearcher extends CassandraIndexSearcher
                                                   // Otherwise, we close right away if empty, and if it's assigned to next it will be called either
                                                   // by the next caller of next, or through closing this iterator is this come before.
                     UnfilteredRowIterator dataIter = filterIfStale(dataCmd.queryMemtableAndDisk(index.baseCfs,
-                                                                                                orderGroup.baseReadOpOrderGroup()),
+                                                                                                executionController.baseReadOpOrderGroup()),
                                                                    hit,
                                                                    indexKey.getKey(),
-                                                                   orderGroup.writeOpOrderGroup(),
+                                                                   executionController.writeOpOrderGroup(),
                                                                    isForThrift(),
                                                                    command.nowInSec());
 

src/java/org/apache/cassandra/net/IAsyncCallback.java

@@ -31,7 +31,7 @@ import org.apache.cassandra.gms.FailureDetector;
  */
 public interface IAsyncCallback<T>
 {
-    public static Predicate<InetAddress> isAlive = new Predicate<InetAddress>()
+    Predicate<InetAddress> isAlive = new Predicate<InetAddress>()
     {
         public boolean apply(InetAddress endpoint)
         {
@@ -42,7 +42,7 @@ public interface IAsyncCallback<T>
     /**
      * @param msg response received.
      */
-    public void response(MessageIn<T> msg);
+    void response(MessageIn<T> msg);
 
     /**
      * @return true if this callback is on the read path and its latency should be

src/java/org/apache/cassandra/net/IAsyncCallbackWithFailure.java

@@ -25,5 +25,5 @@ public interface IAsyncCallbackWithFailure<T> extends IAsyncCallback<T>
     /**
      * Called when there is an exception on the remote node or timeout happens
      */
-    public void onFailure(InetAddress from);
+    void onFailure(InetAddress from);
 }

src/java/org/apache/cassandra/net/IVerbHandler.java

@@ -19,6 +19,8 @@ package org.apache.cassandra.net;
 
 import java.io.IOException;
 
+import org.apache.cassandra.db.ReadCommand;
+
 /**
  * IVerbHandler provides the method that all verb handlers need to implement.
  * The concrete implementation of this interface would provide the functionality
@@ -36,5 +38,5 @@ public interface IVerbHandler<T>
      * @param message - incoming message that needs handling.
      * @param id
      */
-    public void doVerb(MessageIn<T> message, int id) throws IOException;
+    void doVerb(MessageIn<T> message, int id) throws IOException;
 }

src/java/org/apache/cassandra/net/IncomingTcpConnection.java

@@ -184,18 +184,7 @@ public class IncomingTcpConnection extends Thread implements Closeable
         else
             id = input.readInt();
 
-        long timestamp = System.currentTimeMillis();
-        boolean isCrossNodeTimestamp = false;
-        // make sure to readInt, even if cross_node_to is not enabled
-        int partial = input.readInt();
-        if (DatabaseDescriptor.hasCrossNodeTimeout())
-        {
-            long crossNodeTimestamp = (timestamp & 0xFFFFFFFF00000000L) | (((partial & 0xFFFFFFFFL) << 2) >> 2);
-            isCrossNodeTimestamp = (timestamp != crossNodeTimestamp);
-            timestamp = crossNodeTimestamp;
-        }
-
-        MessageIn message = MessageIn.read(input, version, id);
+        MessageIn message = MessageIn.read(input, version, id, MessageIn.readTimestamp(input));
         if (message == null)
         {
             // callback expired; nothing to do
@@ -203,7 +192,7 @@ public class IncomingTcpConnection extends Thread implements Closeable
         }
         if (version <= MessagingService.current_version)
         {
-            MessagingService.instance().receive(message, id, timestamp, isCrossNodeTimestamp);
+            MessagingService.instance().receive(message, id);
         }
         else
         {

src/java/org/apache/cassandra/net/MessageDeliveryTask.java

@@ -32,25 +32,21 @@ public class MessageDeliveryTask implements Runnable
 
     private final MessageIn message;
     private final int id;
-    private final long constructionTime;
-    private final boolean isCrossNodeTimestamp;
 
-    public MessageDeliveryTask(MessageIn message, int id, long timestamp, boolean isCrossNodeTimestamp)
+    public MessageDeliveryTask(MessageIn message, int id)
     {
         assert message != null;
         this.message = message;
         this.id = id;
-        this.constructionTime = timestamp;
-        this.isCrossNodeTimestamp = isCrossNodeTimestamp;
     }
 
     public void run()
     {
         MessagingService.Verb verb = message.verb;
         if (MessagingService.DROPPABLE_VERBS.contains(verb)
-            && System.currentTimeMillis() > constructionTime + message.getTimeout())
+            && System.currentTimeMillis() > message.constructionTime.timestamp + message.getTimeout())
         {
-            MessagingService.instance().incrementDroppedMessages(verb, isCrossNodeTimestamp);
+            MessagingService.instance().incrementDroppedMessages(message);
             return;
         }
 
@@ -82,7 +78,7 @@ public class MessageDeliveryTask implements Runnable
         }
 
         if (GOSSIP_VERBS.contains(message.verb))
-            Gossiper.instance.setLastProcessedMessageAt(constructionTime);
+            Gossiper.instance.setLastProcessedMessageAt(message.constructionTime.timestamp);
     }
 
     private void handleFailure(Throwable t)

src/java/org/apache/cassandra/net/MessageIn.java

@@ -24,39 +24,54 @@ import java.util.Map;
 
 import com.google.common.collect.ImmutableMap;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.apache.cassandra.concurrent.Stage;
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.monitoring.ConstructionTime;
+import org.apache.cassandra.db.monitoring.MonitorableImpl;
 import org.apache.cassandra.io.IVersionedSerializer;
 import org.apache.cassandra.io.util.DataInputPlus;
 import org.apache.cassandra.io.util.FileUtils;
 
 public class MessageIn<T>
 {
-    private static final Logger logger = LoggerFactory.getLogger(MessageIn.class);
-
     public final InetAddress from;
     public final T payload;
     public final Map<String, byte[]> parameters;
     public final MessagingService.Verb verb;
     public final int version;
+    public final ConstructionTime constructionTime;
 
-    private MessageIn(InetAddress from, T payload, Map<String, byte[]> parameters, MessagingService.Verb verb, int version)
+    private MessageIn(InetAddress from,
+                      T payload,
+                      Map<String, byte[]> parameters,
+                      MessagingService.Verb verb,
+                      int version,
+                      ConstructionTime constructionTime)
     {
         this.from = from;
         this.payload = payload;
         this.parameters = parameters;
         this.verb = verb;
         this.version = version;
+        this.constructionTime = constructionTime;
     }
 
-    public static <T> MessageIn<T> create(InetAddress from, T payload, Map<String, byte[]> parameters, MessagingService.Verb verb, int version)
+    public static <T> MessageIn<T> create(InetAddress from,
+                                          T payload,
+                                          Map<String, byte[]> parameters,
+                                          MessagingService.Verb verb,
+                                          int version,
+                                          ConstructionTime constructionTime)
     {
-        return new MessageIn<T>(from, payload, parameters, verb, version);
+        return new MessageIn<>(from, payload, parameters, verb, version, constructionTime);
     }
 
     public static <T2> MessageIn<T2> read(DataInputPlus in, int version, int id) throws IOException
+    {
+        return read(in, version, id, new ConstructionTime());
+    }
+
+    public static <T2> MessageIn<T2> read(DataInputPlus in, int version, int id, ConstructionTime constructionTime) throws IOException
     {
         InetAddress from = CompactEndpointSerializationHelper.deserialize(in);
 
@@ -94,9 +109,31 @@ public class MessageIn<T>
             serializer = (IVersionedSerializer<T2>) callback.serializer;
         }
         if (payloadSize == 0 || serializer == null)
-            return create(from, null, parameters, verb, version);
+            return create(from, null, parameters, verb, version, constructionTime);
+
         T2 payload = serializer.deserialize(in, version);
-        return MessageIn.create(from, payload, parameters, verb, version);
+        return MessageIn.create(from, payload, parameters, verb, version, constructionTime);
+    }
+
+    public static ConstructionTime createTimestamp()
+    {
+        return new ConstructionTime();
+    }
+
+    public static ConstructionTime readTimestamp(DataInputPlus input) throws IOException
+    {
+        // make sure to readInt, even if cross_node_to is not enabled
+        int partial = input.readInt();
+        if(DatabaseDescriptor.hasCrossNodeTimeout())
+        {
+            long timestamp = System.currentTimeMillis();
+            long crossNodeTimestamp = (timestamp & 0xFFFFFFFF00000000L) | (((partial & 0xFFFFFFFFL) << 2) >> 2);
+            return new ConstructionTime(crossNodeTimestamp, timestamp != crossNodeTimestamp);
+        }
+        else
+        {
+            return new ConstructionTime();
+        }
     }
 
     public Stage getMessageType()

src/java/org/apache/cassandra/net/MessagingService.java

@@ -789,7 +789,7 @@ public final class MessagingService implements MessagingServiceMBean
         }
     }
 
-    public void receive(MessageIn message, int id, long timestamp, boolean isCrossNodeTimestamp)
+    public void receive(MessageIn message, int id)
     {
         TraceState state = Tracing.instance.initializeFromMessage(message);
         if (state != null)
@@ -800,7 +800,7 @@ public final class MessagingService implements MessagingServiceMBean
             if (!ms.allowIncomingMessage(message, id))
                 return;
 
-        Runnable runnable = new MessageDeliveryTask(message, id, timestamp, isCrossNodeTimestamp);
+        Runnable runnable = new MessageDeliveryTask(message, id);
         TracingAwareExecutorService stage = StageManager.getStage(message.getMessageType());
         assert stage != null : "No stage for message type " + message.verb;
 
@@ -937,6 +937,11 @@ public final class MessagingService implements MessagingServiceMBean
         incrementDroppedMessages(verb, false);
     }
 
+    public void incrementDroppedMessages(MessageIn message)
+    {
+        incrementDroppedMessages(message.verb, message.constructionTime.isCrossNode);
+    }
+
     public void incrementDroppedMessages(Verb verb, boolean isCrossNodeTimeout)
     {
         assert DROPPABLE_VERBS.contains(verb) : "Verb " + verb + " should not legally be dropped";

src/java/org/apache/cassandra/schema/SchemaKeyspace.java

@@ -270,7 +270,8 @@ public final class SchemaKeyspace
     public static List<KeyspaceMetadata> readSchemaFromSystemTables()
     {
         ReadCommand cmd = getReadCommandForTableSchema(KEYSPACES);
-        try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); PartitionIterator schema = cmd.executeInternal(orderGroup))
+        try (ReadExecutionController executionController = cmd.executionController();
+             PartitionIterator schema = cmd.executeInternal(executionController))
         {
             List<KeyspaceMetadata> keyspaces = new ArrayList<>();
 
@@ -326,8 +327,8 @@ public final class SchemaKeyspace
         for (String table : ALL)
         {
             ReadCommand cmd = getReadCommandForTableSchema(table);
-            try (ReadOrderGroup orderGroup = cmd.startOrderGroup();
-                 PartitionIterator schema = cmd.executeInternal(orderGroup))
+            try (ReadExecutionController executionController = cmd.executionController();
+                 PartitionIterator schema = cmd.executeInternal(executionController))
             {
                 while (schema.hasNext())
                 {
@@ -374,7 +375,8 @@ public final class SchemaKeyspace
     private static void convertSchemaToMutations(Map<DecoratedKey, Mutation> mutationMap, String schemaTableName)
     {
         ReadCommand cmd = getReadCommandForTableSchema(schemaTableName);
-        try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iter = cmd.executeLocally(orderGroup))
+        try (ReadExecutionController executionController = cmd.executionController();
+             UnfilteredPartitionIterator iter = cmd.executeLocally(executionController))
         {
             while (iter.hasNext())
             {

src/java/org/apache/cassandra/serializers/ByteSerializer.java

@@ -28,7 +28,7 @@ public class ByteSerializer implements TypeSerializer<Byte>
 
     public Byte deserialize(ByteBuffer bytes)
     {
-        return bytes.remaining() == 0 ? null : bytes.get(bytes.position());
+        return bytes == null || bytes.remaining() == 0 ? null : bytes.get(bytes.position());
     }
 
     public ByteBuffer serialize(Byte value)

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

@@ -192,7 +192,8 @@ public class ReadCallback implements IAsyncCallbackWithFailure<ReadResponse>
                                                            result,
                                                            Collections.<String, byte[]>emptyMap(),
                                                            MessagingService.Verb.INTERNAL_RESPONSE,
-                                                           MessagingService.current_version);
+                                                           MessagingService.current_version,
+                                                           MessageIn.createTimestamp());
         response(message);
     }
 

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

@@ -47,6 +47,7 @@ import org.apache.cassandra.config.Schema;
 import org.apache.cassandra.db.*;
 import org.apache.cassandra.db.filter.DataLimits;
 import org.apache.cassandra.db.filter.TombstoneOverwhelmingException;
+import org.apache.cassandra.db.monitoring.ConstructionTime;
 import org.apache.cassandra.db.partitions.*;
 import org.apache.cassandra.db.rows.RowIterator;
 import org.apache.cassandra.db.view.ViewUtils;
@@ -1678,10 +1679,25 @@ public class StorageProxy implements StorageProxyMBean
         {
             try
             {
-                try (ReadOrderGroup orderGroup = command.startOrderGroup(); UnfilteredPartitionIterator iterator = command.executeLocally(orderGroup))
+                command.setMonitoringTime(new ConstructionTime(constructionTime), timeout);
+
+                ReadResponse response;
+                try (ReadExecutionController executionController = command.executionController();
+                     UnfilteredPartitionIterator iterator = command.executeLocally(executionController))
                 {
-                    handler.response(command.createResponse(iterator, command.columnFilter()));
+                    response = command.createResponse(iterator, command.columnFilter());
                 }
+
+                if (command.complete())
+                {
+                    handler.response(response);
+                }
+                else
+                {
+                    MessagingService.instance().incrementDroppedMessages(verb);
+                    handler.onFailure(FBUtilities.getBroadcastAddress());
+                }
+
                 MessagingService.instance().addLatency(FBUtilities.getBroadcastAddress(), TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - start));
             }
             catch (Throwable t)
@@ -2314,18 +2330,20 @@ public class StorageProxy implements StorageProxyMBean
      */
     private static abstract class DroppableRunnable implements Runnable
     {
-        private final long constructionTime = System.nanoTime();
-        private final MessagingService.Verb verb;
+        final long constructionTime;
+        final MessagingService.Verb verb;
+        final long timeout;
 
         public DroppableRunnable(MessagingService.Verb verb)
         {
+            this.constructionTime = System.currentTimeMillis();
             this.verb = verb;
+            this.timeout = DatabaseDescriptor.getTimeout(verb);
         }
 
         public final void run()
         {
-
-            if (TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - constructionTime) > DatabaseDescriptor.getTimeout(verb))
+            if (System.currentTimeMillis() > constructionTime + timeout)
             {
                 MessagingService.instance().incrementDroppedMessages(verb);
                 return;

src/java/org/apache/cassandra/service/pager/AbstractQueryPager.java

@@ -54,9 +54,9 @@ abstract class AbstractQueryPager implements QueryPager
         this.remainingInPartition = limits.perPartitionCount();
     }
 
-    public ReadOrderGroup startOrderGroup()
+    public ReadExecutionController executionController()
     {
-        return command.startOrderGroup();
+        return command.executionController();
     }
 
     public PartitionIterator fetchPage(int pageSize, ConsistencyLevel consistency, ClientState clientState) throws RequestValidationException, RequestExecutionException
@@ -68,13 +68,13 @@ abstract class AbstractQueryPager implements QueryPager
         return new PagerIterator(nextPageReadCommand(pageSize).execute(consistency, clientState), limits.forPaging(pageSize), command.nowInSec());
     }
 
-    public PartitionIterator fetchPageInternal(int pageSize, ReadOrderGroup orderGroup) throws RequestValidationException, RequestExecutionException
+    public PartitionIterator fetchPageInternal(int pageSize, ReadExecutionController executionController) throws RequestValidationException, RequestExecutionException
     {
         if (isExhausted())
             return PartitionIterators.EMPTY;
 
         pageSize = Math.min(pageSize, remaining);
-        return new PagerIterator(nextPageReadCommand(pageSize).executeInternal(orderGroup), limits.forPaging(pageSize), command.nowInSec());
+        return new PagerIterator(nextPageReadCommand(pageSize).executeInternal(executionController), limits.forPaging(pageSize), command.nowInSec());
     }
 
     private class PagerIterator extends CountingPartitionIterator

src/java/org/apache/cassandra/service/pager/MultiPartitionPager.java

@@ -108,14 +108,14 @@ public class MultiPartitionPager implements QueryPager
         return true;
     }
 
-    public ReadOrderGroup startOrderGroup()
+    public ReadExecutionController executionController()
     {
         // Note that for all pagers, the only difference is the partition key to which it applies, so in practice we
         // can use any of the sub-pager ReadOrderGroup group to protect the whole pager
         for (int i = current; i < pagers.length; i++)
         {
             if (pagers[i] != null)
-                return pagers[i].startOrderGroup();
+                return pagers[i].executionController();
         }
         throw new AssertionError("Shouldn't be called on an exhausted pager");
     }
@@ -130,10 +130,10 @@ public class MultiPartitionPager implements QueryPager
         return countingIter;
     }
 
-    public PartitionIterator fetchPageInternal(int pageSize, ReadOrderGroup orderGroup) throws RequestValidationException, RequestExecutionException
+    public PartitionIterator fetchPageInternal(int pageSize, ReadExecutionController executionController) throws RequestValidationException, RequestExecutionException
     {
         int toQuery = Math.min(remaining, pageSize);
-        PagersIterator iter = new PagersIterator(toQuery, null, null, orderGroup);
+        PagersIterator iter = new PagersIterator(toQuery, null, null, executionController);
         CountingPartitionIterator countingIter = new CountingPartitionIterator(iter, limit.forPaging(toQuery), nowInSec);
         iter.setCounter(countingIter.counter());
         return countingIter;
@@ -150,14 +150,14 @@ public class MultiPartitionPager implements QueryPager
         private final ClientState clientState;
 
         // For internal queries
-        private final ReadOrderGroup orderGroup;
+        private final ReadExecutionController executionController;
 
-        public PagersIterator(int pageSize, ConsistencyLevel consistency, ClientState clientState, ReadOrderGroup orderGroup)
+        public PagersIterator(int pageSize, ConsistencyLevel consistency, ClientState clientState, ReadExecutionController executionController)
         {
             this.pageSize = pageSize;
             this.consistency = consistency;
             this.clientState = clientState;
-            this.orderGroup = orderGroup;
+            this.executionController = executionController;
         }
 
         public void setCounter(DataLimits.Counter counter)
@@ -178,7 +178,7 @@ public class MultiPartitionPager implements QueryPager
 
                 int toQuery = pageSize - counter.counted();
                 result = consistency == null
-                       ? pagers[current].fetchPageInternal(toQuery, orderGroup)
+                       ? pagers[current].fetchPageInternal(toQuery, executionController)
                        : pagers[current].fetchPage(toQuery, consistency, clientState);
             }
             return result.next();

src/java/org/apache/cassandra/service/pager/QueryPager.java

@@ -18,7 +18,7 @@
 package org.apache.cassandra.service.pager;
 
 import org.apache.cassandra.db.ConsistencyLevel;
-import org.apache.cassandra.db.ReadOrderGroup;
+import org.apache.cassandra.db.ReadExecutionController;
 import org.apache.cassandra.db.partitions.PartitionIterator;
 import org.apache.cassandra.db.partitions.PartitionIterators;
 import org.apache.cassandra.exceptions.RequestExecutionException;
@@ -46,11 +46,11 @@ import org.apache.cassandra.service.ClientState;
  */
 public interface QueryPager
 {
-    public static final QueryPager EMPTY = new QueryPager()
+    QueryPager EMPTY = new QueryPager()
     {
-        public ReadOrderGroup startOrderGroup()
+        public ReadExecutionController executionController()
         {
-            return ReadOrderGroup.emptyGroup();
+            return ReadExecutionController.empty();
         }
 
         public PartitionIterator fetchPage(int pageSize, ConsistencyLevel consistency, ClientState clientState) throws RequestValidationException, RequestExecutionException
@@ -58,7 +58,7 @@ public interface QueryPager
             return PartitionIterators.EMPTY;
         }
 
-        public PartitionIterator fetchPageInternal(int pageSize, ReadOrderGroup orderGroup) throws RequestValidationException, RequestExecutionException
+        public PartitionIterator fetchPageInternal(int pageSize, ReadExecutionController executionController) throws RequestValidationException, RequestExecutionException
         {
             return PartitionIterators.EMPTY;
         }
@@ -99,16 +99,16 @@ public interface QueryPager
      *
      * @return a newly started order group for this {@code QueryPager}.
      */
-    public ReadOrderGroup startOrderGroup();
+    public ReadExecutionController executionController();
 
     /**
      * Fetches the next page internally (in other, this does a local query).
      *
      * @param pageSize the maximum number of elements to return in the next page.
-     * @param orderGroup the {@code ReadOrderGroup} protecting the read.
+     * @param executionController the {@code ReadExecutionController} protecting the read.
      * @return the page of result.
      */
-    public PartitionIterator fetchPageInternal(int pageSize, ReadOrderGroup orderGroup) throws RequestValidationException, RequestExecutionException;
+    public PartitionIterator fetchPageInternal(int pageSize, ReadExecutionController executionController) throws RequestValidationException, RequestExecutionException;
 
     /**
      * Whether or not this pager is exhausted, i.e. whether or not a call to

src/java/org/apache/cassandra/utils/FBUtilities.java

@@ -816,4 +816,16 @@ public class FBUtilities
             throw new RuntimeException(e);
         }
     }
+	
+	public static void sleepQuietly(long millis)
+    {
+        try
+        {
+            Thread.sleep(millis);
+        }
+        catch (InterruptedException e)
+        {
+            throw new RuntimeException(e);
+        }
+    }
 }

src/java/org/apache/cassandra/utils/NanoTimeToCurrentTimeMillis.java

@@ -19,6 +19,7 @@ package org.apache.cassandra.utils;
 
 import java.util.concurrent.TimeUnit;
 
+import org.apache.cassandra.concurrent.ScheduledExecutors;
 import org.apache.cassandra.config.Config;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -36,9 +37,6 @@ public class NanoTimeToCurrentTimeMillis
 
     private static volatile long TIMESTAMP_BASE[] = new long[] { System.currentTimeMillis(), System.nanoTime() };
 
-    @VisibleForTesting
-    public static final Object TIMESTAMP_UPDATE = new Object();
-
     /*
      * System.currentTimeMillis() is 25 nanoseconds. This is 2 nanoseconds (maybe) according to JMH.
      * Faster than calling both currentTimeMillis() and nanoTime().
@@ -48,41 +46,29 @@ public class NanoTimeToCurrentTimeMillis
      * These timestamps don't order with System.currentTimeMillis() because currentTimeMillis() can tick over
      * before this one does. I have seen it behind by as much as 2ms on Linux and 25ms on Windows.
      */
-    public static final long convert(long nanoTime)
+    public static long convert(long nanoTime)
     {
         final long timestampBase[] = TIMESTAMP_BASE;
         return timestampBase[0] + TimeUnit.NANOSECONDS.toMillis(nanoTime - timestampBase[1]);
     }
 
+    public static void updateNow()
+    {
+        ScheduledExecutors.scheduledFastTasks.submit(NanoTimeToCurrentTimeMillis::updateTimestampBase);
+    }
+
     static
     {
-        //Pick up updates from NTP periodically
-        Thread t = new Thread("NanoTimeToCurrentTimeMillis updater")
-        {
-            @Override
-            public void run()
-            {
-                while (true)
-                {
-                    try
-                    {
-                        synchronized (TIMESTAMP_UPDATE)
-                        {
-                            TIMESTAMP_UPDATE.wait(TIMESTAMP_UPDATE_INTERVAL);
-                        }
-                    }
-                    catch (InterruptedException e)
-                    {
-                        return;
-                    }
+        ScheduledExecutors.scheduledFastTasks.scheduleWithFixedDelay(NanoTimeToCurrentTimeMillis::updateTimestampBase,
+                                                                     TIMESTAMP_UPDATE_INTERVAL,
+                                                                     TIMESTAMP_UPDATE_INTERVAL,
+                                                                     TimeUnit.MILLISECONDS);
+    }
 
-                    TIMESTAMP_BASE = new long[] {
-                            Math.max(TIMESTAMP_BASE[0], System.currentTimeMillis()),
-                            Math.max(TIMESTAMP_BASE[1], System.nanoTime()) };
-                }
-            }
-        };
-        t.setDaemon(true);
-        t.start();
+    private static void updateTimestampBase()
+    {
+        TIMESTAMP_BASE = new long[] {
+                                    Math.max(TIMESTAMP_BASE[0], System.currentTimeMillis()),
+                                    Math.max(TIMESTAMP_BASE[1], System.nanoTime()) };
     }
 }

test/conf/logback-test.xml

@@ -42,13 +42,13 @@
       <immediateFlush>false</immediateFlush>
     </encoder>
   </appender>
-  
+
   <appender name="STDOUT" target="System.out" class="org.apache.cassandra.ConsoleAppender">
     <encoder>
       <pattern>%-5level %date{HH:mm:ss,SSS} %msg%n</pattern>
     </encoder>
     <filter class="ch.qos.logback.classic.filter.ThresholdFilter">
-      <level>INFO</level>
+      <level>DEBUG</level>
     </filter>
   </appender>
 
@@ -59,6 +59,8 @@
 
   <logger name="org.apache.hadoop" level="WARN"/>
 
+  <logger name="org.apache.cassandra.db.monitoring" level="DEBUG"/>
+
   <!-- Do not change the name of this appender. LogbackStatusListener uses the thread name
        tied to the appender name to know when to write to real stdout/stderr vs forwarding to logback -->
   <appender name="ASYNC" class="ch.qos.logback.classic.AsyncAppender">

test/unit/org/apache/cassandra/Util.java

@@ -272,7 +272,8 @@ public class Util
 
     public static void assertEmptyUnfiltered(ReadCommand command)
     {
-        try (ReadOrderGroup orderGroup = command.startOrderGroup(); UnfilteredPartitionIterator iterator = command.executeLocally(orderGroup))
+        try (ReadExecutionController executionController = command.executionController();
+             UnfilteredPartitionIterator iterator = command.executeLocally(executionController))
         {
             if (iterator.hasNext())
             {
@@ -286,7 +287,8 @@ public class Util
 
     public static void assertEmpty(ReadCommand command)
     {
-        try (ReadOrderGroup orderGroup = command.startOrderGroup(); PartitionIterator iterator = command.executeInternal(orderGroup))
+        try (ReadExecutionController executionController = command.executionController();
+             PartitionIterator iterator = command.executeInternal(executionController))
         {
             if (iterator.hasNext())
             {
@@ -301,7 +303,8 @@ public class Util
     public static List<ImmutableBTreePartition> getAllUnfiltered(ReadCommand command)
     {
         List<ImmutableBTreePartition> results = new ArrayList<>();
-        try (ReadOrderGroup orderGroup = command.startOrderGroup(); UnfilteredPartitionIterator iterator = command.executeLocally(orderGroup))
+        try (ReadExecutionController executionController = command.executionController();
+             UnfilteredPartitionIterator iterator = command.executeLocally(executionController))
         {
             while (iterator.hasNext())
             {
@@ -317,7 +320,8 @@ public class Util
     public static List<FilteredPartition> getAll(ReadCommand command)
     {
         List<FilteredPartition> results = new ArrayList<>();
-        try (ReadOrderGroup orderGroup = command.startOrderGroup(); PartitionIterator iterator = command.executeInternal(orderGroup))
+        try (ReadExecutionController executionController = command.executionController();
+             PartitionIterator iterator = command.executeInternal(executionController))
         {
             while (iterator.hasNext())
             {
@@ -332,7 +336,8 @@ public class Util
 
     public static Row getOnlyRowUnfiltered(ReadCommand cmd)
     {
-        try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup))
+        try (ReadExecutionController executionController = cmd.executionController();
+             UnfilteredPartitionIterator iterator = cmd.executeLocally(executionController))
         {
             assert iterator.hasNext() : "Expecting one row in one partition but got nothing";
             try (UnfilteredRowIterator partition = iterator.next())
@@ -349,7 +354,8 @@ public class Util
 
     public static Row getOnlyRow(ReadCommand cmd)
     {
-        try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); PartitionIterator iterator = cmd.executeInternal(orderGroup))
+        try (ReadExecutionController executionController = cmd.executionController();
+             PartitionIterator iterator = cmd.executeInternal(executionController))
         {
             assert iterator.hasNext() : "Expecting one row in one partition but got nothing";
             try (RowIterator partition = iterator.next())
@@ -365,7 +371,8 @@ public class Util
 
     public static ImmutableBTreePartition getOnlyPartitionUnfiltered(ReadCommand cmd)
     {
-        try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup))
+        try (ReadExecutionController executionController = cmd.executionController();
+             UnfilteredPartitionIterator iterator = cmd.executeLocally(executionController))
         {
             assert iterator.hasNext() : "Expecting a single partition but got nothing";
             try (UnfilteredRowIterator partition = iterator.next())
@@ -378,7 +385,8 @@ public class Util
 
     public static FilteredPartition getOnlyPartition(ReadCommand cmd)
     {
-        try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); PartitionIterator iterator = cmd.executeInternal(orderGroup))
+        try (ReadExecutionController executionController = cmd.executionController();
+             PartitionIterator iterator = cmd.executeInternal(executionController))
         {
             assert iterator.hasNext() : "Expecting a single partition but got nothing";
             try (RowIterator partition = iterator.next())

test/unit/org/apache/cassandra/db/KeyspaceTest.java

@@ -135,7 +135,8 @@ public class KeyspaceTest extends CQLTester
         ClusteringIndexSliceFilter filter = new ClusteringIndexSliceFilter(slices, reversed);
         SinglePartitionSliceCommand command = singlePartitionSlice(cfs, key, filter, limit);
 
-        try (ReadOrderGroup orderGroup = command.startOrderGroup(); PartitionIterator iterator = command.executeInternal(orderGroup))
+        try (ReadExecutionController executionController = command.executionController();
+             PartitionIterator iterator = command.executeInternal(executionController))
         {
             try (RowIterator rowIterator = iterator.next())
             {
@@ -208,7 +209,8 @@ public class KeyspaceTest extends CQLTester
             PartitionColumns columns = PartitionColumns.of(cfs.metadata.getColumnDefinition(new ColumnIdentifier("c", false)));
             ClusteringIndexSliceFilter filter = new ClusteringIndexSliceFilter(Slices.ALL, false);
             SinglePartitionSliceCommand command = singlePartitionSlice(cfs, "0", filter, null);
-            try (ReadOrderGroup orderGroup = command.startOrderGroup(); PartitionIterator iterator = command.executeInternal(orderGroup))
+            try (ReadExecutionController executionController = command.executionController();
+                 PartitionIterator iterator = command.executeInternal(executionController))
             {
                 try (RowIterator rowIterator = iterator.next())
                 {
@@ -222,7 +224,8 @@ public class KeyspaceTest extends CQLTester
 
     private static void assertRowsInResult(ColumnFamilyStore cfs, SinglePartitionReadCommand command, int ... columnValues)
     {
-        try (ReadOrderGroup orderGroup = command.startOrderGroup(); PartitionIterator iterator = command.executeInternal(orderGroup))
+        try (ReadExecutionController executionController = command.executionController();
+             PartitionIterator iterator = command.executeInternal(executionController))
         {
             if (columnValues.length == 0)
             {

test/unit/org/apache/cassandra/db/ReadCommandTest.java

@@ -0,0 +1,145 @@
+/*
+ * 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.db;
+
+import java.util.List;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.SchemaLoader;
+import org.apache.cassandra.Util;
+import org.apache.cassandra.config.CFMetaData;
+import org.apache.cassandra.db.marshal.AsciiType;
+import org.apache.cassandra.db.marshal.BytesType;
+import org.apache.cassandra.db.partitions.FilteredPartition;
+import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.schema.KeyspaceParams;
+import org.apache.cassandra.utils.ByteBufferUtil;
+
+import static org.junit.Assert.assertEquals;
+
+public class ReadCommandTest
+{
+    private static final String KEYSPACE = "ReadCommandTest";
+    private static final String CF1 = "Standard1";
+    private static final String CF2 = "Standard2";
+
+    @BeforeClass
+    public static void defineSchema() throws ConfigurationException
+    {
+        CFMetaData metadata1 = SchemaLoader.standardCFMD(KEYSPACE, CF1);
+
+        CFMetaData metadata2 = CFMetaData.Builder.create(KEYSPACE, CF2)
+                                                         .addPartitionKey("key", BytesType.instance)
+                                                         .addClusteringColumn("col", AsciiType.instance)
+                                                         .addRegularColumn("a", AsciiType.instance)
+                                                         .addRegularColumn("b", AsciiType.instance).build();
+
+        SchemaLoader.prepareServer();
+        SchemaLoader.createKeyspace(KEYSPACE,
+                                    KeyspaceParams.simple(1),
+                                    metadata1,
+                                    metadata2);
+    }
+
+    @Test
+    public void testPartitionRangeAbort() throws Exception
+    {
+        ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF1);
+
+        new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key1"))
+                .clustering("Column1")
+                .add("val", ByteBufferUtil.bytes("abcd"))
+                .build()
+                .apply();
+
+        cfs.forceBlockingFlush();
+
+        new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key2"))
+                .clustering("Column1")
+                .add("val", ByteBufferUtil.bytes("abcd"))
+                .build()
+                .apply();
+
+        ReadCommand readCommand = Util.cmd(cfs).build();
+        assertEquals(2, Util.getAll(readCommand).size());
+
+        readCommand.abort();
+        assertEquals(0, Util.getAll(readCommand).size());
+    }
+
+    @Test
+    public void testSinglePartitionSliceAbort() throws Exception
+    {
+        ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF2);
+
+        new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key"))
+                .clustering("cc")
+                .add("a", ByteBufferUtil.bytes("abcd"))
+                .build()
+                .apply();
+
+        cfs.forceBlockingFlush();
+
+        new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key"))
+                .clustering("dd")
+                .add("a", ByteBufferUtil.bytes("abcd"))
+                .build()
+                .apply();
+
+        ReadCommand readCommand = Util.cmd(cfs, Util.dk("key")).build();
+
+        List<FilteredPartition> partitions = Util.getAll(readCommand);
+        assertEquals(1, partitions.size());
+        assertEquals(2, partitions.get(0).rowCount());
+
+        readCommand.abort();
+        assertEquals(0, Util.getAll(readCommand).size());
+    }
+
+    @Test
+    public void testSinglePartitionNamesAbort() throws Exception
+    {
+        ColumnFamilyStore cfs = Keyspace.open(KEYSPACE).getColumnFamilyStore(CF2);
+
+        new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key"))
+                .clustering("cc")
+                .add("a", ByteBufferUtil.bytes("abcd"))
+                .build()
+                .apply();
+
+        cfs.forceBlockingFlush();
+
+        new RowUpdateBuilder(cfs.metadata, 0, ByteBufferUtil.bytes("key"))
+                .clustering("cdd")
+                .add("a", ByteBufferUtil.bytes("abcd"))
+                .build()
+                .apply();
+
+        ReadCommand readCommand = Util.cmd(cfs, Util.dk("key")).includeRow("cc").includeRow("dd").build();
+
+        List<FilteredPartition> partitions = Util.getAll(readCommand);
+        assertEquals(1, partitions.size());
+        assertEquals(2, partitions.get(0).rowCount());
+
+        readCommand.abort();
+        assertEquals(0, Util.getAll(readCommand).size());
+    }
+}

test/unit/org/apache/cassandra/db/RepairedDataTombstonesTest.java

@@ -177,7 +177,8 @@ public class RepairedDataTombstonesTest extends CQLTester
         Thread.sleep(1000);
         ReadCommand cmd = Util.cmd(getCurrentColumnFamilyStore()).build();
         int partitionsFound = 0;
-        try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup))
+        try (ReadExecutionController executionController = cmd.executionController();
+             UnfilteredPartitionIterator iterator = cmd.executeLocally(executionController))
         {
             while (iterator.hasNext())
             {
@@ -232,7 +233,8 @@ public class RepairedDataTombstonesTest extends CQLTester
     {
         ReadCommand cmd = Util.cmd(getCurrentColumnFamilyStore()).build();
         int foundRows = 0;
-        try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup))
+        try (ReadExecutionController executionController = cmd.executionController();
+             UnfilteredPartitionIterator iterator = cmd.executeLocally(executionController))
         {
             while (iterator.hasNext())
             {
@@ -263,7 +265,8 @@ public class RepairedDataTombstonesTest extends CQLTester
     {
         ReadCommand cmd = Util.cmd(getCurrentColumnFamilyStore(), Util.dk(ByteBufferUtil.bytes(key))).build();
         int foundRows = 0;
-        try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator iterator = cmd.executeLocally(orderGroup))
+        try (ReadExecutionController executionController = cmd.executionController();
+             UnfilteredPartitionIterator iterator = cmd.executeLocally(executionController))
         {
             while (iterator.hasNext())
             {

test/unit/org/apache/cassandra/db/SecondaryIndexTest.java

@@ -117,7 +117,8 @@ public class SecondaryIndexTest
                                       .build();
 
         Index.Searcher searcher = cfs.indexManager.getBestIndexFor(rc).searcherFor(rc);
-        try (ReadOrderGroup orderGroup = rc.startOrderGroup(); UnfilteredPartitionIterator pi = searcher.search(orderGroup))
+        try (ReadExecutionController executionController = rc.executionController(); 
+             UnfilteredPartitionIterator pi = searcher.search(executionController))
         {
             assertTrue(pi.hasNext());
             pi.next().close();
@@ -501,8 +502,8 @@ public class SecondaryIndexTest
         if (count != 0)
             assertNotNull(searcher);
 
-        try (ReadOrderGroup orderGroup = rc.startOrderGroup();
-             PartitionIterator iter = UnfilteredPartitionIterators.filter(searcher.search(orderGroup),
+        try (ReadExecutionController executionController = rc.executionController();
+             PartitionIterator iter = UnfilteredPartitionIterators.filter(searcher.search(executionController),
                                                                           FBUtilities.nowInSeconds()))
         {
             assertEquals(count, Util.size(iter));

test/unit/org/apache/cassandra/db/SinglePartitionSliceCommandTest.java

@@ -103,7 +103,7 @@ public class SinglePartitionSliceCommandTest
         cmd = ReadCommand.legacyReadCommandSerializer.deserialize(in, MessagingService.VERSION_21);
 
         logger.debug("ReadCommand: {}", cmd);
-        UnfilteredPartitionIterator partitionIterator = cmd.executeLocally(ReadOrderGroup.emptyGroup());
+        UnfilteredPartitionIterator partitionIterator = cmd.executeLocally(ReadExecutionController.empty());
         ReadResponse response = ReadResponse.createDataResponse(partitionIterator, cmd.columnFilter());
 
         logger.debug("creating response: {}", response);
@@ -148,7 +148,7 @@ public class SinglePartitionSliceCommandTest
                                                           sliceFilter);
 
         // check raw iterator for static cell
-        try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator pi = cmd.executeLocally(orderGroup))
+        try (ReadExecutionController executionController = cmd.executionController(); UnfilteredPartitionIterator pi = cmd.executeLocally(executionController))
         {
             checkForS(pi);
         }
@@ -159,7 +159,7 @@ public class SinglePartitionSliceCommandTest
         ReadResponse dst;
 
         // check (de)serialized iterator for memtable static cell
-        try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator pi = cmd.executeLocally(orderGroup))
+        try (ReadExecutionController executionController = cmd.executionController(); UnfilteredPartitionIterator pi = cmd.executeLocally(executionController))
         {
             response = ReadResponse.createDataResponse(pi, cmd.columnFilter());
         }
@@ -175,7 +175,7 @@ public class SinglePartitionSliceCommandTest
 
         // check (de)serialized iterator for sstable static cell
         Schema.instance.getColumnFamilyStoreInstance(cfm.cfId).forceBlockingFlush();
-        try (ReadOrderGroup orderGroup = cmd.startOrderGroup(); UnfilteredPartitionIterator pi = cmd.executeLocally(orderGroup))
+        try (ReadExecutionController executionController = cmd.executionController(); UnfilteredPartitionIterator pi = cmd.executeLocally(executionController))
         {
             response = ReadResponse.createDataResponse(pi, cmd.columnFilter());
         }

test/unit/org/apache/cassandra/db/monitoring/MonitoringTaskTest.java

@@ -0,0 +1,341 @@
+/*
+ * 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.db.monitoring;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+public class MonitoringTaskTest
+{
+    private static final long timeout = 100;
+    private static final long MAX_SPIN_TIME_NANOS = TimeUnit.SECONDS.toNanos(5);
+
+    public static final int REPORT_INTERVAL_MS = 600000; // long enough so that it won't check unless told to do so
+    public static final int MAX_TIMEDOUT_OPERATIONS = -1; // unlimited
+
+    @BeforeClass
+    public static void setup()
+    {
+        MonitoringTask.instance = MonitoringTask.make(REPORT_INTERVAL_MS, MAX_TIMEDOUT_OPERATIONS);
+    }
+
+    private static final class TestMonitor extends MonitorableImpl
+    {
+        private final String name;
+
+        TestMonitor(String name, ConstructionTime constructionTime, long timeout)
+        {
+            this.name = name;
+            setMonitoringTime(constructionTime, timeout);
+        }
+
+        public String name()
+        {
+            return name;
+        }
+
+        @Override
+        public String toString()
+        {
+            return name();
+        }
+    }
+
+    private static void waitForOperationsToComplete(Monitorable... operations) throws InterruptedException
+    {
+        waitForOperationsToComplete(Arrays.asList(operations));
+    }
+
+    private static void waitForOperationsToComplete(List<Monitorable> operations) throws InterruptedException
+    {
+        long timeout = operations.stream().map(Monitorable::timeout).reduce(0L, Long::max);
+        Thread.sleep(timeout * 2 + ApproximateTime.precision());
+
+        long start = System.nanoTime();
+        while(System.nanoTime() - start <= MAX_SPIN_TIME_NANOS)
+        {
+            long numInProgress = operations.stream().filter(Monitorable::isInProgress).count();
+            if (numInProgress == 0)
+                return;
+
+            Thread.yield();
+        }
+    }
+
+    @Test
+    public void testAbort() throws InterruptedException
+    {
+        Monitorable operation = new TestMonitor("Test abort", new ConstructionTime(System.currentTimeMillis()), timeout);
+        waitForOperationsToComplete(operation);
+
+        assertTrue(operation.isAborted());
+        assertFalse(operation.isCompleted());
+        assertEquals(1, MonitoringTask.instance.getFailedOperations().size());
+    }
+
+    @Test
+    public void testAbortIdemPotent() throws InterruptedException
+    {
+        Monitorable operation = new TestMonitor("Test abort", new ConstructionTime(System.currentTimeMillis()), timeout);
+        waitForOperationsToComplete(operation);
+
+        assertTrue(operation.abort());
+
+        assertTrue(operation.isAborted());
+        assertFalse(operation.isCompleted());
+        assertEquals(1, MonitoringTask.instance.getFailedOperations().size());
+    }
+
+    @Test
+    public void testAbortCrossNode() throws InterruptedException
+    {
+        Monitorable operation = new TestMonitor("Test for cross node", new ConstructionTime(System.currentTimeMillis(), true), timeout);
+        waitForOperationsToComplete(operation);
+
+        assertTrue(operation.isAborted());
+        assertFalse(operation.isCompleted());
+        assertEquals(1, MonitoringTask.instance.getFailedOperations().size());
+    }
+
+    @Test
+    public void testComplete() throws InterruptedException
+    {
+        Monitorable operation = new TestMonitor("Test complete", new ConstructionTime(System.currentTimeMillis()), timeout);
+        operation.complete();
+        waitForOperationsToComplete(operation);
+
+        assertFalse(operation.isAborted());
+        assertTrue(operation.isCompleted());
+        assertEquals(0, MonitoringTask.instance.getFailedOperations().size());
+    }
+
+    @Test
+    public void testCompleteIdemPotent() throws InterruptedException
+    {
+        Monitorable operation = new TestMonitor("Test complete", new ConstructionTime(System.currentTimeMillis()), timeout);
+        operation.complete();
+        waitForOperationsToComplete(operation);
+
+        assertTrue(operation.complete());
+
+        assertFalse(operation.isAborted());
+        assertTrue(operation.isCompleted());
+        assertEquals(0, MonitoringTask.instance.getFailedOperations().size());
+    }
+
+    @Test
+    public void testReport() throws InterruptedException
+    {
+        Monitorable operation = new TestMonitor("Test report", new ConstructionTime(System.currentTimeMillis()), timeout);
+        waitForOperationsToComplete(operation);
+
+        assertTrue(operation.isAborted());
+        assertFalse(operation.isCompleted());
+        MonitoringTask.instance.logFailedOperations(ApproximateTime.currentTimeMillis());
+        assertEquals(0, MonitoringTask.instance.getFailedOperations().size());
+    }
+
+    @Test
+    public void testRealScheduling() throws InterruptedException
+    {
+        MonitoringTask.instance = MonitoringTask.make(10, -1);
+        try
+        {
+            Monitorable operation = new TestMonitor("Test report", new ConstructionTime(System.currentTimeMillis()), timeout);
+            waitForOperationsToComplete(operation);
+
+            assertTrue(operation.isAborted());
+            assertFalse(operation.isCompleted());
+
+            Thread.sleep(ApproximateTime.precision() + 500);
+            assertEquals(0, MonitoringTask.instance.getFailedOperations().size());
+        }
+        finally
+        {
+            MonitoringTask.instance = MonitoringTask.make(REPORT_INTERVAL_MS, MAX_TIMEDOUT_OPERATIONS);
+        }
+    }
+
+    @Test
+    public void testMultipleThreads() throws InterruptedException
+    {
+        final int opCount = 50;
+        final ExecutorService executorService = Executors.newFixedThreadPool(20);
+        final List<Monitorable> operations = Collections.synchronizedList(new ArrayList<>(opCount));
+
+        for (int i = 0; i < opCount; i++)
+        {
+            executorService.submit(() ->
+                operations.add(new TestMonitor(UUID.randomUUID().toString(), new ConstructionTime(), timeout))
+            );
+        }
+
+        executorService.shutdown();
+        assertTrue(executorService.awaitTermination(30, TimeUnit.SECONDS));
+        assertEquals(opCount, operations.size());
+
+        waitForOperationsToComplete(operations);
+        assertEquals(opCount, MonitoringTask.instance.getFailedOperations().size());
+    }
+
+    @Test
+    public void testZeroMaxTimedoutOperations() throws InterruptedException
+    {
+        doTestMaxTimedoutOperations(0, 1, 0);
+    }
+
+    @Test
+    public void testMaxTimedoutOperationsExceeded() throws InterruptedException
+    {
+        doTestMaxTimedoutOperations(5, 10, 6);
+    }
+
+    private static void doTestMaxTimedoutOperations(int maxTimedoutOperations,
+                                                    int numThreads,
+                                                    int numExpectedOperations) throws InterruptedException
+    {
+        MonitoringTask.instance = MonitoringTask.make(REPORT_INTERVAL_MS, maxTimedoutOperations);
+        try
+        {
+            final int threadCount = numThreads;
+            ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
+            final CountDownLatch finished = new CountDownLatch(threadCount);
+
+            for (int i = 0; i < threadCount; i++)
+            {
+                final String operationName = "Operation " + Integer.toString(i+1);
+                final int numTimes = i + 1;
+                executorService.submit(() -> {
+                    try
+                    {
+                        for (int j = 0; j < numTimes; j++)
+                        {
+                            Monitorable operation = new TestMonitor(operationName,
+                                                                    new ConstructionTime(System.currentTimeMillis()),
+                                                                    timeout);
+                            waitForOperationsToComplete(operation);
+                        }
+                    }
+                    catch (InterruptedException e)
+                    {
+                        e.printStackTrace();
+                        fail("Unexpected exception");
+                    }
+                    finally
+                    {
+                        finished.countDown();
+                    }
+                });
+            }
+
+            finished.await();
+            assertEquals(0, executorService.shutdownNow().size());
+
+            List<String> failedOperations = MonitoringTask.instance.getFailedOperations();
+            assertEquals(numExpectedOperations, failedOperations.size());
+            if (numExpectedOperations > 0)
+                assertTrue(failedOperations.get(numExpectedOperations - 1).startsWith("..."));
+        }
+        finally
+        {
+            MonitoringTask.instance = MonitoringTask.make(REPORT_INTERVAL_MS, MAX_TIMEDOUT_OPERATIONS);
+        }
+    }
+
+    @Test
+    public void testMultipleThreadsSameName() throws InterruptedException
+    {
+        final int threadCount = 50;
+        final List<Monitorable> operations = new ArrayList<>(threadCount);
+        ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
+        final CountDownLatch finished = new CountDownLatch(threadCount);
+
+        for (int i = 0; i < threadCount; i++)
+        {
+            executorService.submit(() -> {
+                try
+                {
+                    Monitorable operation = new TestMonitor("Test testMultipleThreadsSameName",
+                                                            new ConstructionTime(System.currentTimeMillis()),
+                                                            timeout);
+                    operations.add(operation);
+                }
+                finally
+                {
+                    finished.countDown();
+                }
+            });
+        }
+
+        finished.await();
+        assertEquals(0, executorService.shutdownNow().size());
+
+        waitForOperationsToComplete(operations);
+        //MonitoringTask.instance.checkFailedOperations(ApproximateTime.currentTimeMillis());
+        assertEquals(1, MonitoringTask.instance.getFailedOperations().size());
+    }
+
+    @Test
+    public void testMultipleThreadsNoFailedOps() throws InterruptedException
+    {
+        final int threadCount = 50;
+        final List<Monitorable> operations = new ArrayList<>(threadCount);
+        ExecutorService executorService = Executors.newFixedThreadPool(threadCount);
+        final CountDownLatch finished = new CountDownLatch(threadCount);
+
+        for (int i = 0; i < threadCount; i++)
+        {
+            executorService.submit(() -> {
+                try
+                {
+                    Monitorable operation = new TestMonitor("Test thread " + Thread.currentThread().getName(),
+                                                            new ConstructionTime(System.currentTimeMillis()),
+                                                            timeout);
+                    operations.add(operation);
+                    operation.complete();
+                }
+                finally
+                {
+                    finished.countDown();
+                }
+            });
+        }
+
+        finished.await();
+        assertEquals(0, executorService.shutdownNow().size());
+
+        waitForOperationsToComplete(operations);
+        assertEquals(0, MonitoringTask.instance.getFailedOperations().size());
+    }
+}

test/unit/org/apache/cassandra/hints/HintTest.java

@@ -222,8 +222,8 @@ public class HintTest
     {
         ReadCommand cmd = cmd(key, table);
 
-        try (ReadOrderGroup orderGroup = cmd.startOrderGroup();
-             PartitionIterator iterator = cmd.executeInternal(orderGroup))
+        try (ReadExecutionController executionController = cmd.executionController();
+             PartitionIterator iterator = cmd.executeInternal(executionController))
         {
             assertFalse(iterator.hasNext());
         }

test/unit/org/apache/cassandra/index/StubIndex.java

@@ -227,9 +227,9 @@ public class StubIndex implements Index
                   Optional.empty());
         }
 
-        private UnfilteredPartitionIterator queryStorageInternal(ColumnFamilyStore cfs, ReadOrderGroup orderGroup)
+        private UnfilteredPartitionIterator queryStorageInternal(ColumnFamilyStore cfs, ReadExecutionController executionController)
         {
-            return queryStorage(cfs, orderGroup);
+            return queryStorage(cfs, executionController);
         }
     }
 }

test/unit/org/apache/cassandra/index/internal/CassandraIndexTest.java

@@ -429,8 +429,8 @@ public class CassandraIndexTest extends CQLTester
                                                                                indexKey,
                                                                                ColumnFilter.all(indexCfs.metadata),
                                                                                filter);
-        try (ReadOrderGroup orderGroup = ReadOrderGroup.forCommand(command);
-             UnfilteredRowIterator iter = command.queryMemtableAndDisk(indexCfs, orderGroup.indexReadOpOrderGroup()))
+        try (ReadExecutionController executionController = ReadExecutionController.forCommand(command);
+             UnfilteredRowIterator iter = command.queryMemtableAndDisk(indexCfs, executionController.indexReadOpOrderGroup()))
         {
             while( iter.hasNext())
             {

test/unit/org/apache/cassandra/io/sstable/SSTableReaderTest.java

@@ -47,7 +47,6 @@ import org.apache.cassandra.index.Index;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.FileDataInput;
 import org.apache.cassandra.io.util.MmappedRegions;
-import org.apache.cassandra.io.util.MmappedSegmentedFile;
 import org.apache.cassandra.io.util.SegmentedFile;
 import org.apache.cassandra.schema.CachingParams;
 import org.apache.cassandra.schema.KeyspaceParams;
@@ -549,9 +548,9 @@ public class SSTableReaderTest
                                              .build();
         Index.Searcher searcher = indexedCFS.indexManager.getBestIndexFor(rc).searcherFor(rc);
         assertNotNull(searcher);
-        try (ReadOrderGroup orderGroup = ReadOrderGroup.forCommand(rc))
+        try (ReadExecutionController executionController = ReadExecutionController.forCommand(rc))
         {
-            assertEquals(1, Util.size(UnfilteredPartitionIterators.filter(searcher.search(orderGroup), rc.nowInSec())));
+            assertEquals(1, Util.size(UnfilteredPartitionIterators.filter(searcher.search(executionController), rc.nowInSec())));
         }
     }
 

test/unit/org/apache/cassandra/service/DataResolverTest.java

@@ -715,7 +715,8 @@ public class DataResolverTest
                                 ReadResponse.createRemoteDataResponse(partitionIterator, cmd.columnFilter()),
                                 Collections.EMPTY_MAP,
                                 MessagingService.Verb.REQUEST_RESPONSE,
-                                MessagingService.current_version);
+                                MessagingService.current_version,
+                                MessageIn.createTimestamp());
     }
 
     private RangeTombstone tombstone(Object start, Object end, long markedForDeleteAt, int localDeletionTime)

test/unit/org/apache/cassandra/service/QueryPagerTest.java

@@ -121,7 +121,8 @@ public class QueryPagerTest
         StringBuilder sb = new StringBuilder();
         List<FilteredPartition> partitionList = new ArrayList<>();
         int rows = 0;
-        try (ReadOrderGroup orderGroup = pager.startOrderGroup(); PartitionIterator iterator = pager.fetchPageInternal(toQuery, orderGroup))
+        try (ReadExecutionController executionController = pager.executionController();
+             PartitionIterator iterator = pager.fetchPageInternal(toQuery, executionController))
         {
             while (iterator.hasNext())
             {

test/unit/org/apache/cassandra/utils/NanoTimeToCurrentTimeMillisTest.java

@@ -34,10 +34,7 @@ public class NanoTimeToCurrentTimeMillisTest
             now = Math.max(now, System.currentTimeMillis());
             if (ii % 10000 == 0)
             {
-                synchronized (NanoTimeToCurrentTimeMillis.TIMESTAMP_UPDATE)
-                {
-                    NanoTimeToCurrentTimeMillis.TIMESTAMP_UPDATE.notify();
-                }
+                NanoTimeToCurrentTimeMillis.updateNow();
                 Thread.sleep(1);
             }