diff --git a/CHANGES.txt b/CHANGES.txt
index eeb19d020f..1b7d7117d0 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,5 +1,7 @@
 4.1.12
  * Harden data resurrection startup check with atomic heartbeat file write with fallback (CASSANDRA-21290)
+Merged from 4.0:
+ * Backport CASSANDRA-17810 fix and improve RTBoundValidator error messages (CASSANDRA-18282)
 
 4.1.11
  * Fix ant generate-eclipse-files (CASSANDRA-21215)
diff --git a/src/java/org/apache/cassandra/db/ReadCommand.java b/src/java/org/apache/cassandra/db/ReadCommand.java
index b3a09e0249..5d90808ca2 100644
--- a/src/java/org/apache/cassandra/db/ReadCommand.java
+++ b/src/java/org/apache/cassandra/db/ReadCommand.java
@@ -48,6 +48,7 @@ import org.apache.cassandra.db.transform.RTBoundValidator;
 import org.apache.cassandra.db.transform.RTBoundValidator.Stage;
 import org.apache.cassandra.db.transform.StoppingTransformation;
 import org.apache.cassandra.db.transform.Transformation;
+import org.apache.cassandra.exceptions.QueryCancelledException;
 import org.apache.cassandra.exceptions.UnknownIndexException;
 import org.apache.cassandra.index.Index;
 import org.apache.cassandra.index.IndexNotAvailableException;
@@ -435,7 +436,8 @@ public abstract class ReadCommand extends AbstractReadQuery
             try
             {
                 iterator = withQuerySizeTracking(iterator);
-                iterator = withStateTracking(iterator);
+                iterator = maybeSlowDownForTesting(iterator);
+                iterator = withQueryCancellation(iterator);
                 iterator = RTBoundValidator.validate(withoutPurgeableTombstones(iterator, cfs, executionController), Stage.PURGED, false);
                 iterator = withMetricsRecording(iterator, cfs.metric, startTimeNanos);
 
@@ -614,9 +616,9 @@ public abstract class ReadCommand extends AbstractReadQuery
         return Transformation.apply(iter, new MetricRecording());
     }
 
-    protected class CheckForAbort extends StoppingTransformation<UnfilteredRowIterator>
+    private class QueryCancellationChecker extends StoppingTransformation<UnfilteredRowIterator>
     {
-        long lastChecked = 0;
+        long lastCheckedAt = 0;
 
         @Override
         protected void attachTo(BasePartitions partitions)
@@ -634,51 +636,62 @@ public abstract class ReadCommand extends AbstractReadQuery
             this.rows = rows;
         }
 
+        @Override
         protected UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
         {
-            if (maybeAbort())
-            {
-                partition.close();
-                return null;
-            }
-
+            maybeCancel();
             return Transformation.apply(partition, this);
         }
 
+        @Override
         protected Row applyToRow(Row row)
         {
-            if (TEST_ITERATION_DELAY_MILLIS > 0)
-                maybeDelayForTesting();
-
-            return maybeAbort() ? null : row;
+            maybeCancel();
+            return row;
         }
 
-        private boolean maybeAbort()
+        private void maybeCancel()
         {
-            /**
-             * TODO: this is not a great way to abort early; why not expressly limit checks to 10ms intervals?
+            /*
              * The value returned by approxTime.now() is updated only every
-             * {@link org.apache.cassandra.utils.MonotonicClock.SampledClock.CHECK_INTERVAL_MS}, by default 2 millis. Since MonitorableImpl
-             * relies on approxTime, we don't need to check unless the approximate time has elapsed.
+             * {@link org.apache.cassandra.utils.MonotonicClock.SampledClock.CHECK_INTERVAL_MS}, by default 2 millis.
+             * Since MonitorableImpl relies on approxTime, we don't need to check unless the approximate time has elapsed.
              */
-            if (lastChecked == approxTime.now())
-                return false;
-
-            lastChecked = approxTime.now();
+            if (lastCheckedAt == approxTime.now())
+                return;
+            lastCheckedAt = approxTime.now();
 
             if (isAborted())
             {
                 stop();
-                return true;
+                throw new QueryCancelledException(ReadCommand.this);
             }
+        }
+    }
 
