warn, then drop queries exceeding a configurable number of tombstones

patch by jbellis; reviewed by slebresne for CASSANDRA-6117

CHANGES.txt

@@ -1,4 +1,5 @@
 2.0.2
+ * drop queries exceeding a configurable number of tombstones (CASSANDRA-6117)
  * Track and persist sstable read activity (CASSANDRA-5515)
  * Fixes for speculative retry (CASSANDRA-5932)
  * Improve memory usage of metadata min/max column names (CASSANDRA-6077)

NEWS.txt

@@ -22,6 +22,14 @@ New features
     - Speculative retry defaults to 99th percentile
       (See blog post at TODO)
 
+Upgrading
+---------
+    - tombstone_debug_threshold from 1.2.11 has been changed to
+      tombstone_warn_threshold and tombstone_failure_threshold.
+      Adjust these if your application relies on scanning a large
+      number of tombstones; see the comments in cassandra.yaml for why
+      this is dangerous.
+
 
 2.0.1
 =====

conf/cassandra.yaml

@@ -404,10 +404,17 @@ snapshot_before_compaction: false
 # lose data on truncation or drop.
 auto_snapshot: true
 
-# Log a debug message if more than this many tombstones are scanned
-# in a single-partition query.  Set the threshold on SliceQueryFilter
-# to debug to enable.
-tombstone_debug_threshold: 10000
+# When executing a scan, within or across a partition, we need to keep the
+# tombstones seen in memory so we can return them to the coordinator, which
+# will use them to make sure other replicas also know about the deleted rows.
+# With workloads that generate a lot of tombstones, this can cause performance
+# problems and even exaust the server heap.
+# (http://www.datastax.com/dev/blog/cassandra-anti-patterns-queues-and-queue-like-datasets)
+# Adjust the thresholds here if you understand the dangers and want to
+# scan more tombstones anyway.  These thresholds may also be adjusted at runtime
+# using the StorageService mbean.
+tombstone_warn_threshold: 1000
+tombstone_failure_threshold: 100000
 
 # Add column indexes to a row after its contents reach this size.
 # Increase if your column values are large, or if you have a very large

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

@@ -179,7 +179,8 @@ public class Config
 
     private static boolean outboundBindAny = false;
 
-    public volatile int tombstone_debug_threshold = 10000;
+    public volatile int tombstone_warn_threshold = 1000;
+    public volatile int tombstone_failure_threshold = 100000;
 
     public static boolean getOutboundBindAny()
     {

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

@@ -890,18 +890,24 @@ public class DatabaseDescriptor
         return conf.commitlog_directory;
     }
 
-    /**
-     * How many tombstones need to be scanned before we log a
-     * debug message
-     */
-    public static int getTombstoneDebugThreshold()
+    public static int getTombstoneWarnThreshold()
     {
-        return conf.tombstone_debug_threshold;
+        return conf.tombstone_warn_threshold;
     }
 
