diff --git a/CHANGES.txt b/CHANGES.txt
index 99a53404b0c2..0882bbaa77a6 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -1,4 +1,5 @@
 5.0
+ * Add guardrail for partition tombstones and deprecate compaction_tombstone_warning_threshold (CASSANDRA-17194)
  * Print header and statistics for cassandra-stress output with arbitrary frequency (CASSANDRA-12972)
  * CEP-25: Trie-indexed SSTable format (CASSANDRA-18398)
  * Make cassandra-stress able to read all credentials from a file (CASSANDRA-18544)
diff --git a/NEWS.txt b/NEWS.txt
index 2129ead9156a..294e57d1f837 100644
--- a/NEWS.txt
+++ b/NEWS.txt
@@ -83,6 +83,7 @@ New features
       - Whether DROP KEYSPACE commands are allowed.
       - Column value size
       - Partition size
+      - Partition tombstones
     - It is possible to list ephemeral snapshots by nodetool listsnaphots command when flag "-e" is specified.
     - Added a new flag to `nodetool profileload` and JMX endpoint to set up recurring profile load generation on specified
       intervals (see CASSANDRA-17821)
@@ -193,6 +194,9 @@ Deprecation
     - The config property `compaction_large_partition_warning_threshold` has been deprecated in favour of the new
       guardrail for partition size. That guardrail is based on the properties `partition_size_warn_threshold` and
       `partition_size_fail_threshold`. The warn threshold has a very similar behaviour to the old config property.
+    - The config property `compaction_tombstone_warning_threshold` has been deprecated in favour of the new guardrail
+      for partition tombstones. That guardrail is based on the properties `partition_tombstones_warn_threshold` and
+      `partition_tombstones_fail_threshold`. The warn threshold has a very similar behaviour to the old config property.
 
 4.1
 ===
diff --git a/conf/cassandra.yaml b/conf/cassandra.yaml
index ca865b19dace..ce39c019eee3 100644
--- a/conf/cassandra.yaml
+++ b/conf/cassandra.yaml
@@ -1564,7 +1564,8 @@ unlogged_batch_across_partitions_warn_threshold: 10
 # As of Cassandra 5.0, this property is deprecated in favour of partition_size_warn_threshold.
 compaction_large_partition_warning_threshold: 100MiB
 
-# Log a warning when writing more tombstones than this value to a partition
+# Log a warning when writing more tombstones than this value to a partition.
+# As of Cassandra 5.0, this property is deprecated in favour of partition_tombstones_warn_threshold.
 compaction_tombstone_warning_threshold: 100000
 
 # GC Pauses greater than 200 ms will be logged at INFO level
@@ -1842,6 +1843,15 @@ drop_compact_storage_enabled: false
 # partition_size_warn_threshold:
 # partition_size_fail_threshold:
 #
+# Guardrail to warn or fail when writing partitions with more tombstones than threshold.
+# The guardrail is only checked when writing sstables (flush and compaction), and exceeding the fail threshold on that
+# moment will only log an error message, without interrupting the operation.
+# This operates on a per-sstable basis, so it won't detect a large partition if it is spread across multiple sstables.
+# The warning threshold replaces the deprecated config property compaction_tombstone_warning_threshold.
+# The two thresholds default to -1 to disable.
+# partition_tombstones_warn_threshold: -1
+# partition_tombstones_fail_threshold: -1
+#
 # Guardrail to warn or fail when writing column values larger than threshold.
 # This guardrail is only applied to the values of regular columns because both the serialized partitions keys and the
 # values of the components of the clustering key already have a fixed, relatively small size limit of 65535 bytes, which
diff --git a/src/java/org/apache/cassandra/config/Config.java b/src/java/org/apache/cassandra/config/Config.java
index 6e5f8739f701..56ecb199d411 100644
--- a/src/java/org/apache/cassandra/config/Config.java
+++ b/src/java/org/apache/cassandra/config/Config.java
@@ -332,6 +332,7 @@ public MemtableOptions()
     public volatile DataStorageSpec.IntMebibytesBound compaction_large_partition_warning_threshold = new DataStorageSpec.IntMebibytesBound("100MiB");
     @Replaces(oldName = "min_free_space_per_drive_in_mb", converter = Converters.MEBIBYTES_DATA_STORAGE_INT, deprecated = true)
     public DataStorageSpec.IntMebibytesBound min_free_space_per_drive = new DataStorageSpec.IntMebibytesBound("50MiB");
