diff --git a/hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/QuotaTableUtil.java b/hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/QuotaTableUtil.java
index 2b5cb02b2ca..469e4318be2 100644
--- a/hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/QuotaTableUtil.java
+++ b/hbase-client/src/main/java/org/apache/hadoop/hbase/quotas/QuotaTableUtil.java
@@ -21,12 +21,16 @@ package org.apache.hadoop.hbase.quotas;
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collection;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
+import java.util.Set;
 import java.util.regex.Pattern;
+
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellScanner;
@@ -34,6 +38,7 @@ import org.apache.hadoop.hbase.CompareOperator;
 import org.apache.hadoop.hbase.NamespaceDescriptor;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.Delete;
 import org.apache.hadoop.hbase.client.Get;
 import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.Result;
@@ -53,6 +58,8 @@ import org.apache.yetus.audience.InterfaceStability;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import org.apache.hbase.thirdparty.com.google.common.collect.HashMultimap;
+import org.apache.hbase.thirdparty.com.google.common.collect.Multimap;
 import org.apache.hbase.thirdparty.com.google.protobuf.ByteString;
 import org.apache.hbase.thirdparty.com.google.protobuf.InvalidProtocolBufferException;
 import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;
@@ -540,6 +547,87 @@ public class QuotaTableUtil {
     return p;
   }
 
+  /**
+   * Returns a list of {@code Delete} to remove given table snapshot
+   * entries to remove from quota table
+   * @param snapshotEntriesToRemove the entries to remove
+   */
+  static List<Delete> createDeletesForExistingTableSnapshotSizes(
+      Multimap<TableName, String> snapshotEntriesToRemove) {
+    List<Delete> deletes = new ArrayList<>();
+    for (Map.Entry<TableName, Collection<String>> entry : snapshotEntriesToRemove.asMap()
+        .entrySet()) {
+      for (String snapshot : entry.getValue()) {
+        Delete d = new Delete(getTableRowKey(entry.getKey()));
+        d.addColumns(QUOTA_FAMILY_USAGE,
+            Bytes.add(QUOTA_SNAPSHOT_SIZE_QUALIFIER, Bytes.toBytes(snapshot)));
+        deletes.add(d);
+      }
+    }
+    return deletes;
+  }
+
+  /**
+   * Returns a list of {@code Delete} to remove all table snapshot entries from quota table.
+   * @param connection connection to re-use
+   */
+  static List<Delete> createDeletesForExistingTableSnapshotSizes(Connection connection)
+      throws IOException {
+    return createDeletesForExistingSnapshotsFromScan(connection, createScanForSpaceSnapshotSizes());
+  }
+
+  /**
+   * Returns a list of {@code Delete} to remove given namespace snapshot
+   * entries to removefrom quota table
+   * @param snapshotEntriesToRemove the entries to remove
+   */
+  static List<Delete> createDeletesForExistingNamespaceSnapshotSizes(
+      Set<String> snapshotEntriesToRemove) {
+    List<Delete> deletes = new ArrayList<>();
+    for (String snapshot : snapshotEntriesToRemove) {
+      Delete d = new Delete(getNamespaceRowKey(snapshot));
+      d.addColumns(QUOTA_FAMILY_USAGE, QUOTA_SNAPSHOT_SIZE_QUALIFIER);
+      deletes.add(d);
+    }
+    return deletes;
+  }
+
+  /**
+   * Returns a list of {@code Delete} to remove all namespace snapshot entries from quota table.
+   * @param connection connection to re-use
+   */
+  static List<Delete> createDeletesForExistingNamespaceSnapshotSizes(Connection connection)
+      throws IOException {
+    return createDeletesForExistingSnapshotsFromScan(connection,
+        createScanForNamespaceSnapshotSizes());
+  }
+
+  /**
+   * Returns a list of {@code Delete} to remove all entries returned by the passed scanner.
+   * @param connection connection to re-use
+   * @param scan the scanner to use to generate the list of deletes
+   */
+  static List<Delete> createDeletesForExistingSnapshotsFromScan(Connection connection, Scan scan)
+      throws IOException {
+    List<Delete> deletes = new ArrayList<>();
+    try (Table quotaTable = connection.getTable(QUOTA_TABLE_NAME);
+        ResultScanner rs = quotaTable.getScanner(scan)) {
+      for (Result r : rs) {
+        CellScanner cs = r.cellScanner();
+        while (cs.advance()) {
+          Cell c = cs.current();
+          byte[] family = Bytes.copy(c.getFamilyArray(), c.getFamilyOffset(), c.getFamilyLength());
+          byte[] qual =
+              Bytes.copy(c.getQualifierArray(), c.getQualifierOffset(), c.getQualifierLength());
+          Delete d = new Delete(r.getRow());
+          d.addColumns(family, qual);
+          deletes.add(d);
+        }
+      }
+      return deletes;
+    }
+  }
+
   /**
    * Fetches the computed size of all snapshots against tables in a namespace for space quotas.
    */
