@Test
  public void destroyParallelGatewaySenderShouldNotStopDispatchingFromOtherSendersAttachedToTheRegion() {
    String site2SenderId = "site2-sender";
    String site3SenderId = "site3-sender";
    String regionName = testName.getMethodName();
    int[] ports = getRandomAvailableTCPPorts(3);
    int site1Port = ports[0];
    int site2Port = ports[1];
    int site3Port = ports[2];
    Set<String> site1RemoteLocators =
        Stream.of("localhost[" + site2Port + "]", "localhost[" + site3Port + "]")
            .collect(Collectors.toSet());
    Set<String> site2RemoteLocators =
        Stream.of("localhost[" + site1Port + "]", "localhost[" + site3Port + "]")
            .collect(Collectors.toSet());
    Set<String> site3RemoteLocators =
        Stream.of("localhost[" + site1Port + "]", "localhost[" + site2Port + "]")
            .collect(Collectors.toSet());

    // Start 3 sites.
    vm0.invoke(() -> createLocator(1, site1Port,
        Collections.singleton("localhost[" + site1Port + "]"), site1RemoteLocators));
    vm1.invoke(() -> createLocator(2, site2Port,
        Collections.singleton("localhost[" + site2Port + "]"), site2RemoteLocators));
    vm2.invoke(() -> createLocator(3, site3Port,
        Collections.singleton("localhost[" + site3Port + "]"), site3RemoteLocators));

    // Create the cache on the 3 sites.
    createCacheInVMs(site1Port, vm3);
    createCacheInVMs(site2Port, vm4);
    createCacheInVMs(site3Port, vm5);

    // Create receiver and region on sites 2 and 3.
    asList(vm4, vm5).forEach(vm -> vm.invoke(() -> {
      createReceiver();
      createPartitionedRegion(regionName, null, 1, 113, isOffHeap());
    }));

    // Create senders and partitioned region on site 1.
    vm3.invoke(() -> {
      createSender(site2SenderId, 2, true, 100, 20, false, false, null, false);
      createSender(site3SenderId, 3, true, 100, 20, false, false, null, false);
      waitForSenderRunningState(site2SenderId);
      waitForSenderRunningState(site3SenderId);

      createPartitionedRegion(regionName, String.join(",", site2SenderId, site3SenderId), 1, 113,
          isOffHeap());
    });

    // #################################################################################### //

    final int FIRST_BATCH = 100;
    final int SECOND_BATCH = 200;
    final Map<String, String> firstBatch = new HashMap<>();
    IntStream.range(0, FIRST_BATCH).forEach(i -> firstBatch.put("Key" + i, "Value" + i));
    final Map<String, String> secondBatch = new HashMap<>();
    IntStream.range(FIRST_BATCH, SECOND_BATCH)
        .forEach(i -> secondBatch.put("Key" + i, "Value" + i));

    // Insert first batch and wait until the queues are empty.
    vm3.invoke(() -> {
      cache.getRegion(regionName).putAll(firstBatch);
      checkQueueSize(site2SenderId, 0);
      checkQueueSize(site3SenderId, 0);
    });

    // Wait until sites 2 and 3 have received all updates.
    asList(vm4, vm5).forEach(vm -> vm.invoke(() -> {
      Region<String, String> region = cache.getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isEqualTo(FIRST_BATCH));
      firstBatch.forEach((key, value) -> assertThat(region.get(key)).isEqualTo(value));
    }));

    // Stop sender to site3, remove it from the region and destroy it.
    vm3.invoke(() -> {
      stopSender(site3SenderId);
      removeSenderFromTheRegion(site3SenderId, regionName);
      destroySender(site3SenderId);
      verifySenderDestroyed(site3SenderId, true);
    });

    // Insert second batch and wait until the queue is empty.
    vm3.invoke(() -> {
      cache.getRegion(regionName).putAll(secondBatch);
      checkQueueSize(site2SenderId, 0);
    });

    // Site 3 should only have the first batch.
    vm5.invoke(() -> {
      Region<String, String> region = cache.getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isEqualTo(FIRST_BATCH));
      firstBatch.forEach((key, value) -> assertThat(region.get(key)).isEqualTo(value));
    });

    // Site 2 should have both batches.
    vm4.invoke(() -> {
      Region<String, String> region = cache.getRegion(regionName);
      await().untilAsserted(() -> assertThat(region.size()).isEqualTo(SECOND_BATCH));
      firstBatch.forEach((key, value) -> assertThat(region.get(key)).isEqualTo(value));
      secondBatch.forEach((key, value) -> assertThat(region.get(key)).isEqualTo(value));
    });
  }
Since the sender is attached to a region and in use, it can not be destroyed. Hence, exception
is thrown by the sender API.
geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDistributedTest.java