private void doRetriedTest(final OpType opType) {
    Host host = Host.getHost(0);
    final VM vm0 = host.getVM(0);
    final VM vm1 = host.getVM(1);
    final VM vm3 = host.getVM(3);

    int port0 = createServerRegion(vm0, RegionShortcut.REPLICATE);
    int port1 = createServerRegion(vm1, RegionShortcut.REPLICATE);
    createClientRegion(vm3, port0, port1);

    vm0.invoke(new SerializableRunnable() {

      @Override
      public void run() {
        Region region = getCache().getRegion("region");
        if (opType != OpType.CREATE && opType != OpType.PUT_IF_ABSENT) {
          region.put(0, "value");
        }

        // Add a listener to close vm0 when we send a distributed operation
        // this will cause a retry after we have applied the original operation to
        // the cache, causing a retry
        DistributionMessageObserver
            .setInstance(new DistributionMessageObserver() {

              @Override
              public void beforeSendMessage(ClusterDistributionManager dm,
                  DistributionMessage message) {
                if (message instanceof DistributedCacheOperation.CacheOperationMessage) {
                  DistributedCacheOperation.CacheOperationMessage com =
                      (DistributedCacheOperation.CacheOperationMessage) message;
                  VersionTag tag = com.getVersionTag();
                  if (((opType == OpType.CREATE || opType == OpType.PUT_IF_ABSENT)
                      && tag.getEntryVersion() == 1) || tag.getEntryVersion() == 2) {
                    DistributionMessageObserver.setInstance(null);
                    disconnectFromDS(vm0);
                  }
                }
              }
            });

      }
    });

    // this operation will trigger vm0 to be closed, and the operation will be retried
    vm3.invoke(new SerializableCallable("perform update in client") {
      @Override
      public Object call() throws Exception {
        Region region = getCache().getRegion("region");
        switch (opType) {
          case CREATE:
            region.create(0, "newvalue");
            break;
          case PUT:
            region.put(0, "newvalue");
            break;
          case DESTROY:
            region.destroy(0);
            break;
          case INVALIDATE:
            region.invalidate(0);
            break;
          case PUT_IF_ABSENT:
            region.putIfAbsent(0, "newvalue");
            break;
          case REPLACE:
            region.replace(0, "newvalue");
            break;
          case REPLACE_WITH_OLDVALUE:
            region.replace(0, "value", "newvalue");
            break;
          case REMOVE:
            region.remove(0, "value");
            break;
        }
        return null;
      }
    });

    // Verify the observer was triggered
    vm1.invoke(new SerializableRunnable() {

      @Override
      public void run() {
        // if the observer was triggered, it would have cleared itself
        assertNull(DistributionMessageObserver.getInstance());
        VersionTag tag = ((LocalRegion) getCache().getRegion("region"))
            .getVersionTag(0);
        if (opType == OpType.CREATE || opType == OpType.PUT_IF_ABSENT) {
          assertEquals(1, tag.getRegionVersion());
        } else {
          assertEquals(2, tag.getRegionVersion());
        }
      }
    });

    // Make sure vm0 did in fact shut down
    vm0.invoke(new SerializableRunnable() {
      @Override
      public void run() {
        GemFireCacheImpl cache = (GemFireCacheImpl) basicGetCache();
        assertTrue(cache == null || cache.isClosed());
      }
    });
  }
Test that we can successfully retry a distributed put all and get the version information.
second failure in bug 44951
geode-core/src/distributedTest/java/org/apache/geode/pdx/ClientsWithVersioningRetryDUnitTest.java