GEODE-10412: Clear expired tombstones during region destroy (#7838)

* GEODE-10412: Clear expired tombstones during region destroy

The issue:
During region destroy operation, the expired tombstones aren't cleared
when non-expired ones are available. Later, these expired
tombstones prevent all other regions' tombstones from being cleared
from memory, causing many issues (memory and disk exhaustion).

The solution:
When a region is destroyed, it must clear all the related expired and
non-expired tombstones from memory.

* Add distributed test that reproduce the issue

* Update after review

geode-core/src/distributedTest/java/org/apache/geode/internal/cache/versions/TombstoneDUnitTest.java

@@ -14,11 +14,13 @@
  */
 package org.apache.geode.internal.cache.versions;
 
+import static org.apache.geode.cache.RegionShortcut.PARTITION_PERSISTENT;
 import static org.apache.geode.cache.RegionShortcut.REPLICATE;
 import static org.apache.geode.cache.RegionShortcut.REPLICATE_PERSISTENT;
 import static org.apache.geode.internal.AvailablePortHelper.getRandomAvailableTCPPort;
 import static org.apache.geode.internal.cache.InitialImageOperation.GIITestHookType.DuringApplyDelta;
 import static org.apache.geode.internal.cache.InitialImageOperation.resetAllGIITestHooks;
+import static org.apache.geode.internal.cache.TombstoneService.EXPIRED_TOMBSTONE_LIMIT;
 import static org.apache.geode.test.awaitility.GeodeAwaitility.await;
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
@@ -121,6 +123,35 @@ public void testTombstoneGcMessagesBetweenPersistentAndNonPersistentRegion() {
     });
   }
 
+  @Test
+  public void testTombstoneExpiredAndNonExpiredAreClearedAfterRegionIsDestroyed() {
+    VM vm0 = VM.getVM(0);
+
+    vm0.invoke(() -> {
+      // reduce timeout so that tombstone is immediately marked as expired
+      TombstoneService.REPLICATE_TOMBSTONE_TIMEOUT = 100;
+      createCacheAndRegion(PARTITION_PERSISTENT);
+      region.put("K1", "V1");
+      region.destroy("K1");
+    });
+
+    vm0.invoke(() -> {
+      waitForScheduledTombstoneCount(0);
+      // increase timeout so that next tombstone doesn't expire
+      TombstoneService.REPLICATE_TOMBSTONE_TIMEOUT = 150000;
+      region.put("K1", "V1");
+      region.destroy("K1");
+
+      region.destroyRegion();
+      // force expiry of batch - there is only one expired tombstone at this moment
+      EXPIRED_TOMBSTONE_LIMIT = 1;
+    });
+
+    vm0.invoke(() -> {
+      createCacheAndRegion(PARTITION_PERSISTENT);
+      checkExpiredTombstones(0);
+    });
+  }
 
   @Test
   public void testWhenAnOutOfRangeTimeStampIsSeenWeExpireItInReplicateTombstoneSweeper() {
@@ -562,6 +593,17 @@ private void waitForTombstoneCount(int count) {
     }
   }
 
+  private void waitForScheduledTombstoneCount(int count) {
+    LocalRegion region = (LocalRegion) cache.getRegion(REGION_NAME);
+    await().until(() -> ((InternalCache) cache).getTombstoneService().getSweeper(region).tombstones
+        .size() == count);
+  }
+
+  private void checkExpiredTombstones(int count) {
+    await().until(
+        () -> ((InternalCache) cache).getTombstoneService().getScheduledTombstoneCount() == count);
+  }
+
   private void performGC(int count) throws Exception {
     ((InternalCache) cache).getTombstoneService().forceBatchExpirationForTests(count);
   }

geode-core/src/main/java/org/apache/geode/internal/cache/TombstoneService.java

