@Test
  public void testRetryPutAllInAccessor() {
    Host host = Host.getHost(0);
    final VM vm0 = host.getVM(0);
    final VM vm1 = host.getVM(1);
    final VM vm2 = host.getVM(2);
    final VM vm3 = host.getVM(3);

    LogWriterUtils.getLogWriter().info("creating region in vm0");
    createRegionInPeer(vm0, RegionShortcut.PARTITION_REDUNDANT_PERSISTENT);

    vm0.invoke(new SerializableRunnable() {
      @Override
      public void run() {
        // Make sure the bucket 0 is primary in this member.
        Region region = getCache().getRegion("region");
        region.put(0, "value");
      }
    });

    LogWriterUtils.getLogWriter().info("creating region in vm1");
    createRegionInPeer(vm1, RegionShortcut.PARTITION_REDUNDANT_PERSISTENT);
    LogWriterUtils.getLogWriter().info("creating region in vm2");
    createRegionInPeer(vm2, RegionShortcut.PARTITION_REDUNDANT_PERSISTENT);
    LogWriterUtils.getLogWriter().info("creating region in vm3");
    createRegionInPeer(vm3, RegionShortcut.PARTITION_PROXY);

    expectedExceptions
        .add(IgnoredException.addIgnoredException("RuntimeException", vm2));
    vm2.invoke(
        new SerializableRunnable("install message listener to ignore update") {
          @Override
          public void run() {
            // Add a listener to close vm2 when we send a distributed put all operation
            // this will cause a retry after we have applied the original put all to
            // the cache, causing a retry
            DistributionMessageObserver
                .setInstance(new DistributionMessageObserver() {

                  @Override
                  public void beforeProcessMessage(ClusterDistributionManager dm,
                      DistributionMessage msg) {
                    if (msg instanceof DistributedPutAllOperation.PutAllMessage) {
                      DistributionMessageObserver.setInstance(null);
                      Wait.pause(
                          5000); // give vm1 time to process the message that we're ignoring
                      disconnectFromDS(vm0);
                      // no reply will be sent to vm0 due to this exception, but that's okay
                      // because vm0 has been shut down
                      throw new RuntimeException(
                          "test code is ignoring message: " + msg);
                    }
                  }
                });

          }
        });

    // This will be a put all to bucket 0
    // Here's the expected sequence
    // accessor->vm0 (primary)
    // vm0->vm1, vm2
    // vm2 will ignore the message & kill vm0
    // accessor->vm2 or vm1
    // version tag is recovered and put in the event & cache
    vm3.invoke(new SerializableCallable("perform putAll in accessor") {
      @Override
      public Object call() throws Exception {
        Region region = getCache().getRegion("region");
        Map map = new HashMap();
        map.put(0, "a");
        map.put(113, "b");
        region.putAll(map);
        return null;
      }
    });

    // verify that the version is correct
    vm1.invoke(new SerializableRunnable("verify vm1") {

      @Override
      public void run() {
        // if the observer was triggered, it would have cleared itself
        assertNull(DistributionMessageObserver.getInstance());

        Region region = getCache().getRegion("region");
        VersionTag tag = ((LocalRegion) region).getVersionTag(0);
        assertEquals(2, tag.getEntryVersion());
      }
    });

    // Verify the observer was triggered and the version is correct
    vm2.invoke(new SerializableRunnable("verify vm2") {

      @Override
      public void run() {
        // if the observer was triggered, it would have cleared itself
        assertNull(DistributionMessageObserver.getInstance());

        Region region = getCache().getRegion("region");
        VersionTag tag = ((LocalRegion) region).getVersionTag(0);
        assertEquals(2, tag.getEntryVersion());
      }
    });

    // Make sure vm1 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 putAll on an accessor and get the version
information. bug #48205
geode-core/src/distributedTest/java/org/apache/geode/pdx/ClientsWithVersioningRetryDUnitTest.java