-            return false;
+    private UnfilteredPartitionIterator withQueryCancellation(UnfilteredPartitionIterator iter)
+    {
+        return Transformation.apply(iter, new QueryCancellationChecker());
+    }
+
+    /**
+     * A transformation used for simulating slow queries by tests.
+     */
+    @VisibleForTesting
+    private static class DelayInjector extends Transformation<UnfilteredRowIterator>
+    {
+        @Override
+        protected UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
+        {
+            FBUtilities.sleepQuietly(TEST_ITERATION_DELAY_MILLIS);
+            return Transformation.apply(partition, this);
         }
 
-        private void maybeDelayForTesting()
+        @Override
+        protected Row applyToRow(Row row)
         {
-            if (!metadata().keyspace.startsWith("system"))
-                FBUtilities.sleepQuietly(TEST_ITERATION_DELAY_MILLIS);
+            FBUtilities.sleepQuietly(TEST_ITERATION_DELAY_MILLIS);
+            return row;
         }
     }
 
@@ -766,9 +779,12 @@ public abstract class ReadCommand extends AbstractReadQuery
         return iterator;
     }
 
-    protected UnfilteredPartitionIterator withStateTracking(UnfilteredPartitionIterator iter)
+    private UnfilteredPartitionIterator maybeSlowDownForTesting(UnfilteredPartitionIterator iter)
     {
-        return Transformation.apply(iter, new CheckForAbort());
+        if (TEST_ITERATION_DELAY_MILLIS > 0 && !SchemaConstants.isSystemKeyspace(metadata().keyspace))
+            return Transformation.apply(iter, new DelayInjector());
+        else
+            return iter;
     }
 
     /**
diff --git a/src/java/org/apache/cassandra/db/ReadCommandVerbHandler.java b/src/java/org/apache/cassandra/db/ReadCommandVerbHandler.java
index 7e28c8ab9e..726eba6dd3 100644
--- a/src/java/org/apache/cassandra/db/ReadCommandVerbHandler.java
+++ b/src/java/org/apache/cassandra/db/ReadCommandVerbHandler.java
@@ -19,6 +19,8 @@ package org.apache.cassandra.db;
 
 import java.util.concurrent.TimeUnit;
 
+import com.google.common.base.Preconditions;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -26,6 +28,7 @@ import org.apache.cassandra.config.DatabaseDescriptor;
 import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.exceptions.InvalidRequestException;
+import org.apache.cassandra.exceptions.QueryCancelledException;
 import org.apache.cassandra.locator.Replica;
 import org.apache.cassandra.net.IVerbHandler;
 import org.apache.cassandra.net.Message;
@@ -106,18 +109,29 @@ public class ReadCommandVerbHandler implements IVerbHandler<ReadCommand>
             MessagingService.instance().send(reply, message.from());
             return;
         }
+        catch (AssertionError t)
+        {
+            throw new AssertionError(String.format("Caught an error while trying to process the command: %s", command.toCQLString()), t);
+        }
+        catch (QueryCancelledException e)
+        {
+            logger.debug("Query cancelled (timeout)", e);
+            response = null;
+            Preconditions.checkState(!command.isCompleted(), "Read marked as completed despite being aborted by timeout to table %s", command.metadata());
+        }
 
-        if (!command.complete())
+        if (command.complete())
+        {
+            Tracing.trace("Enqueuing response to {}", message.from());
+            Message<ReadResponse> reply = message.responseWith(response);
+            reply = MessageParams.addToMessage(reply);
+            MessagingService.instance().send(reply, message.from());
+        }
+        else
         {
             Tracing.trace("Discarding partial response to {} (timed out)", message.from());
             MessagingService.instance().metrics.recordDroppedMessage(message, message.elapsedSinceCreated(NANOSECONDS), NANOSECONDS);
-            return;
         }
-
-        Tracing.trace("Enqueuing response to {}", message.from());
-        Message<ReadResponse> reply = message.responseWith(response);
-        reply = MessageParams.addToMessage(reply);
-        MessagingService.instance().send(reply, message.from());
     }
 
     private void validateTransientStatus(Message<ReadCommand> message)
diff --git a/src/java/org/apache/cassandra/db/transform/RTBoundValidator.java b/src/java/org/apache/cassandra/db/transform/RTBoundValidator.java
index eb37f4bc1c..40fc826f3f 100644
--- a/src/java/org/apache/cassandra/db/transform/RTBoundValidator.java
+++ b/src/java/org/apache/cassandra/db/transform/RTBoundValidator.java
@@ -17,14 +17,16 @@
  */
 package org.apache.cassandra.db.transform;
 
