protected void versionTestConcurrentEventsOnNonReplicatedRegion() {

    final boolean noAck = !getRegionAttributes().getScope().isAck();

    // create an empty region in vm0 and replicated regions in VM 1 and 3,
    // then perform concurrent ops
    // on the same key while creating the region in VM2. Afterward make
    // sure that all three regions are consistent

    final String name = getUniqueName() + "-CC";
    SerializableRunnable createRegion = new SerializableRunnable() {
      @Override
      public void run() {
        final RegionFactory<?, ?> f;
        if (VM.getVMId() == 0) {
          f = getCache()
              .createRegionFactory(getRegionAttributes(RegionShortcut.LOCAL.toString()));
          f.setScope(getRegionAttributes().getScope());
        } else {
          f = getCache().createRegionFactory(getRegionAttributes());
        }
        CCRegion = (LocalRegion) f.create(name);
      }
    };

    prepConcurrentEventsTest(name, createRegion);

    await().untilAsserted(() -> {
      // check consistency of the regions
      Map r0Contents = vm0.invoke(MultiVMRegionTestCase::getCCRegionContents);
      Map r1Contents = vm1.invoke(MultiVMRegionTestCase::getCCRegionContents);
      Map r2Contents = vm2.invoke(MultiVMRegionTestCase::getCCRegionContents);
      Map r3Contents = vm3.invoke(MultiVMRegionTestCase::getCCRegionContents);

      for (int i = 0; i < 10; i++) {
        String key = "cckey" + i;
        assertThat(r2Contents.get(key)).describedAs("region contents are not consistent").isEqualTo(
            r1Contents.get(key));
        assertThat(r3Contents.get(key)).describedAs("region contents are not consistent").isEqualTo(
            r2Contents.get(key));
        for (int subi = 1; subi < 3; subi++) {
          String subkey = key + "-" + subi;
          if (r1Contents.containsKey(subkey)) {
            assertThat(r2Contents.get(subkey)).describedAs("region contents are not consistent")
                .isEqualTo(
                    r1Contents.get(subkey));
            assertThat(r3Contents.get(subkey)).describedAs("region contents are not consistent")
                .isEqualTo(
                    r2Contents.get(subkey));
            if (r0Contents.containsKey(subkey)) {
              assertThat(r0Contents.get(subkey)).describedAs("region contents are not consistent")
                  .isEqualTo(
                      r1Contents.get(subkey));
            }
          } else {
            assertThat(r2Contents.containsKey(subkey)).isFalse();
            assertThat(r3Contents.containsKey(subkey)).isFalse();
            assertThat(r0Contents.containsKey(subkey)).isFalse();
          }
        }
      }
    });
    // a non-replicate region should not carry tombstones, so we'll check for that.
    // then we'll use a cache loader and make sure that 1-hop is used to put the
    // entry into the cache in a replicate. The new entry should have a version
    // of 1 when it's first created, the tombstone should have version 2 and
    // the loaded value should have version 3
    final String loadKey = "loadKey";
    vm0.invoke("add cache loader and create destroyed entry", () -> {
      AttributesMutator<Object, Object> attributesMutator = attributesMutatorFor(CCRegion);
      attributesMutator.setCacheLoader(new CacheLoader<Object, Object>() {
        @Override
        public void close() {}

        @Override
        public Object load(LoaderHelper<Object, Object> helper) throws CacheLoaderException {
          logger.info("The test CacheLoader has been invoked for key '" + helper.getKey() + "'");
          return "loadedValue";
        }
      });

      CCRegion.put(loadKey, "willbeLoadedInitialValue");
      CCRegion.destroy(loadKey);
      if (noAck) { // flush for validation to work
        sendSerialMessageToAll();
      }
      // this assertion guarantees that non-replicated regions do not create tombstones.
      // this is currently not the case but is an open issue
      // assertTrue(CCRegion.getRegionEntry(loadKey) == null)

    });
    vm1.invoke("confirm tombstone", () -> {
      assertThat(CCRegion.getRegionEntry(loadKey).getValueInVM(CCRegion)).isSameAs(
          Token.TOMBSTONE);
    });
    vm0.invoke("use cache loader", () -> {
      assertThat(CCRegion.get(loadKey)).isEqualTo("loadedValue");
      assertThat((CCRegion.getRegionEntry(loadKey)).getVersionStamp().getEntryVersion())
          .isEqualTo(
              (long) 3);
    });
    if (!noAck) { // might be 3 or 4 with no-ack
      vm1.invoke("verify version number", () -> {
        assertThat(CCRegion.get(loadKey)).isEqualTo("loadedValue");
        assertThat((CCRegion.getRegionEntry(loadKey)).getVersionStamp().getEntryVersion())
            .isEqualTo(
                (long) 3);
      });
    }
  }
This tests the concurrency versioning system to ensure that event conflation happens correctly
and that the statistic is being updated properly
geode-dunit/src/main/java/org/apache/geode/cache30/MultiVMRegionTestCase.java