@Test
  public void testLotsOfTombstones() throws Exception {
    VM vm0 = getVM(0);

    // I think we just need to assert the number of tombstones, maybe?
    // Bruce has code that won't even let the tombstones expire for 10 minutes
    // That means on recovery we need to recover them all? Or do we need to recover
    // any? We're doing a GII. Won't we GII tombstones anyway? Ahh, but we need
    // to know that we don't need to record the new tombstones.

    InternalRegion region = createRegion(vm0);

    int initialCount = getTombstoneCount(region);
    assertThat(initialCount).isEqualTo(0);

    int entryCount = 20;
    for (int i = 0; i < entryCount; i++) {
      region.put(i, new byte[100]);
      region.destroy(i);
    }

    assertThat(getTombstoneCount(region)).isEqualTo(entryCount);

    // roll to a new oplog
    region.getDiskStore().forceRoll();

    // Force a compaction. This should do nothing, because the tombstones are not garbage, so only
    // 50% of the oplog is garbage (the creates).
    region.getDiskStore().forceCompaction();

    assertThat(region.getDiskStore().numCompactableOplogs()).isEqualTo(0);
    assertThat(getTombstoneCount(region)).isEqualTo(entryCount);

    getCache().close();

    region = createRegion(vm0);

    assertThat(getTombstoneCount(region)).isEqualTo(entryCount);

    TombstoneService tombstoneService = getCache().getTombstoneService();

    // Before expiring tombstones, no oplogs are available for compaction
    assertThat(region.getDiskStore().numCompactableOplogs()).isEqualTo(0);

    region.getDiskStore().forceCompaction();

    assertThat(
        tombstoneService.forceBatchExpirationForTests(entryCount / 2, TIMEOUT_MILLIS, MILLISECONDS))
            .isTrue();
    assertThat(getTombstoneCount(region)).isEqualTo(entryCount / 2);

    // After expiring, we should have an oplog available for compaction.
    assertThat(region.getDiskStore().numCompactableOplogs()).isEqualTo(1);

    // Test after restart the tombstones are still missing
    getCache().close();
    region = createRegion(vm0);
    assertThat(getTombstoneCount(region)).isEqualTo(entryCount / 2);

    // Should have an oplog available for compaction, because the tombstones were garbage collected
    assertThat(region.getDiskStore().numCompactableOplogs()).isEqualTo(1);

    // This should compact some oplogs
    region.getDiskStore().forceCompaction();
    assertThat(region.getDiskStore().numCompactableOplogs()).isEqualTo(0);

    // Restart again, and make sure the compaction didn't mess up our tombstone count
    getCache().close();
    region = createRegion(vm0);
    assertThat(getTombstoneCount(region)).isEqualTo(entryCount / 2);

    // Add a test hook that will shutdown the system as soon as we write a GC RVV record
    DiskStoreObserver.setInstance(new DiskStoreObserver() {

      @Override
      public void afterWriteGCRVV(DiskRegion dr) {
        // This will cause the disk store to shut down, preventing us from writing any other records
        throw new DiskAccessException();
      }
    });

    addIgnoredException(DiskAccessException.class);
    addIgnoredException(RegionDestroyedException.class);

    // Force expiration, with our test hook that should close the cache
    tombstoneService = getCache().getTombstoneService();
    assertThat(
        tombstoneService.forceBatchExpirationForTests(entryCount / 4, TIMEOUT_MILLIS, MILLISECONDS))
            .isTrue();

    getCache().close();

    // Restart again, and make sure the tombstones are in fact removed
    region = createRegion(vm0);
    assertThat(getTombstoneCount(region)).isEqualTo(entryCount / 4);
  }
Test that we correctly recover and expire recovered tombstones, with compaction enabled
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/persistence/PersistentRVVRecoveryDUnitTest.java