+    @Deprecated
     public volatile Integer compaction_tombstone_warning_threshold = 100000;
 
     // fraction of free disk space available for compaction after min free space is subtracted
@@ -875,6 +876,8 @@ public static void setClientMode(boolean clientMode)
     public volatile boolean simplestrategy_enabled = true;
     public volatile DataStorageSpec.LongBytesBound partition_size_warn_threshold = null;
     public volatile DataStorageSpec.LongBytesBound partition_size_fail_threshold = null;
+    public volatile long partition_tombstones_warn_threshold = -1;
+    public volatile long partition_tombstones_fail_threshold = -1;
     public volatile DataStorageSpec.LongBytesBound column_value_size_warn_threshold = null;
     public volatile DataStorageSpec.LongBytesBound column_value_size_fail_threshold = null;
     public volatile DataStorageSpec.LongBytesBound collection_size_warn_threshold = null;
diff --git a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
index 2531e5c6e47a..424ac67cf0d8 100644
--- a/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
+++ b/src/java/org/apache/cassandra/config/DatabaseDescriptor.java
@@ -2178,11 +2178,13 @@ public static long getCompactionLargePartitionWarningThreshold()
         return conf.compaction_large_partition_warning_threshold.toBytesInLong();
     }
 
+    @Deprecated
     public static int getCompactionTombstoneWarningThreshold()
     {
         return conf.compaction_tombstone_warning_threshold;
     }
 
+    @Deprecated
     public static void setCompactionTombstoneWarningThreshold(int count)
     {
         conf.compaction_tombstone_warning_threshold = count;
diff --git a/src/java/org/apache/cassandra/config/GuardrailsOptions.java b/src/java/org/apache/cassandra/config/GuardrailsOptions.java
index 7affe6f7cfd8..f58408e8b213 100644
--- a/src/java/org/apache/cassandra/config/GuardrailsOptions.java
+++ b/src/java/org/apache/cassandra/config/GuardrailsOptions.java
@@ -77,6 +77,7 @@ public GuardrailsOptions(Config config)
         config.write_consistency_levels_warned = validateConsistencyLevels(config.write_consistency_levels_warned, "write_consistency_levels_warned");
         config.write_consistency_levels_disallowed = validateConsistencyLevels(config.write_consistency_levels_disallowed, "write_consistency_levels_disallowed");
         validateSizeThreshold(config.partition_size_warn_threshold, config.partition_size_fail_threshold, false, "partition_size");
+        validateMaxLongThreshold(config.partition_tombstones_warn_threshold, config.partition_tombstones_fail_threshold, "partition_tombstones", false);
         validateSizeThreshold(config.column_value_size_warn_threshold, config.column_value_size_fail_threshold, false, "column_value_size");
         validateSizeThreshold(config.collection_size_warn_threshold, config.collection_size_fail_threshold, false, "collection_size");
         validateMaxIntThreshold(config.items_per_collection_warn_threshold, config.items_per_collection_fail_threshold, "items_per_collection");
@@ -567,6 +568,31 @@ public void setPartitionSizeThreshold(@Nullable DataStorageSpec.LongBytesBound w
                                   x -> config.partition_size_fail_threshold = x);
     }
 
+    @Override
+    public long getPartitionTombstonesWarnThreshold()
+    {
+        return config.partition_tombstones_warn_threshold;
+    }
+
+    @Override
+    public long getPartitionTombstonesFailThreshold()
+    {
+        return config.partition_tombstones_fail_threshold;
+    }
+
+    public void setPartitionTombstonesThreshold(long warn, long fail)
+    {
+        validateMaxLongThreshold(warn, fail, "partition_tombstones", false);
+        updatePropertyWithLogging("partition_tombstones_warn_threshold",
+                                  warn,
+                                  () -> config.partition_tombstones_warn_threshold,
+                                  x -> config.partition_tombstones_warn_threshold = x);
+        updatePropertyWithLogging("partition_tombstones_fail_threshold",
+                                  fail,
+                                  () -> config.partition_tombstones_fail_threshold,
+                                  x -> config.partition_tombstones_fail_threshold = x);
+    }
+
     @Override
     @Nullable
     public DataStorageSpec.LongBytesBound getColumnValueSizeWarnThreshold()
