@Override
  protected void process(ClusterDistributionManager dm) {
    final boolean isDebugEnabled = logger.isDebugEnabled();
    final InternalCache cache = dm.getCache();
    if (cache != null) {
      final InitializationLevel oldLevel =
          LocalRegion.setThreadInitLevelRequirement(BEFORE_INITIAL_IMAGE);
      try {
        for (Object name : regionToDispatchedKeysMap.keySet()) {
          final String regionName = (String) name;
          final PartitionedRegion region = (PartitionedRegion) cache.getRegion(regionName);
          if (region == null) {
            continue;
          } else {
            AbstractGatewaySender abstractSender = region.getParallelGatewaySender();
            // Find the map: bucketId to dispatchedKeys
            // Find the bucket
            // Destroy the keys
            Map bucketIdToDispatchedKeys = (Map) regionToDispatchedKeysMap.get(regionName);
            for (Object bId : bucketIdToDispatchedKeys.keySet()) {
              final String bucketFullPath =
                  SEPARATOR + PartitionedRegionHelper.PR_ROOT_REGION_NAME + SEPARATOR
                      + region.getBucketName((Integer) bId);
              AbstractBucketRegionQueue brq =
                  (AbstractBucketRegionQueue) cache.getInternalRegionByPath(bucketFullPath);
              if (isDebugEnabled) {
                logger.debug(
                    "ParallelQueueRemovalMessage : The bucket in the cache is bucketRegionName : {} bucket: {}",
                    bucketFullPath, brq);
              }

              List dispatchedKeys = (List) bucketIdToDispatchedKeys.get(bId);
              if (dispatchedKeys != null) {
                for (Object key : dispatchedKeys) {
                  // First, clear the Event from tempQueueEvents at AbstractGatewaySender level, if
                  // exists
                  // synchronize on AbstractGatewaySender.queuedEventsSync while doing so
                  abstractSender.removeFromTempQueueEvents(key);

                  if (brq != null) {
                    if (brq.isInitialized()) {
                      if (isDebugEnabled) {
                        logger.debug(
                            "ParallelQueueRemovalMessage : The bucket {} is initialized. Destroying the key {} from BucketRegionQueue.",
                            bucketFullPath, key);
                      }
                      // fix for #48082
                      afterAckForSecondary_EventInBucket(abstractSender, brq, key);
                      destroyKeyFromBucketQueue(brq, key, region);
                    } else {
                      // if bucket is not initialized, the event should either be in bucket or
                      // tempQueue
                      boolean isDestroyed = false;
                      if (isDebugEnabled) {
                        logger.debug(
                            "ParallelQueueRemovalMessage : The bucket {} is not yet initialized.",
                            bucketFullPath);
                      }
                      brq.getInitializationLock().readLock().lock();
                      try {
                        if (brq.containsKey(key)) {
                          // fix for #48082
                          afterAckForSecondary_EventInBucket(abstractSender, brq, key);
                          destroyKeyFromBucketQueue(brq, key, region);
                          isDestroyed = true;
                        }

                        // Even if BucketRegionQueue does not have the key, it could be in the
                        // tempQueue
                        // remove it from there..defect #49196
                        destroyFromTempQueue(brq.getPartitionedRegion(), (Integer) bId, key);

                        // Finally, add the key to the failed batch removal keys so that it is
                        // definitely removed from the bucket region queue
                        brq.addToFailedBatchRemovalMessageKeys(key);
                      } finally {
                        brq.getInitializationLock().readLock().unlock();
                      }
                    }
                  } else {// brq is null. Destroy the event from tempQueue. Defect #49196
                    destroyFromTempQueue(region, (Integer) bId, key);
                  }
                }
              }
            }
          }
        } // for loop regionToDispatchedKeysMap.keySet()
      } finally {
        LocalRegion.setThreadInitLevelRequirement(oldLevel);
      }
    } // cache != null
  }
Removes a batch of events from the remote secondary queues

@since GemFire 8.0
geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelQueueRemovalMessage.java