@Test
  public void testWANPDX_PR_ParallelSender_WithDelayedTypeRegistry()
      throws InterruptedException, ExecutionException {
    Integer lnPort = vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
    Integer nyPort = vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));

    // Create the receiver side of the WAN gateway. Only vm2 will be a receiver, vm3 is
    // just a peer
    createCacheInVMs(nyPort, vm2, vm3);
    vm2.invoke(WANTestBase::createReceiver);
    vm2.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", null, 0, 4,
        isOffHeap()));
    vm3.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", null, 0, 4,
        isOffHeap()));

    AsyncInvocation<Void> deserializationFuture;
    try {
      // Delay processing of sending type registry update from vm2
      vm2.invoke(() -> {
        DistributionMessageObserver.setInstance(new BlockingPdxTypeUpdateObserver());
      });

      // Create the sender side of the WAN connection. 2 VMs, with paused senders
      vm4.invoke(() -> WANTestBase.createCache(lnPort));
      vm5.invoke(() -> WANTestBase.createCache(lnPort));

      vm4.invoke(() -> WANTestBase.createSender("ln", 2, true, 100, 10, false, false, null, false));
      vm5.invoke(() -> WANTestBase.createSender("ln", 2, true, 100, 10, false, false, null, false));

      // Create the partitioned region in vm4
      vm4.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", "ln", 0, 4,
          isOffHeap()));

      vm5.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", "ln", 0, 4,
          isOffHeap()));

      vm5.invoke(() -> {
        Region region = cache.getRegion(getTestMethodName() + "_PR");
        PartitionRegionHelper.assignBucketsToPartitions(region);
      });

      vm4.invoke(() -> WANTestBase.pauseSender("ln"));
      vm5.invoke(() -> WANTestBase.pauseSender("ln"));

      // Do some puts to fill up our queues
      vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName() + "_PR", 20));

      vm4.invoke(() -> {
        final Region r = cache.getRegion(SEPARATOR + getTestMethodName() + "_PR");
        PdxValue result = (PdxValue) r.put(KEY_0, new PdxValue(0));
      });

      // Force VM4 to be the primary
      vm4.invoke(() -> {
        final Region region = cache.getRegion(SEPARATOR + getTestMethodName() + "_PR");
        DistributedMember primary = PartitionRegionHelper.getPrimaryMemberForKey(region, KEY_0);
        // If we are not the primary
        DistributedMember localMember = cache.getDistributedSystem().getDistributedMember();
        if (!primary.equals(localMember)) {
          PartitionRegionHelper.moveBucketByKey(region, primary, localMember, KEY_0);

        }
      });

      vm5.invoke(() -> WANTestBase.resumeSender("ln"));

      boolean blocking = vm2.invoke(() -> {
        BlockingPdxTypeUpdateObserver observer =
            (BlockingPdxTypeUpdateObserver) DistributionMessageObserver.getInstance();
        return observer.startedBlocking.await(1, TimeUnit.MINUTES);
      });

      assertTrue(blocking);

      vm4.invoke(() -> WANTestBase.resumeSender("ln"));

      vm2.invoke(() -> {
        final Region region = cache.getRegion(SEPARATOR + getTestMethodName() + "_PR");
        await().until(() -> region.containsKey(KEY_0));

      });

      // Make sure vm3 can deserialize the value
      deserializationFuture = vm3.invokeAsync(() -> {
        final Region r = cache.getRegion(SEPARATOR + getTestMethodName() + "_PR");
        PdxValue result = (PdxValue) r.get(KEY_0);
        assertEquals(result, new PdxValue(0));
      });

      try {
        deserializationFuture.await(10, TimeUnit.SECONDS);
        fail("Get should have been blocked waiting for PDX type to be distributed");
      } catch (TimeoutException e) {
        // This is what we hope will happen. The get will be blocked by some sort of lock, rather
        // than failing due to a missing type.
      }

    } finally {

      vm2.invoke(() -> {
        BlockingPdxTypeUpdateObserver observer =
            (BlockingPdxTypeUpdateObserver) DistributionMessageObserver.getInstance();
        observer.latch.countDown();
      });
    }

    deserializationFuture.get();
  }
Test 1> Site 1 : 1 locator, 1 member 2> Site 2 : 1 locator, 1 member 3> PR is defined on member
1 on site1 4> Parallel GatewaySender is defined on member 1 on site1 5> Same PR is defined on
site2 member 1 6> Put is done with value which is PDXSerializable 7> Validate whether other
sites member receive this put operation.
geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/misc/PDXNewWanDUnitTest.java