+import java.util.Objects;
+
 import org.apache.cassandra.db.DeletionTime;
 import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
 import org.apache.cassandra.db.rows.RangeTombstoneMarker;
 import org.apache.cassandra.db.rows.UnfilteredRowIterator;
-import org.apache.cassandra.schema.TableMetadata;
 
 /**
- * A validating transformation that sanity-checks the sequence of RT bounds and boundaries in every partition.
+ * A validating transformation that sanity-checks the sequence of Range Tombstone bounds and boundaries in every
+ * partition.
  *
  * What we validate, specifically:
  * - that open markers are only followed by close markers
@@ -51,29 +53,27 @@ public final class RTBoundValidator extends Transformation<UnfilteredRowIterator
 
     public static UnfilteredRowIterator validate(UnfilteredRowIterator partition, Stage stage, boolean enforceIsClosed)
     {
-        return Transformation.apply(partition, new RowsTransformation(stage, partition.metadata(), partition.isReverseOrder(), enforceIsClosed));
+        return Transformation.apply(partition, new RowsTransformation(stage, partition, enforceIsClosed));
     }
 
     @Override
     public UnfilteredRowIterator applyToPartition(UnfilteredRowIterator partition)
     {
-        return Transformation.apply(partition, new RowsTransformation(stage, partition.metadata(), partition.isReverseOrder(), enforceIsClosed));
+        return Transformation.apply(partition, new RowsTransformation(stage, partition, enforceIsClosed));
     }
 
     private final static class RowsTransformation extends Transformation
     {
         private final Stage stage;
-        private final TableMetadata metadata;
-        private final boolean isReverseOrder;
         private final boolean enforceIsClosed;
+        private final UnfilteredRowIterator partition;
 
         private DeletionTime openMarkerDeletionTime;
 
-        private RowsTransformation(Stage stage, TableMetadata metadata, boolean isReverseOrder, boolean enforceIsClosed)
+        private RowsTransformation(Stage stage, UnfilteredRowIterator partition, boolean enforceIsClosed)
         {
             this.stage = stage;
-            this.metadata = metadata;
-            this.isReverseOrder = isReverseOrder;
+            this.partition = partition;
             this.enforceIsClosed = enforceIsClosed;
         }
 
@@ -83,25 +83,25 @@ public final class RTBoundValidator extends Transformation<UnfilteredRowIterator
             if (null == openMarkerDeletionTime)
             {
                  // there is no open RT in the stream - we are expecting a *_START_BOUND
-                if (marker.isClose(isReverseOrder))
-                    throw ise("unexpected end bound or boundary " + marker.toString(metadata));
+                if (marker.isClose(partition.isReverseOrder()))
+                    throw ise("unexpected end bound or boundary " + marker.toString(partition.metadata()));
             }
             else
             {
                 // there is an open RT in the stream - we are expecting a *_BOUNDARY or an *_END_BOUND
-                if (!marker.isClose(isReverseOrder))
-                    throw ise("start bound followed by another start bound " + marker.toString(metadata));
+                if (!marker.isClose(partition.isReverseOrder()))
+                    throw ise("start bound followed by another start bound " + marker.toString(partition.metadata()));
 
                 // deletion times of open/close markers must match
-                DeletionTime deletionTime = marker.closeDeletionTime(isReverseOrder);
+                DeletionTime deletionTime = marker.closeDeletionTime(partition.isReverseOrder());
                 if (!deletionTime.equals(openMarkerDeletionTime))
-                    throw ise("open marker and close marker have different deletion times");
+                    throw ise("open marker and close marker have different deletion times, close=" + deletionTime);
 
                 openMarkerDeletionTime = null;
             }
 
-            if (marker.isOpen(isReverseOrder))
-                openMarkerDeletionTime = marker.openDeletionTime(isReverseOrder);
+            if (marker.isOpen(partition.isReverseOrder()))
+                openMarkerDeletionTime = marker.openDeletionTime(partition.isReverseOrder());
 
             return marker;
         }
@@ -110,14 +110,22 @@ public final class RTBoundValidator extends Transformation<UnfilteredRowIterator
         public void onPartitionClose()
         {
             if (enforceIsClosed && null != openMarkerDeletionTime)
-                throw ise("expected all RTs to be closed, but the last one is open");
+                throw ise("expected all Range Tombstones to be closed, but the last one is open");
         }
 
         private IllegalStateException ise(String why)
         {
-            String message =
-                String.format("%s UnfilteredRowIterator for %s has an illegal RT bounds sequence: %s", stage, metadata, why);
-            throw new IllegalStateException(message);
+            throw new IllegalStateException(message(why));
+        }
+
+        private String message(String why)
+        {
+            return String.format("%s UnfilteredRowIterator for %s (key: %s omdt: [%s]) has an illegal Range Tombstone bounds sequence: %s",
+                                 stage,
+                                 partition.metadata(),
+                                 partition.metadata().partitionKeyType.getString(partition.partitionKey().getKey()),
+                                 Objects.toString(openMarkerDeletionTime, "not present"),
+                                 why);
         }
     }
 }
diff --git a/src/java/org/apache/cassandra/exceptions/QueryCancelledException.java b/src/java/org/apache/cassandra/exceptions/QueryCancelledException.java
new file mode 100644
index 0000000000..73b2d9c718
--- /dev/null
+++ b/src/java/org/apache/cassandra/exceptions/QueryCancelledException.java
@@ -0,0 +1,28 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.cassandra.exceptions;
+
+import org.apache.cassandra.db.ReadCommand;
+
+public class QueryCancelledException extends RuntimeException
+{
+    public QueryCancelledException(ReadCommand command)
+    {
+        super("Query cancelled for taking too long: " + command.toCQLString());
+    }
+}
diff --git a/src/java/org/apache/cassandra/service/StorageProxy.java b/src/java/org/apache/cassandra/service/StorageProxy.java
index 41703c439c..266b0b917a 100644
--- a/src/java/org/apache/cassandra/service/StorageProxy.java
+++ b/src/java/org/apache/cassandra/service/StorageProxy.java
@@ -90,6 +90,7 @@ import org.apache.cassandra.exceptions.CasWriteUnknownResultException;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.exceptions.IsBootstrappingException;
 import org.apache.cassandra.exceptions.OverloadedException;
+import org.apache.cassandra.exceptions.QueryCancelledException;
 import org.apache.cassandra.exceptions.ReadAbortException;
 import org.apache.cassandra.exceptions.ReadFailureException;
 import org.apache.cassandra.exceptions.ReadTimeoutException;
@@ -2171,10 +2172,16 @@ public class StorageProxy implements StorageProxyMBean
                 {
                     if (!command.isTrackingWarnings())
                         throw e;
-                    
+
                     response = command.createEmptyResponse();
                     readRejected = true;
                 }
+                catch (QueryCancelledException e)
+                {
+                    logger.debug("Query cancelled (timeout)", e);
+                    response = null;
+                    Preconditions.checkState(!command.isCompleted(), "Local read marked as completed despite being aborted by timeout to table %s", command.metadata());
+                }
 
                 if (command.complete())
                 {
diff --git a/test/distributed/org/apache/cassandra/distributed/test/TimeoutAbortTest.java b/test/distributed/org/apache/cassandra/distributed/test/TimeoutAbortTest.java
new file mode 100644
index 0000000000..1e0fca9b2e
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/TimeoutAbortTest.java
@@ -0,0 +1,62 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.cassandra.distributed.test;
+
+import java.io.IOException;
+import java.util.List;
+
+import org.junit.Test;
+
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class TimeoutAbortTest extends TestBaseImpl
+{
+    @Test
+    public void timeoutTest() throws IOException, InterruptedException
+    {
+        System.setProperty("cassandra.test.read_iteration_delay_ms", "5000");
+        try (Cluster cluster = init(Cluster.build(1).start()))
+        {
+            cluster.schemaChange(withKeyspace("create table %s.tbl (id int, ck1 int, ck2 int, d int, primary key (id, ck1, ck2))"));
+            cluster.coordinator(1).execute(withKeyspace("delete from %s.tbl using timestamp 5 where id = 1 and ck1 = 77 "), ConsistencyLevel.ALL);
+            cluster.get(1).flush(KEYSPACE);
+            Thread.sleep(1000);
+            for (int i = 0; i < 100; i++)
+                cluster.coordinator(1).execute(withKeyspace("insert into %s.tbl (id, ck1, ck2, d) values (1,77,?,1) using timestamp 10"), ConsistencyLevel.ALL, i);
+            cluster.get(1).flush(KEYSPACE);
+            boolean caughtException = false;
+            try
+            {
+                cluster.coordinator(1).execute(withKeyspace("select * from %s.tbl where id=1 and ck1 = 77"), ConsistencyLevel.ALL);
+            }
+            catch (Exception e)
+            {
+                assertTrue(e.getClass().getName().contains("ReadTimeoutException"));
+                caughtException = true;
+            }
+            assertTrue(caughtException);
+            List<String> errors = cluster.get(1).logs().grepForErrors().getResult();
+            assertFalse(errors.toString(), errors.stream().anyMatch(s -> s.contains("open RT bound")));
+        }
+    }
+}
diff --git a/test/unit/org/apache/cassandra/db/ReadCommandTest.java b/test/unit/org/apache/cassandra/db/ReadCommandTest.java
index 43a7952175..bf272b87cf 100644
--- a/test/unit/org/apache/cassandra/db/ReadCommandTest.java
+++ b/test/unit/org/apache/cassandra/db/ReadCommandTest.java
@@ -56,6 +56,7 @@ import org.apache.cassandra.db.rows.UnfilteredRowIterators;
 import org.apache.cassandra.dht.Range;
 import org.apache.cassandra.dht.Token;
 import org.apache.cassandra.exceptions.ConfigurationException;
+import org.apache.cassandra.exceptions.QueryCancelledException;
 import org.apache.cassandra.io.sstable.format.SSTableReader;
 import org.apache.cassandra.io.util.DataInputBuffer;
 import org.apache.cassandra.io.util.DataOutputBuffer;
@@ -232,7 +233,16 @@ public class ReadCommandTest
         assertEquals(2, Util.getAll(readCommand).size());
 
         readCommand.abort();
-        assertEquals(0, Util.getAll(readCommand).size());
+        boolean cancelled = false;
+        try
+        {
+            Util.getAll(readCommand);
+        }
+        catch (QueryCancelledException e)
+        {
+            cancelled = true;
+        }
+        assertTrue(cancelled);
     }
 
     @Test
@@ -263,7 +273,16 @@ public class ReadCommandTest
         assertEquals(2, partitions.get(0).rowCount());
 
         readCommand.abort();
-        assertEquals(0, Util.getAll(readCommand).size());
+        boolean cancelled = false;
+        try
+        {
+            Util.getAll(readCommand);
+        }
+        catch (QueryCancelledException e)
+        {
+            cancelled = true;
+        }
+        assertTrue(cancelled);
     }
 
     @Test
@@ -294,7 +313,16 @@ public class ReadCommandTest
         assertEquals(2, partitions.get(0).rowCount());
 
         readCommand.abort();
-        assertEquals(0, Util.getAll(readCommand).size());
+        boolean cancelled = false;
+        try
+        {
+            Util.getAll(readCommand);
+        }
+        catch (QueryCancelledException e)
+        {
+            cancelled = true;
+        }
+        assertTrue(cancelled);
     }
 
     @Test