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

    createCacheInVMs(nyPort, vm2, vm3);
    createReceiverInVMs(vm2, vm3);

    createCacheInVMs(lnPort, vm4, vm5, vm6, vm7);

    LogWriterUtils.getLogWriter().info("Created cache on local site");

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

    LogWriterUtils.getLogWriter().info("Created senders on local site");

    vm4.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", "ln", 1, 100,
        isOffHeap()));
    vm5.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", "ln", 1, 100,
        isOffHeap()));
    vm6.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", "ln", 1, 100,
        isOffHeap()));
    vm7.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", "ln", 1, 100,
        isOffHeap()));

    LogWriterUtils.getLogWriter().info("Created PRs on local site");

    vm2.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", null, 1, 100,
        isOffHeap()));
    vm3.invoke(() -> WANTestBase.createPartitionedRegion(getTestMethodName() + "_PR", null, 1, 100,
        isOffHeap()));
    LogWriterUtils.getLogWriter().info("Created PRs on remote site");

    vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName() + "_PR", 1000));
    LogWriterUtils.getLogWriter().info("Done 1000 puts on local site");

    // Since puts are already done on userPR, it will have the buckets created.
    // During sender start, it will wait until those buckets are created for shadowPR as well.
    // Start the senders in async threads, so colocation of shadowPR will be complete and
    // missing buckets will be created in PRHARedundancyProvider.createMissingBuckets().
    startSenderInVMsAsync("ln", vm4, vm5, vm6, vm7);

    vm4.invoke(() -> WANTestBase.waitForSenderRunningState("ln"));
    vm5.invoke(() -> WANTestBase.waitForSenderRunningState("ln"));
    vm6.invoke(() -> WANTestBase.waitForSenderRunningState("ln"));
    vm7.invoke(() -> WANTestBase.waitForSenderRunningState("ln"));

    LogWriterUtils.getLogWriter().info("Started senders on local site");

    vm4.invoke(() -> WANTestBase.doPuts(getTestMethodName() + "_PR", 5000));
    LogWriterUtils.getLogWriter().info("Done 5000 puts on local site");

    vm4.invoke(() -> WANTestBase.pauseSender("ln"));
    vm5.invoke(() -> WANTestBase.pauseSender("ln"));
    vm6.invoke(() -> WANTestBase.pauseSender("ln"));
    vm7.invoke(() -> WANTestBase.pauseSender("ln"));
    LogWriterUtils.getLogWriter().info("Paused senders on local site");

    vm4.invoke(() -> WANTestBase.verifySenderPausedState("ln"));
    vm5.invoke(() -> WANTestBase.verifySenderPausedState("ln"));
    vm6.invoke(() -> WANTestBase.verifySenderPausedState("ln"));
    vm7.invoke(() -> WANTestBase.verifySenderPausedState("ln"));

    AsyncInvocation<Void> inv1 =
        vm4.invokeAsync(() -> WANTestBase.doPuts(getTestMethodName() + "_PR", 1000));
    LogWriterUtils.getLogWriter().info("Started 1000 async puts on local site");

    vm4.invoke(() -> WANTestBase.resumeSender("ln"));
    vm5.invoke(() -> WANTestBase.resumeSender("ln"));
    vm6.invoke(() -> WANTestBase.resumeSender("ln"));
    vm7.invoke(() -> WANTestBase.resumeSender("ln"));
    LogWriterUtils.getLogWriter().info("Resumed senders on local site");

    vm4.invoke(() -> WANTestBase.verifySenderResumedState("ln"));
    vm5.invoke(() -> WANTestBase.verifySenderResumedState("ln"));
    vm6.invoke(() -> WANTestBase.verifySenderResumedState("ln"));
    vm7.invoke(() -> WANTestBase.verifySenderResumedState("ln"));

    try {
      inv1.join();
    } catch (InterruptedException e) {
      e.printStackTrace();
      fail("Interrupted the async invocation.");
    }

    // verify all buckets drained on all sender nodes.
    vm4.invoke(() -> WANTestBase.validateParallelSenderQueueAllBucketsDrained("ln"));
    vm5.invoke(() -> WANTestBase.validateParallelSenderQueueAllBucketsDrained("ln"));
    vm6.invoke(() -> WANTestBase.validateParallelSenderQueueAllBucketsDrained("ln"));
    vm7.invoke(() -> WANTestBase.validateParallelSenderQueueAllBucketsDrained("ln"));

    vm2.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_PR", 5000));
    vm3.invoke(() -> WANTestBase.validateRegionSize(getTestMethodName() + "_PR", 5000));
  }
Normal scenario in which a combinations of start, pause, resume operations is tested
geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/concurrent/ConcurrentParallelGatewaySenderOperation_1_DUnitTest.java