@Test
  public void testRetryPutAll() {
    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);

    createServerRegion(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");

        // Add a listener to close vm1 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 beforeSendMessage(ClusterDistributionManager dm,
                  DistributionMessage message) {
                if (message instanceof DistributedPutAllOperation.PutAllMessage) {
                  DistributionMessageObserver.setInstance(null);
                  disconnectFromDS(vm1);
                }
              }
            });

      }
    });

    int port1 = createServerRegion(vm1,
        RegionShortcut.PARTITION_REDUNDANT_PERSISTENT);
    int port2 = createServerRegion(vm2,
        RegionShortcut.PARTITION_REDUNDANT_PERSISTENT);
    createClientRegion(vm3, port1, port2);

    // This will be a put all to bucket 0
    // Here's the expected sequence
    // client->vm1 (accessor0)
    // vm1->vm0
    // vm0 will kill vm1
    // vm0->vm2
    // client will retry the putall
    vm3.invoke(new SerializableCallable() {
      @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);
        RegionEntry entry = ((LocalRegion) region).getRegionEntry(0);
        assertNotNull(entry);
        assertNotNull(entry.getVersionStamp());
        assertEquals(2, entry.getVersionStamp().getEntryVersion());
        return null;
      }
    });

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

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

    // Make sure vm1 did in fact shut down
    vm1.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. bug
#45059
geode-core/src/distributedTest/java/org/apache/geode/pdx/ClientsWithVersioningRetryDUnitTest.java