@@ -575,6 +663,34 @@ public class QuotaTableUtil {
     return QuotaProtos.SpaceQuotaSnapshot.parseFrom(bs).getQuotaUsage();
   }
 
+  /**
+   * Returns a scanner for all existing namespace snapshot entries.
+   */
+  static Scan createScanForNamespaceSnapshotSizes() {
+    return createScanForNamespaceSnapshotSizes(null);
+  }
+
+  /**
+   * Returns a scanner for all namespace snapshot entries of the given namespace
+   * @param namespace name of the namespace whose snapshot entries are to be scanned
+   */
+  static Scan createScanForNamespaceSnapshotSizes(String namespace) {
+    Scan s = new Scan();
+    if (namespace == null || namespace.isEmpty()) {
+      // Read all namespaces, just look at the row prefix
+      s.setRowPrefixFilter(QUOTA_NAMESPACE_ROW_KEY_PREFIX);
+    } else {
+      // Fetch the exact row for the table
+      byte[] rowkey = getNamespaceRowKey(namespace);
+      // Fetch just this one row
+      s.withStartRow(rowkey).withStopRow(rowkey, true);
+    }
+
+    // Just the usage family and only the snapshot size qualifiers
+    return s.addFamily(QUOTA_FAMILY_USAGE)
+        .setFilter(new ColumnPrefixFilter(QUOTA_SNAPSHOT_SIZE_QUALIFIER));
+  }
+
   static Scan createScanForSpaceSnapshotSizes() {
     return createScanForSpaceSnapshotSizes(null);
   }
@@ -621,6 +737,46 @@ public class QuotaTableUtil {
     }
   }
 
