protected void versionTestClearOnNonReplicateWithConcurrentEvents() {
    // 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 String name = getUniqueName() + "-CC";

    Stream.of(vm0, vm1, vm2, vm3).forEach(vm -> {
      vm.invoke("Create Region", () -> {
        RegionFactory<?, ?> f;
        if (VM.getVMId() == 0) {
          f = getCache().createRegionFactory(
              getRegionAttributes(RegionShortcut.REPLICATE_PROXY.toString()));
        } else if (VM.getVMId() == 1) {
          f = getCache().createRegionFactory(getRegionAttributes());
          f.setDataPolicy(DataPolicy.NORMAL);
        } else {
          f = getCache().createRegionFactory(getRegionAttributes());
        }
        CCRegion = (LocalRegion) f.create(name);
      });
    });


    AsyncInvocation<Void> a0 = vm0.invokeAsync("perform concurrent ops", () -> {
      doOpsLoop(true);
      if (CCRegion.getScope().isDistributedNoAck()) {
        sendSerialMessageToAll(); // flush the ops
      }
    });

    AsyncInvocation<Void> a1 = vm1.invokeAsync("perform concurrent ops", () -> {

      doOpsLoop(true);
      if (CCRegion.getScope().isDistributedNoAck()) {
        sendSerialMessageToAll(); // flush the ops
      }
    });


    waitForAsyncProcessing(a0, "");
    waitForAsyncProcessing(a1, "");

    await().untilAsserted(() -> {
      Map r2Contents = vm2.invoke(MultiVMRegionTestCase::getCCRegionContents); // replicated
      Map r3Contents = vm3.invoke(MultiVMRegionTestCase::getCCRegionContents); // replicated

      assertThat(r2Contents.size()).isEqualTo(r3Contents.size());

      for (int i = 0; i < 10; i++) {
        String key = "cckey" + i;
        // because datapolicy=normal regions do not accept ops when an entry is not present
        // they may miss invalidates/creates/destroys that other VMs see while applying ops
        // that they originate that maybe should have been elided. For this reason we can't
        // guarantee their consistency and don't check for it here.
        assertThat(r3Contents.get(key)).describedAs("region contents are not consistent for " + key)
            .isEqualTo(
                r2Contents.get(key));
        for (int subi = 1; subi < 3; subi++) {
          String subkey = key + "-" + subi;
          if (r2Contents.containsKey(subkey)) {
            assertThat(r3Contents.get(subkey)).describedAs(
                "region contents are not consistent for " + subkey)
                .isEqualTo(r2Contents.get(subkey));
          }
        }
      }
    });
    // vm0 has no storage, but the others should have hit no timeouts waiting for
    // their region version vector to dominate the vector sent with the clear() operation.
    // If they did, then some ops were not properly recorded in the vector and something
    // is broken.
    vm1.invoke(MultiVMRegionTestCase::assertNoClearTimeouts);
    vm2.invoke(MultiVMRegionTestCase::assertNoClearTimeouts);
    vm3.invoke(MultiVMRegionTestCase::assertNoClearTimeouts);
  }
Clear all existing transaction listeners and install the given listener.
geode-dunit/src/main/java/org/apache/geode/cache30/MultiVMRegionTestCase.java