CASSANDRA-21072: Prevent TombstoneOverwhelmingException from being swallowed when tracking warnings

When trackWarnings is enabled, ReadCommandVerbHandler catches all
RejectException subclasses and sends an empty success response with
failure info in MessageParams. TombstoneOverwhelmingException is a hard
abort that should propagate as a proper failure, not masquerade as
success. Re-throw it regardless of warning tracking state.

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

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

@@ -21,6 +21,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import org.apache.cassandra.config.DatabaseDescriptor;
+import org.apache.cassandra.db.filter.TombstoneOverwhelmingException;
 import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
 import org.apache.cassandra.dht.AbstractBounds;
 import org.apache.cassandra.dht.Token;
@@ -88,7 +89,7 @@ public class ReadCommandVerbHandler implements IVerbHandler<ReadCommand>
         }
         catch (RejectException e)
         {
-            if (!command.isTrackingWarnings())
+            if (!command.isTrackingWarnings() || e instanceof TombstoneOverwhelmingException)
                 throw e;
 
             // make sure to log as the exception is swallowed

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

@@ -33,6 +33,7 @@ import org.apache.cassandra.db.filter.ClusteringIndexSliceFilter;
 import org.apache.cassandra.db.filter.ColumnFilter;
 import org.apache.cassandra.db.filter.DataLimits;
 import org.apache.cassandra.db.filter.RowFilter;
+import org.apache.cassandra.db.filter.TombstoneOverwhelmingException;
 import org.apache.cassandra.db.partitions.UnfilteredPartitionIterator;
 import org.apache.cassandra.exceptions.InvalidRequestException;
 import org.apache.cassandra.locator.InetAddressAndPort;
@@ -146,6 +147,27 @@ public class ReadCommandVerbHandlerTest
                               .build());
     }
 
+    @Test(expected = TombstoneOverwhelmingException.class)
+    public void tombstoneExceptionPropagatesWithoutWarningTracking()
+    {
+        ReadCommand command = new TombstoneThrowingReadCommand(metadata);
+        handler.doVerb(Message.builder(READ_REQ, command)
+                              .from(peer())
+                              .withId(messageId())
+                              .build());
+    }
+
+    @Test(expected = TombstoneOverwhelmingException.class)
+    public void tombstoneExceptionPropagatesWithWarningTracking()
+    {
+        ReadCommand command = new TombstoneThrowingReadCommand(metadata);
+        handler.doVerb(Message.builder(READ_REQ, command)
+                              .from(peer())
+                              .withFlag(MessageFlag.TRACK_WARNINGS)
+                              .withId(messageId())
+                              .build());
+    }
+
     private static int messageId()
     {
         return random.nextInt();
@@ -199,6 +221,34 @@ public class ReadCommandVerbHandlerTest
         }
     }
 
+    private static class TombstoneThrowingReadCommand extends SinglePartitionReadCommand
+    {
+        TombstoneThrowingReadCommand(TableMetadata metadata)
+        {
+            super(metadata.epoch,
+                  false,
+                  0,
+                  false,
+                  PotentialTxnConflicts.DISALLOW,
+                  metadata,
+                  FBUtilities.nowInSeconds(),
+                  ColumnFilter.all(metadata),
+                  RowFilter.none(),
+                  DataLimits.NONE,
+                  KEY,
+                  new ClusteringIndexSliceFilter(Slices.ALL, false),
+                  null,
+                  false,
+                  null);
+        }
+
+        @Override
+        public UnfilteredPartitionIterator executeLocally(ReadExecutionController executionController)
+        {
+            throw new TombstoneOverwhelmingException(1000, "test_query", metadata(), KEY, Clustering.EMPTY);
+        }
+    }
+
     private static DecoratedKey key(TableMetadata metadata, int key)
     {
         return metadata.partitioner.decorateKey(ByteBufferUtil.bytes(key));