+  /**
+   * Returns a multimap for all existing table snapshot entries.
+   * @param conn connection to re-use
+   */
+  public static Multimap<TableName, String> getTableSnapshots(Connection conn) throws IOException {
+    try (Table quotaTable = conn.getTable(QUOTA_TABLE_NAME);
+        ResultScanner rs = quotaTable.getScanner(createScanForSpaceSnapshotSizes())) {
+      Multimap<TableName, String> snapshots = HashMultimap.create();
+      for (Result r : rs) {
+        CellScanner cs = r.cellScanner();
+        while (cs.advance()) {
+          Cell c = cs.current();
+
+          final String snapshot = extractSnapshotNameFromSizeCell(c);
+          snapshots.put(getTableFromRowKey(r.getRow()), snapshot);
+        }
+      }
+      return snapshots;
+    }
+  }
+
+  /**
+   * Returns a set of the names of all namespaces containing snapshot entries.
+   * @param conn connection to re-use
+   */
+  public static Set<String> getNamespaceSnapshots(Connection conn) throws IOException {
+    try (Table quotaTable = conn.getTable(QUOTA_TABLE_NAME);
+        ResultScanner rs = quotaTable.getScanner(createScanForNamespaceSnapshotSizes())) {
+      Set<String> snapshots = new HashSet<>();
+      for (Result r : rs) {
+        CellScanner cs = r.cellScanner();
+        while (cs.advance()) {
+          Cell c = cs.current();
+          snapshots.add(getNamespaceFromRowKey(r.getRow()));
+        }
+      }
+      return snapshots;
+    }
+  }
+
   /**
    * Returns the current space quota snapshot of the given {@code tableName} from
    * {@code QuotaTableUtil.QUOTA_TABLE_NAME} or null if the no quota information is available for
diff --git a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/SnapshotQuotaObserverChore.java b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/SnapshotQuotaObserverChore.java
index f74bae0c512..9111b8d1c42 100644
--- a/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/SnapshotQuotaObserverChore.java
+++ b/hbase-server/src/main/java/org/apache/hadoop/hbase/quotas/SnapshotQuotaObserverChore.java
@@ -21,6 +21,7 @@ import java.util.Arrays;
 import java.util.Collection;
 import java.util.HashMap;
 import java.util.HashSet;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
@@ -37,6 +38,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.hbase.client.Admin;
 import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.client.Delete;
 import org.apache.hadoop.hbase.client.Table;
 import org.apache.hadoop.hbase.master.HMaster;
 import org.apache.hadoop.hbase.master.MetricsMaster;
@@ -110,6 +112,12 @@ public class SnapshotQuotaObserverChore extends ScheduledChore {
       metrics.incrementSnapshotFetchTime((System.nanoTime() - start) / 1_000_000);
     }
 
+    // Remove old table snapshots data
+    pruneTableSnapshots(snapshotsToComputeSize);
+
+    // Remove old namespace snapshots data
+    pruneNamespaceSnapshots(snapshotsToComputeSize);
+
     // For each table, compute the size of each snapshot
     Map<String,Long> namespaceSnapshotSizes = computeSnapshotSizes(snapshotsToComputeSize);
 
@@ -118,6 +126,43 @@ public class SnapshotQuotaObserverChore extends ScheduledChore {
     persistSnapshotSizesForNamespaces(namespaceSnapshotSizes);
   }
 
+  /**
+   * Removes the snapshot entries that are present in Quota table but not in snapshotsToComputeSize
+   *
+   * @param snapshotsToComputeSize list of snapshots to be persisted
+   */
+  void pruneTableSnapshots(Multimap<TableName, String> snapshotsToComputeSize) throws IOException {
+    Multimap<TableName, String> existingSnapshotEntries = QuotaTableUtil.getTableSnapshots(conn);
+    Multimap<TableName, String> snapshotEntriesToRemove = HashMultimap.create();
+    for (Entry<TableName, Collection<String>> entry : existingSnapshotEntries.asMap().entrySet()) {
+      TableName tn = entry.getKey();
+      Set<String> setOfSnapshots = new HashSet<>(entry.getValue());
+      for (String snapshot : snapshotsToComputeSize.get(tn)) {
+        setOfSnapshots.remove(snapshot);
+      }
+
+      for (String snapshot : setOfSnapshots) {
+        snapshotEntriesToRemove.put(tn, snapshot);
+      }
+    }
+    removeExistingTableSnapshotSizes(snapshotEntriesToRemove);
+  }
+
+  /**
+   * Removes the snapshot entries that are present in Quota table but not in snapshotsToComputeSize
+   *
+   * @param snapshotsToComputeSize list of snapshots to be persisted
+   */
+  void pruneNamespaceSnapshots(Multimap<TableName, String> snapshotsToComputeSize)
+      throws IOException {
+    Set<String> existingSnapshotEntries = QuotaTableUtil.getNamespaceSnapshots(conn);
+    for (TableName tableName : snapshotsToComputeSize.keySet()) {
+      existingSnapshotEntries.remove(tableName.getNamespaceAsString());
+    }
+    // here existingSnapshotEntries is left with the entries to be removed
+    removeExistingNamespaceSnapshotSizes(existingSnapshotEntries);
+  }
+
   /**
    * Fetches each table with a quota (table or namespace quota), and then fetch the name of each
    * snapshot which was created from that table.
@@ -220,6 +265,24 @@ public class SnapshotQuotaObserverChore extends ScheduledChore {
     }
   }
 
+  void removeExistingTableSnapshotSizes(Multimap<TableName, String> snapshotEntriesToRemove)
+      throws IOException {
+    removeExistingSnapshotSizes(
+        QuotaTableUtil.createDeletesForExistingTableSnapshotSizes(snapshotEntriesToRemove));
+  }
+
+  void removeExistingNamespaceSnapshotSizes(Set<String> snapshotEntriesToRemove)
+      throws IOException {
+    removeExistingSnapshotSizes(
+        QuotaTableUtil.createDeletesForExistingNamespaceSnapshotSizes(snapshotEntriesToRemove));
+  }
+
+  void removeExistingSnapshotSizes(List<Delete> deletes) throws IOException {
+    try (Table quotaTable = conn.getTable(QuotaUtil.QUOTA_TABLE_NAME)) {
+      quotaTable.delete(deletes);
+    }
+  }
+
   /**
    * Extracts the period for the chore from the configuration.
    *
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaTableUtil.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaTableUtil.java
index 6aac054ba8f..2fa22f55afb 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaTableUtil.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestQuotaTableUtil.java
@@ -24,9 +24,12 @@ import static org.junit.Assert.assertTrue;
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
+
 import org.apache.hadoop.hbase.Cell;
 import org.apache.hadoop.hbase.CellScanner;
 import org.apache.hadoop.hbase.HBaseClassTestRule;
@@ -36,6 +39,7 @@ import org.apache.hadoop.hbase.NamespaceDescriptor;
 import org.apache.hadoop.hbase.TableName;
 import org.apache.hadoop.hbase.client.Connection;
 import org.apache.hadoop.hbase.client.ConnectionFactory;
+import org.apache.hadoop.hbase.client.Delete;
 import org.apache.hadoop.hbase.client.Put;
 import org.apache.hadoop.hbase.client.Result;
 import org.apache.hadoop.hbase.client.ResultScanner;
@@ -53,6 +57,8 @@ import org.junit.Test;
 import org.junit.experimental.categories.Category;
 import org.junit.rules.TestName;
 
+import org.apache.hbase.thirdparty.com.google.common.collect.HashMultimap;
+import org.apache.hbase.thirdparty.com.google.common.collect.Multimap;
 import org.apache.hbase.thirdparty.com.google.protobuf.UnsafeByteOperations;
 
 import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
@@ -109,6 +115,68 @@ public class TestQuotaTableUtil {
     this.connection.close();
   }
 
+  @Test
+  public void testDeleteSnapshots() throws Exception {
+    TableName tn = TableName.valueOf(name.getMethodName());
+    try (Table t = connection.getTable(QuotaTableUtil.QUOTA_TABLE_NAME)) {
+      Quotas quota = Quotas.newBuilder().setSpace(
+          QuotaProtos.SpaceQuota.newBuilder().setSoftLimit(7L)
+              .setViolationPolicy(QuotaProtos.SpaceViolationPolicy.NO_WRITES).build()).build();
+      QuotaUtil.addTableQuota(connection, tn, quota);
+
+      String snapshotName = name.getMethodName() + "_snapshot";
+      t.put(QuotaTableUtil.createPutForSnapshotSize(tn, snapshotName, 3L));
+      t.put(QuotaTableUtil.createPutForSnapshotSize(tn, snapshotName, 5L));
+      assertEquals(1, QuotaTableUtil.getObservedSnapshotSizes(connection).size());
+
+      List<Delete> deletes = QuotaTableUtil.createDeletesForExistingTableSnapshotSizes(connection);
+      assertEquals(1, deletes.size());
+
+      t.delete(deletes);
+      assertEquals(0, QuotaTableUtil.getObservedSnapshotSizes(connection).size());
+
+      String ns = name.getMethodName();
+      t.put(QuotaTableUtil.createPutForNamespaceSnapshotSize(ns, 5L));
+      t.put(QuotaTableUtil.createPutForNamespaceSnapshotSize(ns, 3L));
+      assertEquals(3L, QuotaTableUtil.getNamespaceSnapshotSize(connection, ns));
+
+      deletes = QuotaTableUtil.createDeletesForExistingNamespaceSnapshotSizes(connection);
+      assertEquals(1, deletes.size());
+
+      t.delete(deletes);
+      assertEquals(0L, QuotaTableUtil.getNamespaceSnapshotSize(connection, ns));
+
+      t.put(QuotaTableUtil.createPutForSnapshotSize(TableName.valueOf("t1"), "s1", 3L));
+      t.put(QuotaTableUtil.createPutForSnapshotSize(TableName.valueOf("t2"), "s2", 3L));
+      t.put(QuotaTableUtil.createPutForSnapshotSize(TableName.valueOf("t3"), "s3", 3L));
+      t.put(QuotaTableUtil.createPutForSnapshotSize(TableName.valueOf("t4"), "s4", 3L));
+      t.put(QuotaTableUtil.createPutForSnapshotSize(TableName.valueOf("t1"), "s5", 3L));
+
+      t.put(QuotaTableUtil.createPutForNamespaceSnapshotSize("ns1", 3L));
+      t.put(QuotaTableUtil.createPutForNamespaceSnapshotSize("ns2", 3L));
+      t.put(QuotaTableUtil.createPutForNamespaceSnapshotSize("ns3", 3L));
+
+      assertEquals(5,QuotaTableUtil.getTableSnapshots(connection).size());
+      assertEquals(3,QuotaTableUtil.getNamespaceSnapshots(connection).size());
+
+      Multimap<TableName, String> tableSnapshotEntriesToRemove = HashMultimap.create();
+      tableSnapshotEntriesToRemove.put(TableName.valueOf("t1"), "s1");
+      tableSnapshotEntriesToRemove.put(TableName.valueOf("t3"), "s3");
+      tableSnapshotEntriesToRemove.put(TableName.valueOf("t4"), "s4");
+
+      Set<String> namespaceSnapshotEntriesToRemove = new HashSet<>();
+      namespaceSnapshotEntriesToRemove.add("ns2");
+      namespaceSnapshotEntriesToRemove.add("ns1");
+
+      deletes =
+          QuotaTableUtil.createDeletesForExistingTableSnapshotSizes(tableSnapshotEntriesToRemove);
+      assertEquals(3, deletes.size());
+      deletes = QuotaTableUtil
+          .createDeletesForExistingNamespaceSnapshotSizes(namespaceSnapshotEntriesToRemove);
+      assertEquals(2, deletes.size());
+    }
+  }
+
   @Test
   public void testTableQuotaUtil() throws Exception {
     final TableName tableName = TableName.valueOf(name.getMethodName());
diff --git a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestSnapshotQuotaObserverChore.java b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestSnapshotQuotaObserverChore.java
index bddfe261a1e..b9bd8a38dac 100644
--- a/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestSnapshotQuotaObserverChore.java
+++ b/hbase-server/src/test/java/org/apache/hadoop/hbase/quotas/TestSnapshotQuotaObserverChore.java
@@ -323,6 +323,86 @@ public class TestSnapshotQuotaObserverChore {
     }
   }
 
+  @Test
+  public void testRemovedSnapshots() throws Exception {
+    // Create a table and set a quota
+    TableName tn1 = helper.createTableWithRegions(1);
+    admin.setQuota(QuotaSettingsFactory.limitTableSpace(tn1, SpaceQuotaHelperForTests.ONE_GIGABYTE,
+        SpaceViolationPolicy.NO_INSERTS));
+
+    // Write some data and flush it
+    helper.writeData(tn1, 256L * SpaceQuotaHelperForTests.ONE_KILOBYTE); // 256 KB
+
+    final AtomicReference<Long> lastSeenSize = new AtomicReference<>();
+    // Wait for the Master chore to run to see the usage (with a fudge factor)
+    TEST_UTIL.waitFor(30_000, new SpaceQuotaSnapshotPredicate(conn, tn1) {
+      @Override boolean evaluate(SpaceQuotaSnapshot snapshot) throws Exception {
+        lastSeenSize.set(snapshot.getUsage());
+        return snapshot.getUsage() > 230L * SpaceQuotaHelperForTests.ONE_KILOBYTE;
+      }
+    });
+
+    // Create a snapshot on the table
+    final String snapshotName1 = tn1 + "snapshot1";
+    admin.snapshot(new SnapshotDescription(snapshotName1, tn1, SnapshotType.SKIPFLUSH));
+
+    // Snapshot size has to be 0 as the snapshot shares the data with the table
+    final Table quotaTable = conn.getTable(QuotaUtil.QUOTA_TABLE_NAME);
+    TEST_UTIL.waitFor(30_000, new Predicate<Exception>() {
+      @Override public boolean evaluate() throws Exception {
+        Get g = QuotaTableUtil.makeGetForSnapshotSize(tn1, snapshotName1);
+        Result r = quotaTable.get(g);
+        if (r == null || r.isEmpty()) {
+          return false;
+        }
+        r.advance();
+        Cell c = r.current();
+        return QuotaTableUtil.parseSnapshotSize(c) == 0;
+      }
+    });
+    // Total usage has to remain same as what we saw before taking a snapshot
+    TEST_UTIL.waitFor(30_000, new SpaceQuotaSnapshotPredicate(conn, tn1) {
+      @Override boolean evaluate(SpaceQuotaSnapshot snapshot) throws Exception {
+        return snapshot.getUsage() == lastSeenSize.get();
+      }
+    });
+
+    // Major compact the table to force a rewrite
+    TEST_UTIL.compact(tn1, true);
+    // Now the snapshot size has to prev total size
+    TEST_UTIL.waitFor(30_000, new Predicate<Exception>() {
+      @Override public boolean evaluate() throws Exception {
+        Get g = QuotaTableUtil.makeGetForSnapshotSize(tn1, snapshotName1);
+        Result r = quotaTable.get(g);
+        if (r == null || r.isEmpty()) {
+          return false;
+        }
+        r.advance();
+        Cell c = r.current();
+        // The compaction result file has an additional compaction event tracker
+        return lastSeenSize.get() == QuotaTableUtil.parseSnapshotSize(c);
+      }
+    });
+    // The total size now has to be equal/more than double of prev total size
+    // as double the number of store files exist now.
+    final AtomicReference<Long> sizeAfterCompaction = new AtomicReference<>();
+    TEST_UTIL.waitFor(30_000, new SpaceQuotaSnapshotPredicate(conn, tn1) {
+      @Override boolean evaluate(SpaceQuotaSnapshot snapshot) throws Exception {
+        sizeAfterCompaction.set(snapshot.getUsage());
+        return snapshot.getUsage() >= 2 * lastSeenSize.get();
+      }
+    });
+
+    // Delete the snapshot
+    admin.deleteSnapshot(snapshotName1);
+    // Total size has to come down to prev totalsize - snapshot size(which was removed)
+    TEST_UTIL.waitFor(30_000, new SpaceQuotaSnapshotPredicate(conn, tn1) {
+      @Override boolean evaluate(SpaceQuotaSnapshot snapshot) throws Exception {
+        return snapshot.getUsage() == (sizeAfterCompaction.get() - lastSeenSize.get());
+      }
+    });
+  }
+
   @Test
   public void testBucketingFilesToSnapshots() throws Exception {
     // Create a table and set a quota