@Ignore("Bug50247")
  @Test
  public void testPersistentPartitionedRegionWithGatewaySender_Restart() {
    // create locator on local site
    Integer lnPort = vm0.invoke(() -> WANTestBase.createFirstLocatorWithDSId(1));
    // create locator on remote site
    Integer nyPort = vm1.invoke(() -> WANTestBase.createFirstRemoteLocator(2, lnPort));

    // create receiver on remote site
    createCacheInVMs(nyPort, vm2, vm3);
    createReceiverInVMs(vm2, vm3);

    // create cache in local site
    createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);

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

    // create PR on remote site
    vm2.invoke(() -> WANTestBase.createPersistentPartitionedRegion(getTestMethodName(), null, 1,
        100, isOffHeap()));
    vm3.invoke(() -> WANTestBase.createPersistentPartitionedRegion(getTestMethodName(), null, 1,
        100, isOffHeap()));

    // create PR on local site
    vm4.invoke(() -> WANTestBase.createPersistentPartitionedRegion(getTestMethodName(), "ln", 1,
        100, isOffHeap()));
    vm5.invoke(() -> WANTestBase.createPersistentPartitionedRegion(getTestMethodName(), "ln", 1,
        100, isOffHeap()));
    vm6.invoke(() -> WANTestBase.createPersistentPartitionedRegion(getTestMethodName(), "ln", 1,
        100, isOffHeap()));
    vm7.invoke(() -> WANTestBase.createPersistentPartitionedRegion(getTestMethodName(), "ln", 1,
        100, isOffHeap()));

    // start the senders on local site
    startSenderInVMs("ln", vm4, vm5, vm6, vm7);

    // wait for senders to become running
    vm4.invoke(waitForSenderRunnable());
    vm5.invoke(waitForSenderRunnable());
    vm6.invoke(waitForSenderRunnable());
    vm7.invoke(waitForSenderRunnable());

    // pause the senders
    vm4.invoke(pauseSenderRunnable());
    vm5.invoke(pauseSenderRunnable());
    vm6.invoke(pauseSenderRunnable());
    vm7.invoke(pauseSenderRunnable());

    // start puts in region on local site
    vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
    logger.info("Completed puts in the region");

    // ----------------- Close and rebuild local site -------------------------------------
    // kill the senders
    vm4.invoke(killSenderRunnable());
    vm5.invoke(killSenderRunnable());
    vm6.invoke(killSenderRunnable());
    vm7.invoke(killSenderRunnable());

    logger.info("Killed all the senders.");

    // restart the vm
    createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);

    logger.info("Created back the cache");

    // create back the senders
    vm4.invoke(() -> WANTestBase.createSender("ln", 2, true, 100, 10, false, false, null, true));
    vm5.invoke(() -> WANTestBase.createSender("ln", 2, true, 100, 10, false, false, null, true));
    vm6.invoke(() -> WANTestBase.createSender("ln", 2, true, 100, 10, false, false, null, true));
    vm7.invoke(() -> WANTestBase.createSender("ln", 2, true, 100, 10, false, false, null, true));

    logger.info("Created the senders again");

    // start the senders
    startSenderInVMsAsync("ln", vm4, vm5, vm6, vm7);

    logger.info("Started the senders.");

    logger.info("Waiting for senders running.");

    // wait for senders running
    vm4.invoke(waitForSenderRunnable());
    vm5.invoke(waitForSenderRunnable());
    vm6.invoke(waitForSenderRunnable());
    vm7.invoke(waitForSenderRunnable());

    logger.info("All the senders are now running...");

    // create PR on local site
    AsyncInvocation<Void> inv1 = vm4.invokeAsync(() -> WANTestBase
        .createPersistentPartitionedRegion(getTestMethodName(), "ln", 1, 100, isOffHeap()));
    AsyncInvocation<Void> inv2 = vm5.invokeAsync(() -> WANTestBase
        .createPersistentPartitionedRegion(getTestMethodName(), "ln", 1, 100, isOffHeap()));
    AsyncInvocation<Void> inv3 = vm6.invokeAsync(() -> WANTestBase
        .createPersistentPartitionedRegion(getTestMethodName(), "ln", 1, 100, isOffHeap()));
    AsyncInvocation<Void> inv4 = vm7.invokeAsync(() -> WANTestBase
        .createPersistentPartitionedRegion(getTestMethodName(), "ln", 1, 100, isOffHeap()));

    try {
      inv1.join();
      inv2.join();
      inv3.join();
      inv4.join();
    } catch (InterruptedException e) {
      e.printStackTrace();
      fail();
    }

    logger.info("Created back the partitioned regions");

    // -------------------------------------------------------------------------------------------

    // start puts in region on local site
    vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName(), 3000));
    logger.info("Completed puts in the region");

    vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 3000));
    vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName(), 3000));
  }
Create persistent PR and non-persistent GatewaySender. After doing puts, close the local site.
Rebuild the PR from persistent disk store and create the sender again. Do more puts. Check if
the remote site receives newly added events.

This test can be used as DUnit test for defect #50247 reported by Indian Railways. At present,
customer is using this configuration and which is not recommended since it can lead to event
loss of GatewaySender events.
geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelWANPersistenceEnabledGatewaySenderDUnitTest.java