-    public static void setTombstoneDebugThreshold(int tombstoneDebugThreshold)
+    public static void setTombstoneWarnThreshold(int threshold)
     {
-        conf.tombstone_debug_threshold = tombstoneDebugThreshold;
+        conf.tombstone_warn_threshold = threshold;
+    }
+
+    public static int getTombstoneFailureThreshold()
+    {
+        return conf.tombstone_failure_threshold;
+    }
+
+    public static void setTombstoneFailureThreshold(int threshold)
+    {
+        conf.tombstone_failure_threshold = threshold;
     }
 
     /**

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

@@ -20,6 +20,7 @@ package org.apache.cassandra.db;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.cassandra.db.filter.TombstoneOverwhelmingException;
 import org.apache.cassandra.net.IVerbHandler;
 import org.apache.cassandra.net.MessageIn;
 import org.apache.cassandra.net.MessageOut;
@@ -40,7 +41,16 @@ public class ReadVerbHandler implements IVerbHandler<ReadCommand>
 
         ReadCommand command = message.payload;
         Keyspace keyspace = Keyspace.open(command.ksName);
-        Row row = command.getRow(keyspace);
+        Row row;
+        try
+        {
+            row = command.getRow(keyspace);
+        }
+        catch (TombstoneOverwhelmingException e)
+        {
+            // error already logged.  Drop the request
+            return;
+        }
 
         MessageOut<ReadResponse> reply = new MessageOut<ReadResponse>(MessagingService.Verb.REQUEST_RESPONSE,
                                                                       getResponse(command, row),

src/java/org/apache/cassandra/db/columniterator/IdentityQueryFilter.java

@@ -29,4 +29,10 @@ public class IdentityQueryFilter extends SliceQueryFilter
     {
         super(ByteBufferUtil.EMPTY_BYTE_BUFFER, ByteBufferUtil.EMPTY_BYTE_BUFFER, false, Integer.MAX_VALUE);
     }
+
+    @Override
+    protected boolean respectTombstoneFailures()
+    {
+        return false;
+    }
 }

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

@@ -194,12 +194,24 @@ public class SliceQueryFilter implements IDiskAtomFilter
             if (columnCounter.live() > count)
                 break;
 
+            if (respectTombstoneFailures() && columnCounter.ignored() > DatabaseDescriptor.getTombstoneFailureThreshold())
+            {
+                Tracing.trace("Scanned over {} tombstones; query aborted (see tombstone_fail_threshold)", DatabaseDescriptor.getTombstoneFailureThreshold());
+                logger.error("Scanned over {} tombstones; query aborted (see tombstone_fail_threshold)", DatabaseDescriptor.getTombstoneFailureThreshold());
+                throw new TombstoneOverwhelmingException();
+            }
+
             container.addIfRelevant(column, tester, gcBefore);
         }
 
         Tracing.trace("Read {} live and {} tombstoned cells", columnCounter.live(), columnCounter.ignored());
-        if (columnCounter.ignored() > DatabaseDescriptor.getTombstoneDebugThreshold())
-            logger.debug("Read {} live and {} tombstoned cells", columnCounter.live(), columnCounter.ignored());
+        if (columnCounter.ignored() > DatabaseDescriptor.getTombstoneWarnThreshold())
+            logger.warn("Read {} live and {} tombstoned cells (see tombstone_warn_threshold)", columnCounter.live(), columnCounter.ignored());
+    }
+
+    protected boolean respectTombstoneFailures()
+    {
+        return true;
     }
 
     public int getLiveCount(ColumnFamily cf, long now)

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

@@ -0,0 +1,5 @@
+package org.apache.cassandra.db.filter;
+
+public class TombstoneOverwhelmingException extends RuntimeException
+{
+}

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

@@ -19,6 +19,7 @@ package org.apache.cassandra.service;
 
 import org.apache.cassandra.db.AbstractRangeCommand;
 import org.apache.cassandra.db.RangeSliceReply;
+import org.apache.cassandra.db.filter.TombstoneOverwhelmingException;
 import org.apache.cassandra.net.IVerbHandler;
 import org.apache.cassandra.net.MessageIn;
 import org.apache.cassandra.net.MessagingService;
@@ -39,6 +40,10 @@ public class RangeSliceVerbHandler implements IVerbHandler<AbstractRangeCommand>
             Tracing.trace("Enqueuing response to {}", message.from);
             MessagingService.instance().sendReply(reply.createMessage(), id, message.from);
         }
+        catch (TombstoneOverwhelmingException e)
+        {
+            // error already logged.  Drop the request
+        }
         catch (Exception ex)
         {
             throw new RuntimeException(ex);

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

@@ -3654,13 +3654,23 @@ public class StorageService extends NotificationBroadcasterSupport implements IE
         return DatabaseDescriptor.getPartitionerName();
     }
 
-    public int getTombstoneDebugThreshold()
+    public int getTombstoneWarnThreshold()
     {
-        return DatabaseDescriptor.getTombstoneDebugThreshold();
+        return DatabaseDescriptor.getTombstoneWarnThreshold();
     }
 
-    public void setTombstoneDebugThreshold(int tombstoneDebugThreshold)
+    public void setTombstoneWarnThreshold(int threshold)
     {
-        DatabaseDescriptor.setTombstoneDebugThreshold(tombstoneDebugThreshold);
+        DatabaseDescriptor.setTombstoneWarnThreshold(threshold);
+    }
+
+    public int getTombstoneFailureThreshold()
+    {
+        return DatabaseDescriptor.getTombstoneFailureThreshold();
+    }
+
+    public void setTombstoneFailureThreshold(int threshold)
+    {
+        DatabaseDescriptor.setTombstoneFailureThreshold(threshold);
     }
 }

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

@@ -477,8 +477,13 @@ public interface StorageServiceMBean extends NotificationEmitter
     /** Returns the cluster partitioner */
     public String getPartitionerName();
 
-    /** Returns the threshold for returning debugging queries with many tombstones */
-    public int getTombstoneDebugThreshold();
-    /** Sets the threshold for returning debugging queries with many tombstones */
-    public void setTombstoneDebugThreshold(int tombstoneDebugThreshold);
+    /** Returns the threshold for warning of queries with many tombstones */
+    public int getTombstoneWarnThreshold();
+    /** Sets the threshold for warning queries with many tombstones */
+    public void setTombstoneWarnThreshold(int tombstoneDebugThreshold);
+
+    /** Returns the threshold for abandoning queries with many tombstones */
+    public int getTombstoneFailureThreshold();
+    /** Sets the threshold for abandoning queries with many tombstones */
+    public void setTombstoneFailureThreshold(int tombstoneDebugThreshold);
 }