@Test
  @Parameters({"true", "false"})
  @TestCaseName("{method}({params})")
  public void testPersistConflictOperations(boolean diskSynchronous) throws Exception {
    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(this::addSleepBeforeSendAbstractUpdateMessage);
    }

    AsyncInvocation<Void> createPersistentRegionInVM0 = vm0.invokeAsync(() -> {
      createReplicateRegion(regionName, getDiskDirs(getVMId()), diskSynchronous);
    });
    AsyncInvocation<Void> createPersistentRegionInVM1 = vm1.invokeAsync(() -> {
      createReplicateRegion(regionName, getDiskDirs(getVMId()), diskSynchronous);
    });

    createPersistentRegionInVM0.await();
    createPersistentRegionInVM1.await();

    AsyncInvocation<Void> putInVM0 = vm0.invokeAsync(() -> {
      getCache().<String, String>getRegion(regionName).put("A", "vm0");
    });
    AsyncInvocation<Void> putInVM1 = vm1.invokeAsync(() -> {
      getCache().<String, String>getRegion(regionName).put("A", "vm1");
    });

    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> SLEEP.get().countDown());
    }

    putInVM0.await();
    putInVM1.await();

    RegionVersionVector rvvInVM0 = toRVV(vm0.invoke(this::getRVVBytes));
    RegionVersionVector rvvInVM1 = toRVV(vm1.invoke(this::getRVVBytes));
    assertSameRVV(rvvInVM1, rvvInVM0);

    Object valueInVM0 = vm0.invoke(() -> getCache().getRegion(regionName).get("A"));
    Object valueInVM1 = vm1.invoke(() -> getCache().getRegion(regionName).get("A"));
    assertThat(valueInVM1).isEqualTo(valueInVM0);

    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> getCache().getRegion(regionName).close());
    }

    // recover
    createPersistentRegionInVM1 = vm1.invokeAsync(() -> {
      createReplicateRegion(regionName, getDiskDirs(getVMId()), diskSynchronous);
    });
    createPersistentRegionInVM0 = vm0.invokeAsync(() -> {
      createReplicateRegion(regionName, getDiskDirs(getVMId()), diskSynchronous);
    });

    createPersistentRegionInVM1.await();
    createPersistentRegionInVM0.await();

    valueInVM0 = vm0.invoke(() -> getCache().getRegion(regionName).get("A"));
    valueInVM1 = vm1.invoke(() -> getCache().getRegion(regionName).get("A"));
    assertThat(valueInVM1).isEqualTo(valueInVM0);

    rvvInVM0 = toRVV(vm0.invoke(this::getRVVBytes));
    rvvInVM1 = toRVV(vm1.invoke(this::getRVVBytes));
    assertSameRVV(rvvInVM1, rvvInVM0);

    // round 2: async disk write
    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(this::addSleepBeforeSendAbstractUpdateMessage);
    }

    putInVM0 = vm0.invokeAsync(() -> {
      Region<String, String> region = getCache().getRegion(regionName);
      for (int i = 0; i < 1000; i++) {
        region.put("A", "vm0-" + i);
      }
    });
    putInVM1 = vm1.invokeAsync(() -> {
      Region<String, String> region = getCache().getRegion(regionName);
      for (int i = 0; i < 1000; i++) {
        region.put("A", "vm1-" + i);
      }
    });

    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> SLEEP.get().countDown());
    }

    putInVM0.await();
    putInVM1.await();

    rvvInVM0 = toRVV(vm0.invoke(this::getRVVBytes));
    rvvInVM1 = toRVV(vm1.invoke(this::getRVVBytes));
    assertSameRVV(rvvInVM1, rvvInVM0);

    valueInVM0 = vm0.invoke(() -> getCache().getRegion(regionName).get("A"));
    valueInVM1 = vm1.invoke(() -> getCache().getRegion(regionName).get("A"));
    assertThat(valueInVM1).isEqualTo(valueInVM0);

    for (VM vm : toArray(vm0, vm1)) {
      vm.invoke(() -> getCache().close());
    }

    // recover again
    createPersistentRegionInVM1 = vm1.invokeAsync(() -> {
      createReplicateRegion(regionName, getDiskDirs(getVMId()), diskSynchronous);
    });
    createPersistentRegionInVM0 = vm0.invokeAsync(() -> {
      createReplicateRegion(regionName, getDiskDirs(getVMId()), diskSynchronous);
    });

    createPersistentRegionInVM1.await();
    createPersistentRegionInVM0.await();

    valueInVM0 = vm0.invoke(() -> getCache().getRegion(regionName).get("A"));
    valueInVM1 = vm1.invoke(() -> getCache().getRegion(regionName).get("A"));
    assertThat(valueInVM1).isEqualTo(valueInVM0);

    rvvInVM0 = toRVV(vm0.invoke(this::getRVVBytes));
    rvvInVM1 = toRVV(vm1.invoke(this::getRVVBytes));
    assertSameRVV(rvvInVM1, rvvInVM0);
  }
vm0 and vm1 are peers, each holds a DR. They do put to the same key for different value at the
same time. Use DistributionMessageObserver.beforeSendMessage to hold on the distribution
message. One of the member will persist the conflict version tag, while another member will
persist both of the 2 operations. Overall, their RVV should match after the operations.
geode-core/src/distributedTest/java/org/apache/geode/internal/cache/persistence/PersistentRecoveryOrderDUnitTest.java