public void destroyParallelGatewaySenderRegion(Operation op, boolean cacheWrite, boolean lock,
      boolean callbackEvents) {

    if (logger.isDebugEnabled()) {
      logger.debug("Destroying parallel queue region for senders: {}",
          getParallelGatewaySenderIds());
    }

    boolean keepWaiting = true;

    while (true) {
      List<String> pausedSenders = new ArrayList<>();
      List<ConcurrentParallelGatewaySenderQueue> parallelQueues =
          new ArrayList<>();
      isDestroyedForParallelWAN = true;
      int countOfQueueRegionsToBeDestroyed = 0;
      for (String senderId : getParallelGatewaySenderIds()) {
        AbstractGatewaySender sender =
            (AbstractGatewaySender) cache.getGatewaySender(senderId);
        if (sender == null || sender.getEventProcessor() == null) {
          continue;
        }

        if (cacheWrite) { // in case of destroy operation senders should be
                          // resumed
          if (sender.isPaused()) {
            pausedSenders.add(senderId);
            continue;
          }
        }

        if (pausedSenders.isEmpty()) { // if there are puase sender then only
                                       // check for other pause senders instead
                                       // of creating list of shadowPR
          AbstractGatewaySenderEventProcessor ep = sender.getEventProcessor();
          if (ep == null) {
            continue;
          }
          ConcurrentParallelGatewaySenderQueue parallelQueue =
              (ConcurrentParallelGatewaySenderQueue) ep.getQueue();
          PartitionedRegion parallelQueueRegion = parallelQueue.getRegion(getFullPath());

          // this may be removed in previous iteration
          if (parallelQueueRegion == null || parallelQueueRegion.isDestroyed
              || parallelQueueRegion.isClosed) {
            continue;
          }

          parallelQueues.add(parallelQueue);
          countOfQueueRegionsToBeDestroyed++;
        }
      }

      if (!pausedSenders.isEmpty()) {
        String exception = null;
        if (pausedSenders.size() == 1) {
          exception =
              String.format("GatewaySender %s is paused. Resume it before destroying region %s.",
                  pausedSenders, getName());
        } else {
          exception =
              String.format(
                  "GatewaySenders %s are paused. Resume them before destroying region %s.",
                  pausedSenders, getName());
        }
        isDestroyedForParallelWAN = false;
        throw new GatewaySenderException(exception);
      }

      if (countOfQueueRegionsToBeDestroyed == 0) {
        break;
      }

      for (ConcurrentParallelGatewaySenderQueue parallelQueue : parallelQueues) {
        PartitionedRegion parallelQueueRegion = parallelQueue.getRegion(getFullPath());
        // CacheWrite true == distributedDestoy. So in case of false, dont wait
        // for queue to drain
        // parallelQueueRegion.size() = With DistributedDestroym wait for queue
        // to drain
        // keepWaiting : comes from the MAXIMUM_SHUTDOWN_WAIT_TIME case handled
        if (cacheWrite && parallelQueueRegion.size() != 0 && keepWaiting) {
          continue;
        } else {// In any case, destroy shadow PR locally. distributed destroy of
                // userPR will take care of detsroying shadowPR locally on other
                // nodes.
          RegionEventImpl event = null;
          if (op.isClose()) { // In case of cache close operation, we want SPR's basic destroy to go
                              // through CACHE_CLOSE condition of postDestroyRegion not
                              // closePartitionedRegion code
            event = new RegionEventImpl(parallelQueueRegion, op, null, false, getMyId(),
                generateEventID());
          } else {
            event = new RegionEventImpl(parallelQueueRegion, Operation.REGION_LOCAL_DESTROY, null,
                false, getMyId(), generateEventID());
          }
          parallelQueueRegion.basicDestroyRegion(event, false, lock, callbackEvents);
          parallelQueue.removeShadowPR(getFullPath());
          countOfQueueRegionsToBeDestroyed--;
          continue;
        }
      }

      if (countOfQueueRegionsToBeDestroyed == 0) {
        break;
      }

      if (cacheWrite) {
        if (AbstractGatewaySender.MAXIMUM_SHUTDOWN_WAIT_TIME == -1) {
          keepWaiting = true;
          try {
            Thread.sleep(5000);
          } catch (InterruptedException ignore) {
            // interrupted
          }
        } else {
          try {
            Thread.sleep(AbstractGatewaySender.MAXIMUM_SHUTDOWN_WAIT_TIME * 1000L);
          } catch (InterruptedException ignore) {/* ignore */
            // interrupted
          }
          keepWaiting = false;
        }
      }
    }
  }
Returns existing ColocationLogger or creates and sets a new instance if one does not yet exist.
Spins to ensure that if another thread sets ColocationLogger, we return that one instead of
setting another instance. Checks CancelCriterion during spin and either returns an instance or
throws NullPointerException. The only way to exist the loop is with an instance or by throwing
CancelException if the Cache closes.
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegion.java