diff --git a/db/db_table_properties_test.cc b/db/db_table_properties_test.cc
index 265e9cb2e..b12a56802 100644
--- a/db/db_table_properties_test.cc
+++ b/db/db_table_properties_test.cc
@@ -13,6 +13,7 @@
 #include "db/db_test_util.h"
 #include "port/stack_trace.h"
 #include "rocksdb/db.h"
+#include "rocksdb/utilities/table_properties_collectors.h"
 #include "util/testharness.h"
 #include "util/testutil.h"
 
@@ -250,6 +251,37 @@ TEST_F(DBTablePropertiesTest, GetColumnFamilyNameProperty) {
   }
 }
 
+TEST_F(DBTablePropertiesTest, DeletionTriggeredCompactionMarking) {
+  const int kNumKeys = 1000;
+  const int kWindowSize = 100;
+  const int kNumDelsTrigger = 90;
+
+  Options opts = CurrentOptions();
+  opts.table_properties_collector_factories.emplace_back(
+      NewCompactOnDeletionCollectorFactory(kWindowSize, kNumDelsTrigger));
+  Reopen(opts);
+
+  // add an L1 file to prevent tombstones from dropping due to obsolescence
+  // during flush
+  Put(Key(0), "val");
+  Flush();
+  MoveFilesToLevel(1);
+
+  for (int i = 0; i < kNumKeys; ++i) {
+    if (i >= kNumKeys - kWindowSize &&
+        i < kNumKeys - kWindowSize + kNumDelsTrigger) {
+      Delete(Key(i));
+    } else {
+      Put(Key(i), "val");
+    }
+  }
+  Flush();
+
+  dbfull()->TEST_WaitForCompact();
+  ASSERT_EQ(0, NumTableFilesAtLevel(0));
+  ASSERT_GT(NumTableFilesAtLevel(1), 0);
+}
+
 }  // namespace rocksdb
 
 #endif  // ROCKSDB_LITE
diff --git a/utilities/table_properties_collectors/compact_on_deletion_collector.cc b/utilities/table_properties_collectors/compact_on_deletion_collector.cc
index 304cdfff8..1d6f96952 100644
--- a/utilities/table_properties_collectors/compact_on_deletion_collector.cc
+++ b/utilities/table_properties_collectors/compact_on_deletion_collector.cc
@@ -12,26 +12,16 @@
 namespace rocksdb {
 
 CompactOnDeletionCollector::CompactOnDeletionCollector(
-    size_t sliding_window_size,
-    size_t deletion_trigger) {
-  deletion_trigger_ = deletion_trigger;
-
-  // First, compute the number of keys in each bucket.
-  bucket_size_ =
-      (sliding_window_size + kNumBuckets - 1) / kNumBuckets;
+    size_t sliding_window_size, size_t deletion_trigger)
+    : bucket_size_((sliding_window_size + kNumBuckets - 1) / kNumBuckets),
+      current_bucket_(0),
+      num_keys_in_current_bucket_(0),
+      num_deletions_in_observation_window_(0),
+      deletion_trigger_(deletion_trigger),
+      need_compaction_(false),
+      finished_(false) {
   assert(bucket_size_ > 0U);
-
-  Reset();
-}
-
-void CompactOnDeletionCollector::Reset() {
-  for (int i = 0; i < kNumBuckets; ++i) {
-    num_deletions_in_buckets_[i] = 0;
-  }
-  current_bucket_ = 0;
-  num_keys_in_current_bucket_ = 0;
-  num_deletions_in_observation_window_ = 0;
-  need_compaction_ = false;
+  memset(num_deletions_in_buckets_, 0, sizeof(size_t) * kNumBuckets);
 }
 
 // AddUserKey() will be called when a new key/value pair is inserted into the
@@ -43,6 +33,7 @@ Status CompactOnDeletionCollector::AddUserKey(
     const Slice& key, const Slice& value,
     EntryType type, SequenceNumber seq,
     uint64_t file_size) {
+  assert(!finished_);
   if (need_compaction_) {
     // If the output file already needs to be compacted, skip the check.
     return Status::OK();
diff --git a/utilities/table_properties_collectors/compact_on_deletion_collector.h b/utilities/table_properties_collectors/compact_on_deletion_collector.h
index bd240e517..9b8205108 100644
--- a/utilities/table_properties_collectors/compact_on_deletion_collector.h
+++ b/utilities/table_properties_collectors/compact_on_deletion_collector.h
@@ -61,7 +61,7 @@ class CompactOnDeletionCollector : public TablePropertiesCollector {
   // @params properties  User will add their collected statistics to
   // `properties`.
   virtual Status Finish(UserCollectedProperties* properties) override {
-    Reset();
+    finished_ = true;
     return Status::OK();
   }
 
@@ -98,6 +98,7 @@ class CompactOnDeletionCollector : public TablePropertiesCollector {
   size_t deletion_trigger_;
   // true if the current SST file needs to be compacted.
   bool need_compaction_;
+  bool finished_;
 };
 }  // namespace rocksdb
 #endif  // !ROCKSDB_LITE
diff --git a/utilities/table_properties_collectors/compact_on_deletion_collector_test.cc b/utilities/table_properties_collectors/compact_on_deletion_collector_test.cc
index 3c946bf41..cfa444139 100644
--- a/utilities/table_properties_collectors/compact_on_deletion_collector_test.cc
+++ b/utilities/table_properties_collectors/compact_on_deletion_collector_test.cc
@@ -58,12 +58,12 @@ int main(int argc, char** argv) {
     const int kBias = (kNumDeletionTrigger + kBucketSize - 1) / kBucketSize;
     // Simple test
     {
-      std::unique_ptr<rocksdb::TablePropertiesCollector> collector;
       auto factory = rocksdb::NewCompactOnDeletionCollectorFactory(
           kWindowSize, kNumDeletionTrigger);
-      collector.reset(factory->CreateTablePropertiesCollector(context));
       const int kSample = 10;
       for (int delete_rate = 0; delete_rate <= kSample; ++delete_rate) {
+        std::unique_ptr<rocksdb::TablePropertiesCollector> collector(
+            factory->CreateTablePropertiesCollector(context));
         int deletions = 0;
         for (int i = 0; i < kPaddedWindowSize; ++i) {
           if (i % kSample < delete_rate) {
@@ -90,12 +90,12 @@ int main(int argc, char** argv) {
 
     // Only one section of a file satisfies the compaction trigger
     {
-      std::unique_ptr<rocksdb::TablePropertiesCollector> collector;
       auto factory = rocksdb::NewCompactOnDeletionCollectorFactory(
           kWindowSize, kNumDeletionTrigger);
-      collector.reset(factory->CreateTablePropertiesCollector(context));
       const int kSample = 10;
       for (int delete_rate = 0; delete_rate <= kSample; ++delete_rate) {
+        std::unique_ptr<rocksdb::TablePropertiesCollector> collector(
+            factory->CreateTablePropertiesCollector(context));
         int deletions = 0;
         for (int section = 0; section < 5; ++section) {
           int initial_entries = rnd.Uniform(kWindowSize) + kWindowSize;