protected void versionTestTombstones() {
    disconnectAllFromDS();
    final int numEntries = 100;

    // create replicated regions in VM 0 and 1, then perform concurrent ops
    // on the same key while creating the region in VM2. Afterward make
    // sure that all three regions are consistent
    final long oldServerTimeout = TombstoneService.REPLICATE_TOMBSTONE_TIMEOUT;
    final long oldClientTimeout = TombstoneService.NON_REPLICATE_TOMBSTONE_TIMEOUT;
    final int oldExpiredTombstoneLimit = TombstoneService.EXPIRED_TOMBSTONE_LIMIT;
    final boolean oldIdleExpiration = TombstoneService.IDLE_EXPIRATION;
    final double oldLimit = TombstoneService.GC_MEMORY_THRESHOLD;
    final long oldMaxSleepTime = TombstoneService.MAX_SLEEP_TIME;
    try {
      SerializableRunnable setTimeout = new SerializableRunnable() {
        @Override
        public void run() {
          TombstoneService.REPLICATE_TOMBSTONE_TIMEOUT = 1000;
          TombstoneService.NON_REPLICATE_TOMBSTONE_TIMEOUT = 900;
          TombstoneService.EXPIRED_TOMBSTONE_LIMIT = numEntries;
          TombstoneService.IDLE_EXPIRATION = true;
          TombstoneService.GC_MEMORY_THRESHOLD = 0; // turn this off so heap profile won't cause
          // test to fail
          TombstoneService.MAX_SLEEP_TIME = 500;
        }
      };
      vm0.invoke(setTimeout);
      vm1.invoke(setTimeout);
      final String name = getUniqueName() + "-CC";
      SerializableRunnable createRegion = new SerializableRunnable() {
        @Override
        public void run() {
          RegionFactory f = getCache().createRegionFactory(getRegionAttributes());
          CCRegion = (LocalRegion) f.create(name);
          for (int i = 0; i < numEntries; i++) {
            CCRegion.put("cckey" + i, "ccvalue");
          }
          if (CCRegion.getScope().isDistributedNoAck()) {
            sendSerialMessageToAll(); // flush the ops
          }
        }
      };

      vm0.invoke("Create Region", createRegion);
      vm1.invoke("Create Region", createRegion);

      vm0.invoke("destroy entries and check tombstone count", () -> {
        try {
          for (int i = 0; i < numEntries; i++) {
            CCRegion.destroy("cckey" + i);
            boolean condition1 = !CCRegion.containsKey("cckey" + i);
            assertThat(condition1).describedAs("entry should not exist").isTrue();
            boolean condition = !CCRegion.containsValueForKey("cckey" + i);
            assertThat(condition).describedAs("entry should not contain a value").isTrue();
          }
          checkCCRegionTombstoneCount("after destroys in this vm ", numEntries);
          boolean condition = !CCRegion.containsValue(Token.TOMBSTONE);
          assertThat(condition).describedAs("region should not contain a tombstone").isTrue();
          if (CCRegion.getScope().isDistributedNoAck()) {
            sendSerialMessageToAll(); // flush the ops
          }
        } catch (CacheException e) {
          fail("while performing destroy operations", e);
        }
      });

      vm1.invoke("check tombstone count(2)", () -> {
        checkCCRegionTombstoneCount("after destroys in other vm ", numEntries);
        try {
          waitForAllTombstonesToExpire(numEntries);
        } catch (ConditionTimeoutException e) {
          CCRegion.dumpBackingMap();
          logger.info("tombstone service state: " + CCRegion.getCache().getTombstoneService());
          throw e;
        }
      });

      vm0.invoke("create/destroy entries and check tombstone count", () -> {
        final int origCount = CCRegion.getTombstoneCount();
        try {
          waitForAllTombstonesToExpire(origCount);
          logger.debug("creating tombstones.  current count={}", CCRegion.getTombstoneCount());
          for (int i = 0; i < numEntries; i++) {
            CCRegion.create("cckey" + i, i);
            CCRegion.destroy("cckey" + i);
          }
          logger.debug("done creating tombstones.  current count={}",
              CCRegion.getTombstoneCount());
          checkCCRegionTombstoneCount("after create+destroy in this vm ", numEntries);
          assertThat(CCRegion.size()).isEqualTo(0);
          afterCreates = 0;
          AttributesMutator<Object, Object> m = attributesMutatorFor(CCRegion);
          m.addCacheListener(new CacheListenerAdapter<Object, Object>() {
            @Override
            public void afterCreate(EntryEvent<Object, Object> event) {
              afterCreates++;
            }
          });
          if (CCRegion.getScope().isDistributedNoAck()) {
            sendSerialMessageToAll(); // flush the ops
          }
        } catch (AssertionError e) {
          CCRegion.dumpBackingMap();
          logger.info("tombstone service state: " + CCRegion.getCache().getTombstoneService());
          throw e;
        } catch (CacheException e) {
          fail("while performing create/destroy operations", e);
        }
      });

      vm1.invoke("check tombstone count and install listener", () -> {
        checkCCRegionTombstoneCount("after create+destroy in other vm ", numEntries);
        afterCreates = 0;
        AttributesMutator<Object, Object> m = attributesMutatorFor(CCRegion);
        m.addCacheListener(new CacheListenerAdapter<Object, Object>() {
          @Override
          public void afterCreate(EntryEvent<Object, Object> event) {
            afterCreates++;
          }
        });
      });

      // Now check to see if tombstones are resurrected by a create.
      // The entries should be created okay and the callback should be afterCreate.
      // The tombstone count won't go down until the entries are swept, but then
      // the count should fall to zero.

      vm0.invoke("create entries and check afterCreate and tombstone count", () -> {
        try {
          for (int i = 0; i < numEntries; i++) {
            CCRegion.create("cckey" + i, i);
          }
          checkCCRegionTombstoneCount("after create in this vm", 0);
          assertThat(afterCreates).describedAs("expected " + numEntries + " afterCreates")
              .isEqualTo(
                  numEntries);
          assertThat(CCRegion.size()).isEqualTo(numEntries);
          if (CCRegion.getScope().isDistributedNoAck()) {
            sendSerialMessageToAll(); // flush the ops
          }

          await()
              .until(
                  () -> CCRegion.getCache().getTombstoneService()
                      .getScheduledTombstoneCount() == 0);

        } catch (CacheException e) {
          fail("while performing create operations", e);
        } catch (ConditionTimeoutException timeout) {
          fail("Timed out waiting for all scheduled tombstones to be removed.  There are now "
              + CCRegion.getCache().getTombstoneService().getScheduledTombstoneCount()
              + " tombstones left out of " + numEntries + " initial tombstones. "
              + CCRegion.getCache().getTombstoneService(), timeout);
        }
      });

      vm1.invoke("check afterCreate and tombstone count", () -> {
        checkCCRegionTombstoneCount("after create in other vm", 0);
        assertThat(afterCreates).describedAs("expected " + numEntries + " afterCreates")
            .isEqualTo(
                numEntries);
        assertThat(CCRegion.size()).isEqualTo(numEntries);
        try {
          await()
              .until(() -> CCRegion.getCache().getTombstoneService()
                  .getScheduledTombstoneCount() == 0);
        } catch (ConditionTimeoutException timeout) {
          String message =
              "Timed out waiting for all scheduled tombstones to be removed.  There are now "
                  + CCRegion.getCache().getTombstoneService().getScheduledTombstoneCount()
                  + " tombstones left out of " + numEntries + " initial tombstones. "
                  + CCRegion.getCache().getTombstoneService();
          fail(message, timeout);
        }
      });
    } finally {
      SerializableRunnable resetTimeout = new SerializableRunnable() {
        @Override
        public void run() {
          TombstoneService.REPLICATE_TOMBSTONE_TIMEOUT = oldServerTimeout;
          TombstoneService.NON_REPLICATE_TOMBSTONE_TIMEOUT = oldClientTimeout;
          TombstoneService.EXPIRED_TOMBSTONE_LIMIT = oldExpiredTombstoneLimit;
          TombstoneService.IDLE_EXPIRATION = oldIdleExpiration;
          TombstoneService.GC_MEMORY_THRESHOLD = oldLimit;
          TombstoneService.MAX_SLEEP_TIME = oldMaxSleepTime;
        }
      };
      vm0.invoke(resetTimeout);
      vm1.invoke(resetTimeout);
    }
  }
Clear all existing transaction listeners and install the given listener.
geode-dunit/src/main/java/org/apache/geode/cache30/MultiVMRegionTestCase.java