@@ -926,7 +926,11 @@ private boolean removeUnexpiredIf(Predicate<Tombstone> predicate) {
      * @return true if predicate ever returned true
      */
     private boolean removeIf(Predicate<Tombstone> predicate) {
-      return removeUnexpiredIf(predicate) || removeExpiredIf(predicate);
+      boolean isTombstoneRemoved = removeUnexpiredIf(predicate);
+      if (removeExpiredIf(predicate)) {
+        isTombstoneRemoved = true;
+      }
+      return isTombstoneRemoved;
     }
 
     synchronized void start() {

geode-core/src/test/java/org/apache/geode/internal/cache/TombstoneServiceTest.java

@@ -14,6 +14,7 @@
  */
 package org.apache.geode.internal.cache;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
 import static org.mockito.Mockito.when;
@@ -40,7 +41,9 @@ public class TombstoneServiceTest {
   DistributedRegion region;
   VersionTag destroyedVersion;
   private TombstoneService.ReplicateTombstoneSweeper replicateTombstoneSweeper;
-  private TombstoneService.Tombstone tombstone;
+  private TombstoneService.Tombstone tombstone1;
+
+  private TombstoneService.Tombstone tombstone2;
 
 
   @Before
@@ -55,18 +58,19 @@ public void setUp() throws Exception {
     destroyedVersion = mock(VersionTag.class);
     replicateTombstoneSweeper = new TombstoneService.ReplicateTombstoneSweeper(cacheTime, stats,
         cancelCriterion, executor);
-    tombstone = new TombstoneService.Tombstone(entry, region, destroyedVersion);
-    tombstone.entry = entry;
+    tombstone1 = new TombstoneService.Tombstone(entry, region, destroyedVersion);
+    tombstone2 = new TombstoneService.Tombstone(entry, region, destroyedVersion);
+    tombstone1.entry = entry;
   }
 
   @Test
   public void validateThatRemoveIsNotCalledOnTombstoneInRegionThatIsNotInitialized() {
     when(region.isInitialized()).thenReturn(false);
     when(region.getRegionMap()).thenReturn(regionMap);
 
-    replicateTombstoneSweeper.expireTombstone(tombstone);
+    replicateTombstoneSweeper.expireTombstone(tombstone1);
     replicateTombstoneSweeper.expireBatch();
-    verify(regionMap, Mockito.never()).removeTombstone(tombstone.entry, tombstone);
+    verify(regionMap, Mockito.never()).removeTombstone(tombstone1.entry, tombstone1);
   }
 
   @Test
@@ -80,8 +84,36 @@ public void validateThatRemoveIsCalledOnTombstoneInRegionThatIsInitialized() {
     when(region.getDiskRegion()).thenReturn(mock(DiskRegion.class));
 
 
-    replicateTombstoneSweeper.expireTombstone(tombstone);
+    replicateTombstoneSweeper.expireTombstone(tombstone1);
     replicateTombstoneSweeper.expireBatch();
-    verify(regionMap).removeTombstone(tombstone.entry, tombstone);
+    verify(regionMap).removeTombstone(tombstone1.entry, tombstone1);
+  }
+
+  @Test
+  public void validateThatTheExpiredTombstonesAreCleared() {
+    when(region.getRegionMap()).thenReturn(regionMap);
+    replicateTombstoneSweeper.expireTombstone(tombstone1);
+    assertThat(replicateTombstoneSweeper.getScheduledTombstoneCount()).isOne();
+    replicateTombstoneSweeper.unscheduleTombstones(region);
+    assertThat(replicateTombstoneSweeper.getScheduledTombstoneCount()).isZero();
+  }
+
+  @Test
+  public void validateThatTheNonExpiredTombstonesAreCleared() {
+    when(region.getRegionMap()).thenReturn(regionMap);
+    replicateTombstoneSweeper.scheduleTombstone(tombstone1);
+    assertThat(replicateTombstoneSweeper.getScheduledTombstoneCount()).isOne();
+    replicateTombstoneSweeper.unscheduleTombstones(region);
+    assertThat(replicateTombstoneSweeper.getScheduledTombstoneCount()).isZero();
+  }
+
+  @Test
+  public void validateThatTheNonExpiredAndExpiredTombstonesAreCleared() {
+    when(region.getRegionMap()).thenReturn(regionMap);
+    replicateTombstoneSweeper.scheduleTombstone(tombstone1);
+    replicateTombstoneSweeper.expireTombstone(tombstone2);
+    assertThat(replicateTombstoneSweeper.getScheduledTombstoneCount()).isEqualTo(2);
+    replicateTombstoneSweeper.unscheduleTombstones(region);
+    assertThat(replicateTombstoneSweeper.getScheduledTombstoneCount()).isZero();
   }
 }