diff --git a/src/java/org/apache/cassandra/db/guardrails/Guardrails.java b/src/java/org/apache/cassandra/db/guardrails/Guardrails.java
index d2289b4bb6dc..4f736337e807 100644
--- a/src/java/org/apache/cassandra/db/guardrails/Guardrails.java
+++ b/src/java/org/apache/cassandra/db/guardrails/Guardrails.java
@@ -316,13 +316,25 @@ public final class Guardrails implements GuardrailsMBean
      */
     public static final MaxThreshold partitionSize =
     new MaxThreshold("partition_size",
-                     "Too large partitions can cause performance problems. ",
+                     "Too large partitions can cause performance problems.",
                      state -> sizeToBytes(CONFIG_PROVIDER.getOrCreate(state).getPartitionSizeWarnThreshold()),
                      state -> sizeToBytes(CONFIG_PROVIDER.getOrCreate(state).getPartitionSizeFailThreshold()),
                      (isWarning, what, value, threshold) ->
                              format("Partition %s has size %s, this exceeds the %s threshold of %s.",
                                     what, value, isWarning ? "warning" : "failure", threshold));
 
+    /**
+     * Guardrail on the number of rows of a partition.
+     */
+    public static final MaxThreshold partitionTombstones =
+    new MaxThreshold("partition_tombstones",
+                     "Partitions with too many tombstones can cause performance problems.",
+                     state -> CONFIG_PROVIDER.getOrCreate(state).getPartitionTombstonesWarnThreshold(),
+                     state -> CONFIG_PROVIDER.getOrCreate(state).getPartitionTombstonesFailThreshold(),
+                     (isWarning, what, value, threshold) ->
+                             format("Partition %s has %s tombstones, this exceeds the %s threshold of %s.",
+                                    what, value, isWarning ? "warning" : "failure", threshold));
+
     /**
      * Guardrail on the size of a collection.
      */
@@ -823,6 +835,24 @@ public void setPartitionSizeThreshold(@Nullable String warnSize, @Nullable Strin
         DEFAULT_CONFIG.setPartitionSizeThreshold(sizeFromString(warnSize), sizeFromString(failSize));
     }
 
+    @Override
+    public long getPartitionTombstonesWarnThreshold()
+    {
+        return DEFAULT_CONFIG.getPartitionTombstonesWarnThreshold();
+    }
+
+    @Override
+    public long getPartitionTombstonesFailThreshold()
+    {
+        return DEFAULT_CONFIG.getPartitionTombstonesFailThreshold();
+    }
+
+    @Override
+    public void setPartitionTombstonesThreshold(long warn, long fail)
+    {
+        DEFAULT_CONFIG.setPartitionTombstonesThreshold(warn, fail);
+    }
+
     @Override
     @Nullable
     public String getColumnValueSizeWarnThreshold()
diff --git a/src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java b/src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java
index af1b5b48c029..96390665950a 100644
--- a/src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java
+++ b/src/java/org/apache/cassandra/db/guardrails/GuardrailsConfig.java
@@ -250,6 +250,16 @@ public interface GuardrailsConfig
     @Nullable
     DataStorageSpec.LongBytesBound getPartitionSizeFailThreshold();
 
+    /**
+     * @return The threshold to warn when writing partitions with more tombstones than threshold.
+     */
+    long getPartitionTombstonesWarnThreshold();
+
+    /**
+     * @return The threshold to fail when writing partitions with more tombstones than threshold.
+     */
+    long getPartitionTombstonesFailThreshold();
+
     /**
      * @return The threshold to warn when writing column values larger than threshold.
      */
diff --git a/src/java/org/apache/cassandra/db/guardrails/GuardrailsMBean.java b/src/java/org/apache/cassandra/db/guardrails/GuardrailsMBean.java
index cabbe0c59e5c..0c8ff344abe2 100644
--- a/src/java/org/apache/cassandra/db/guardrails/GuardrailsMBean.java
+++ b/src/java/org/apache/cassandra/db/guardrails/GuardrailsMBean.java
@@ -372,13 +372,13 @@ public interface GuardrailsMBean
      * @return The threshold to warn when an IN query creates a cartesian product with a size exceeding threshold.
      * -1 means disabled.
      */
-    public int getInSelectCartesianProductWarnThreshold();
+    int getInSelectCartesianProductWarnThreshold();
 
     /**
      * @return The threshold to prevent IN queries creating a cartesian product with a size exceeding threshold.
      * -1 means disabled.
      */
