@Test
  public void testNoDistributedDeadlockWithGatewaySenderStop() throws Exception {
    addIgnoredException("Broken pipe");
    Integer[] locatorPorts = createLNAndNYLocators();
    Integer lnPort = locatorPorts[0];
    Integer nyPort = locatorPorts[1];
    VM[] senders = {vm4, vm5, vm6, vm7};
    try {
      for (VM sender : senders) {
        sender.invoke(() -> AbstractGatewaySender.doSleepForTestingInDistribute.set(true));
        sender.invoke(() -> ClusterOperationExecutors.maxPrThreadsForTest.set(2));
      }
      vm2.invoke(() -> ClusterOperationExecutors.maxPrThreadsForTest.set(2));
      vm3.invoke(() -> ClusterOperationExecutors.maxPrThreadsForTest.set(2));

      createSendersReceiversAndPartitionedRegion(lnPort, nyPort, false, true, true);

      // make sure all the senders are running before doing any puts
      waitForSendersRunning();

      // Send a fairly big amount of operations to provoke the deadlock
      int invocationsPerServer = 4;
      AsyncInvocation[] invocations = new AsyncInvocation[senders.length * invocationsPerServer];
      for (int i = 0; i < senders.length; i++) {
        for (int j = 0; j < invocationsPerServer; j++) {
          invocations[i + (j * invocationsPerServer)] =
              senders[i].invokeAsync(() -> doPuts(getUniqueName() + "_PR", 100));
        }
      }

      // Wait for some elements to be replicated before stopping the senders
      for (int i = 0; i < senders.length; i++) {
        senders[i].invoke(() -> await()
            .untilAsserted(() -> assertThat(getSenderStats("ln", -1).get(3)).isGreaterThan(1)));
      }

      stopSendersAsync();
      for (int i = 0; i < invocations.length; i++) {
        invocations[i].await();
      }
    } finally {
      for (int i = 0; i < senders.length; i++) {
        senders[i].invoke(() -> AbstractGatewaySender.doSleepForTestingInDistribute.set(false));
      }
    }
  }
Verifies that no distributed deadlock occurs when stopping a gateway sender while receiving
traffic.
The distributed deadlock may occur when the gateway sender tries to get the
size of the gateway sender queue (sending a size message to other members) while holding the
lifeCycleLock lock. This lock is also taken when an event is to be distributed by the gateway
sender.
As this issue has only been observed in the field with a lot of traffic, in order to reproduce
it in a test case, conserve-sockets is set to true (although the deadlock has also
been seen with conserve-sockets=false), the size of the PartitionedRegion thread pool is set
to a small value and an artificial timeout is added at a point in the distribute() call
of the AbstractGatewaySeder class.
geode-wan/src/distributedTest/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderOperationsDistributedTest.java