-    public int getInSelectCartesianProductFailThreshold();
+    int getInSelectCartesianProductFailThreshold();
 
     /**
      * @param warn The threshold to warn when an IN query creates a cartesian product with a size exceeding threshold.
@@ -386,7 +386,7 @@ public interface GuardrailsMBean
      * @param fail The threshold to prevent IN queries creating a cartesian product with a size exceeding threshold.
      *             -1 means disabled.
      */
-    public void setInSelectCartesianProductThreshold(int warn, int fail);
+    void setInSelectCartesianProductThreshold(int warn, int fail);
 
     /**
      * @return consistency levels that are warned about when reading.
@@ -495,6 +495,26 @@ public interface GuardrailsMBean
      */
     void setPartitionSizeThreshold(@Nullable String warnSize, @Nullable String failSize);
 
+    /**
+     * @return The threshold to warn when encountering partitions with more tombstones than threshold. -1 means disabled.
+     */
+    long getPartitionTombstonesWarnThreshold();
+
+    /**
+     * @return The threshold to fail when encountering partitions with more tombstones than threshold. -1 means disabled.
+     * Triggering a failure emits a log message and a diagnostic event, but it doesn't throw an exception interrupting
+     * the offending sstable write.
+     */
+    long getPartitionTombstonesFailThreshold();
+
+    /**
+     * @param warn The threshold to warn when encountering partitions with more tombstones than threshold. -1 means disabled.
+     * @param fail The threshold to fail when encountering partitions with more tombstones than threshold. -1 means disabled.
+     *             Triggering a failure emits a log message and a diagnostic event, but it desn't throw an exception
+     *             interrupting the offending sstable write.
+     */
+    void setPartitionTombstonesThreshold(long warn, long fail);
+
     /**
      * @return The threshold to warn when encountering column values larger than threshold, as a string  formatted as
      * in, for example, {@code 10GiB}, {@code 20MiB}, {@code 30KiB} or {@code 40B}. A {@code null} value means disabled.
diff --git a/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java b/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java
index 377169faf077..2deda2a6f772 100644
--- a/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java
+++ b/src/java/org/apache/cassandra/io/sstable/format/SortedTableWriter.java
@@ -32,6 +32,7 @@
 import org.apache.cassandra.db.DeletionPurger;
 import org.apache.cassandra.db.DeletionTime;
 import org.apache.cassandra.db.guardrails.Guardrails;
+import org.apache.cassandra.db.guardrails.Threshold;
 import org.apache.cassandra.db.lifecycle.LifecycleNewTracker;
 import org.apache.cassandra.db.rows.ComplexColumnData;
 import org.apache.cassandra.db.rows.PartitionSerializationException;
@@ -213,7 +214,8 @@ private AbstractRowIndexEntry endPartition(DecoratedKey key, DeletionTime partit
 
         long endPosition = dataWriter.position();
         long rowSize = endPosition - partitionWriter.getInitialPosition();
-        guardPartitionSize(key, rowSize);
+        guardPartitionThreshold(Guardrails.partitionSize, key, rowSize);
+        guardPartitionThreshold(Guardrails.partitionTombstones, key, metadataCollector.totalTombstones);
         maybeLogLargePartitionWarning(key, rowSize);
         maybeLogManyTombstonesWarning(key, metadataCollector.totalTombstones);
         metadataCollector.addPartitionSizeInBytes(rowSize);
@@ -325,19 +327,20 @@ protected FileHandle openDataFile(long lengthOverride, StatsMetadata statsMetada
         return dataFile;
     }
 
-    private void guardPartitionSize(DecoratedKey key, long rowSize)
+    private void guardPartitionThreshold(Threshold guardrail, DecoratedKey key, long size)
     {
-        if (Guardrails.partitionSize.triggersOn(rowSize, null))
+        if (guardrail.triggersOn(size, null))
         {
-            String what = String.format("%s.%s:%s on sstable %s",
-                                        metadata.keyspace,
-                                        metadata.name,
-                                        metadata().partitionKeyType.getString(key.getKey()),
-                                        getFilename());
-            Guardrails.partitionSize.guard(rowSize, what, true, null);
+            String message = String.format("%s.%s:%s on sstable %s",
+                                           metadata.keyspace,
+                                           metadata.name,
+                                           metadata().partitionKeyType.getString(key.getKey()),
+                                           getFilename());
+            guardrail.guard(size, message, true, null);
         }
     }
 
+    @Deprecated
     private void maybeLogLargePartitionWarning(DecoratedKey key, long rowSize)
     {
         if (rowSize > DatabaseDescriptor.getCompactionLargePartitionWarningThreshold())
@@ -347,6 +350,7 @@ private void maybeLogLargePartitionWarning(DecoratedKey key, long rowSize)
         }
     }
 
+    @Deprecated
     private void maybeLogManyTombstonesWarning(DecoratedKey key, int tombstoneCount)
     {
         if (tombstoneCount > DatabaseDescriptor.getCompactionTombstoneWarningThreshold())
diff --git a/src/java/org/apache/cassandra/service/StorageService.java b/src/java/org/apache/cassandra/service/StorageService.java
index 5e6671857793..5f99fef4bcf1 100644
--- a/src/java/org/apache/cassandra/service/StorageService.java
+++ b/src/java/org/apache/cassandra/service/StorageService.java
@@ -6949,6 +6949,7 @@ public void setKeyspaceCountWarnThreshold(int value)
         DatabaseDescriptor.setKeyspaceCountWarnThreshold(value);
     }
 
+    @Override
     public void setCompactionTombstoneWarningThreshold(int count)
     {
         if (count < 0)
@@ -6957,6 +6958,7 @@ public void setCompactionTombstoneWarningThreshold(int count)
         DatabaseDescriptor.setCompactionTombstoneWarningThreshold(count);
     }
 
+    @Override
     public int getCompactionTombstoneWarningThreshold()
     {
         return DatabaseDescriptor.getCompactionTombstoneWarningThreshold();
diff --git a/src/java/org/apache/cassandra/service/StorageServiceMBean.java b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
index 5897afcce36a..154b0d86d303 100644
--- a/src/java/org/apache/cassandra/service/StorageServiceMBean.java
+++ b/src/java/org/apache/cassandra/service/StorageServiceMBean.java
@@ -1115,8 +1115,10 @@ public void enableAuditLog(String loggerName, String includedKeyspaces, String e
     @Deprecated
     void setKeyspaceCountWarnThreshold(int value);
 
-    public void setCompactionTombstoneWarningThreshold(int count);
-    public int getCompactionTombstoneWarningThreshold();
+    @Deprecated
+    void setCompactionTombstoneWarningThreshold(int count);
+    @Deprecated
+    int getCompactionTombstoneWarningThreshold();
 
     public boolean getReadThresholdsEnabled();
     public void setReadThresholdsEnabled(boolean value);
diff --git a/test/distributed/org/apache/cassandra/distributed/test/guardrails/GuardrailPartitionSizeTest.java b/test/distributed/org/apache/cassandra/distributed/test/guardrails/GuardrailPartitionSizeTest.java
index b2ee8d7d94d9..8dc49e14c941 100644
--- a/test/distributed/org/apache/cassandra/distributed/test/guardrails/GuardrailPartitionSizeTest.java
+++ b/test/distributed/org/apache/cassandra/distributed/test/guardrails/GuardrailPartitionSizeTest.java
@@ -75,12 +75,11 @@ protected Cluster getCluster()
     @Test
     public void testPartitionSize()
     {
+        // test yaml-loaded config
         testPartitionSize(WARN_THRESHOLD, FAIL_THRESHOLD);
-    }
+        schemaChange("DROP TABLE %s");
 
-    @Test
-    public void testPartitionSizeWithDynamicUpdate()
-    {
+        // test dynamic config
         int warn = WARN_THRESHOLD * 2;
         int fail = FAIL_THRESHOLD * 2;
         cluster.get(1).runOnInstance(() -> Guardrails.instance.setPartitionSizeThreshold(warn + "B", fail + "B"));
diff --git a/test/distributed/org/apache/cassandra/distributed/test/guardrails/GuardrailPartitionTombstonesTest.java b/test/distributed/org/apache/cassandra/distributed/test/guardrails/GuardrailPartitionTombstonesTest.java
new file mode 100644
index 000000000000..24045cad33e9
--- /dev/null
+++ b/test/distributed/org/apache/cassandra/distributed/test/guardrails/GuardrailPartitionTombstonesTest.java
@@ -0,0 +1,142 @@
+/*
+ * 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.guardrails;
+
+import java.io.IOException;
+
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import org.apache.cassandra.db.guardrails.Guardrails;
+import org.apache.cassandra.distributed.Cluster;
+import org.apache.cassandra.distributed.api.ConsistencyLevel;
+
+/**
+ * Tests the guardrail for the number of tombstones in a partition, {@link Guardrails#partitionTombstones}.
+ * <p>
+ * This test only includes the activation of the guardrail during sstable writes, focusing on the emmitted log messages.
+ * The tests for config, client warnings and diagnostic events are in
+ * {@link org.apache.cassandra.db.guardrails.GuardrailPartitionTombstonesTest}.
+ */
+public class GuardrailPartitionTombstonesTest extends GuardrailTester
+{
+    private static final int WARN_THRESHOLD = 500; // high enough to exceed system tables, which aren't excluded
+    private static final int FAIL_THRESHOLD = WARN_THRESHOLD * 2;
+
+    private static Cluster cluster;
+
+    @BeforeClass
+    public static void setupCluster() throws IOException
+    {
+        cluster = init(Cluster.build(1)
+                              .withConfig(c -> c.set("partition_tombstones_warn_threshold", WARN_THRESHOLD)
+                                                .set("partition_tombstones_fail_threshold", FAIL_THRESHOLD)
+                                                .set("memtable_heap_space", "512MiB")) // avoids flushes
+                              .start());
+        cluster.disableAutoCompaction(KEYSPACE);
+    }
+
+    @AfterClass
+    public static void teardownCluster()
+    {
+        if (cluster != null)
+            cluster.close();
+    }
+
+    @Override
+    protected Cluster getCluster()
+    {
+        return cluster;
+    }
+
+    @Test
+    public void testPartitionTombstones()
+    {
+        // test yaml-loaded config
+        testPartitionTombstones(WARN_THRESHOLD, FAIL_THRESHOLD);
+        schemaChange("DROP TABLE %s");
+
+        // test dynamic config
+        int warn = WARN_THRESHOLD + 10;
+        int fail = FAIL_THRESHOLD + 10;
+        cluster.get(1).runOnInstance(() -> Guardrails.instance.setPartitionTombstonesThreshold(warn, fail));
+        testPartitionTombstones(warn, fail);
+    }
+
+    private void testPartitionTombstones(int warn, int fail)
+    {
+        schemaChange("CREATE TABLE %s (k int, c int, v int, PRIMARY KEY (k, c))");
+
+        // empty table
+        assertNotWarnedOnFlush();
+        assertNotWarnedOnCompact();
+
+        // keep partition tombstones lower than thresholds
+        populateTable(warn);
+        assertNotWarnedOnFlush();
+        assertNotWarnedOnCompact();
+
+        // exceed warn threshold
+        populateTable(warn + 1);
+        assertWarnedOnFlush(expectedMessages(1));
+        assertWarnedOnCompact(expectedMessages(1));
+
+        // exceed fail threshold
+        populateTable(fail + 1);
+        assertFailedOnFlush(expectedMessages(1));
+        assertFailedOnCompact(expectedMessages(1));
+
+        // remove most of the tombstones to be under the threshold again
+        execute("DELETE FROM %s WHERE k = 1 AND c > 1");
+        assertNotWarnedOnFlush();
+        assertNotWarnedOnCompact();
+
+        // exceed warn threshold in multiple partitions,
+        // this shouldn't trigger the fail threshold because the count is per-partition.
+        for (int c = 0; c <= warn; c++)
+        {
+            execute("DELETE FROM %s WHERE k = 2 AND c = ?", c);
+            execute("DELETE FROM %s WHERE k = 3 AND c = ?", c);
+        }
+        assertWarnedOnFlush(expectedMessages(2, 3));
+        assertWarnedOnCompact(expectedMessages(2, 3));
+    }
+
+    private void populateTable(int numTombstones)
+    {
+        for (int c = 0; c < numTombstones; c++)
+        {
+            execute("DELETE FROM %s WHERE k = 1 AND c = ?", c);
+        }
+    }
+
+    private void execute(String query, Object... args)
+    {
+        cluster.coordinator(1).execute(format(query), ConsistencyLevel.ALL, args);
+    }
+
+    private String[] expectedMessages(int... keys)
+    {
+        String[] messages = new String[keys.length];
+        for (int i = 0; i < keys.length; i++)
+            messages[i] = String.format("Guardrail partition_tombstones violated: Partition %s:%d", qualifiedTableName, keys[i]);
+        return messages;
+    }
+}
diff --git a/test/unit/org/apache/cassandra/db/guardrails/GuardrailPartitionTombstonesTest.java b/test/unit/org/apache/cassandra/db/guardrails/GuardrailPartitionTombstonesTest.java
new file mode 100644
index 000000000000..bb5618bc5cb2
--- /dev/null
+++ b/test/unit/org/apache/cassandra/db/guardrails/GuardrailPartitionTombstonesTest.java
@@ -0,0 +1,127 @@
+/*
+ * 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.guardrails;
+
+import org.junit.Test;
+
+/**
+ * Tests the guardrail for the number of tombstones in a partition, {@link Guardrails#partitionTombstones}.
+ * <p>
+ * The emission on unredacted log messages is tested in {@link org.apache.cassandra.distributed.test.guardrails.GuardrailPartitionTombstonesTest}.
+ */
+public class GuardrailPartitionTombstonesTest extends ThresholdTester
+{
+    private static final int WARN_THRESHOLD = 500; // high enough to exceed system tables, which aren't excluded
+    private static final int FAIL_THRESHOLD = WARN_THRESHOLD + 1;
+    private static final String REDACTED_MESSAGE = "Guardrail partition_tombstones violated: Partition <redacted>";
+
+    public GuardrailPartitionTombstonesTest()
+    {
+        super(WARN_THRESHOLD,
+              FAIL_THRESHOLD,
+              Guardrails.partitionTombstones,
+              Guardrails::setPartitionTombstonesThreshold,
+              Guardrails::getPartitionTombstonesWarnThreshold,
+              Guardrails::getPartitionTombstonesFailThreshold);
+    }
+
+    @Test
+    public void testPartitionTombstonesEnabled() throws Throwable
+    {
+        // Insert tombstones under both thresholds.
+        populateTable(WARN_THRESHOLD);
+
+        flush();
+        listener.assertNotWarned();
+        listener.assertNotFailed();
+        listener.clear();
+
+        compact();
+        listener.assertNotWarned();
+        listener.assertNotFailed();
+        listener.clear();
+
+        // Insert enough tombstones to trigger the warning guardrail, but not the failure one.
+        populateTable(WARN_THRESHOLD + 1);
+
+        flush();
+        listener.assertWarned(REDACTED_MESSAGE);
+        listener.clear();
+
+        compact();
+        listener.assertWarned(REDACTED_MESSAGE);
+        listener.clear();
+
+        // Insert enough tombstones to trigger the failure guardrail.
+        populateTable(FAIL_THRESHOLD + 1);
+
+        flush();
+        listener.assertFailed(REDACTED_MESSAGE);
+        listener.clear();
+
+        compact();
+        listener.assertFailed(REDACTED_MESSAGE);
+        listener.clear();
+
+        // remove most of the data to be under the threshold again
+        assertValid("DELETE FROM %s WHERE k = 1 AND c > 0");
+
+        flush();
+        listener.assertNotWarned();
+        listener.assertNotFailed();
+
+        compact();
+        listener.assertNotWarned();
+        listener.assertNotFailed();
+        listener.clear();
+    }
+
+    @Test
+    public void testPartitionTombstonesDisabled() throws Throwable
+    {
+        guardrails().setPartitionTombstonesThreshold(-1, -1);
+
+        populateTable(FAIL_THRESHOLD + 1);
+
+        flush();
+        listener.assertNotWarned();
+        listener.assertNotFailed();
+
+        compact();
+        listener.assertNotWarned();
+        listener.assertNotFailed();
+    }
+
+    private void populateTable(int numTombstones) throws Throwable
+    {
+        createTable("CREATE TABLE IF NOT EXISTS %s (k int, c int, v int, PRIMARY KEY(k, c))");
+        disableCompaction();
+
+        for (int c = 0; c < numTombstones; c++)
+        {
+            assertValid("DELETE FROM %s WHERE k = 1 AND c = " + c);
+        }
+
+        // insert some additional partitions to make sure the guardrail is per-partition
+        for (int k = 0; k <= FAIL_THRESHOLD; k++)
+        {
+            assertValid("DELETE FROM %s WHERE k = " + k + " AND c = 0");
+        }
+    }
+}