public void addShadowPartitionedRegionForUserPR(PartitionedRegion userPR,
      PartitionedRegion childPR) {
    if (logger.isDebugEnabled()) {
      logger.debug(
          "{} addShadowPartitionedRegionForUserPR: Attempting to create queue region: {}; child region: {}",
          this, userPR.getDisplayName(), childPR == null ? "null" : childPR.getDisplayName());
    }
    sender.getLifeCycleLock().writeLock().lock();

    PartitionedRegion prQ = null;
    try {
      String regionName = userPR.getFullPath();
      // Find if there is any parent region for this userPR
      // if there is then no need to add another q for the same
      String leaderRegionName = ColocationHelper.getLeaderRegion(userPR).getFullPath();
      if (!regionName.equals(leaderRegionName)) {
        // Fix for defect #50364. Allow user to attach GatewaySender to child PR (without attaching
        // to leader PR)
        // though, internally, colocate the GatewaySender's shadowPR with the leader PR in
        // colocation chain
        if (!userRegionNameToShadowPRMap.containsKey(leaderRegionName)) {
          addShadowPartitionedRegionForUserPR(ColocationHelper.getLeaderRegion(userPR), userPR);
        }
        return;
      }

      if (userRegionNameToShadowPRMap.containsKey(regionName)) {
        return;
      }

      if ((childPR == null ? userPR : childPR).getDataPolicy().withPersistence()
          && !sender.isPersistenceEnabled()) {
        throw new GatewaySenderException(
            String.format(
                "Non persistent gateway sender %s can not be attached to persistent region %s",
                sender.getId(), userPR.getFullPath()));
      }

      InternalCache cache = sender.getCache();
      boolean isAccessor = (userPR.getLocalMaxMemory() == 0);

      final String prQName = sender.getId() + QSTRING + convertPathToName(userPR.getFullPath());

      prQ = (PartitionedRegion) cache.getRegion(prQName, true);
      if (prQ != null && prQ.isDestroyed()) {
        PartitionedRegion oldPrQ = prQ;
        while (oldPrQ == prQ) {
          try {
            Thread.sleep(50);
          } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
          }
          if (logger.isDebugEnabled()) {
            logger.debug("wait for destroy to finish");
          }
          prQ = (PartitionedRegion) cache.getRegion(prQName, true);
        }
      }

      if ((prQ != null) && (index == 0) && cleanQueues) {
        cleanOverflowStats(cache);
        prQ.clearPREvictionControllerFromDiskInitialization();
        prQ.destroyRegion(null);
        prQ = null;
      }

      if (prQ == null) {
        RegionShortcut regionShortcut;
        if (sender.isPersistenceEnabled() && !isAccessor) {
          regionShortcut = RegionShortcut.PARTITION_PERSISTENT;
        } else {
          regionShortcut = RegionShortcut.PARTITION;
        }
        InternalRegionFactory fact = cache.createInternalRegionFactory(regionShortcut);

        fact.setConcurrencyChecksEnabled(false);
        PartitionAttributesFactory pfact = new PartitionAttributesFactory();
        pfact.setTotalNumBuckets(userPR.getTotalNumberOfBuckets());
        pfact.setRedundantCopies(userPR.getRedundantCopies());
        pfact.setColocatedWith(regionName);
        // EITHER set localMaxMemory to 0 for accessor node
        // OR override shadowPRs default local max memory with the sender's max
        // queue memory (Fix for bug#44254)
        int localMaxMemory = isAccessor ? 0 : sender.getMaximumQueueMemory();
        pfact.setLocalMaxMemory(localMaxMemory);

        pfact.setStartupRecoveryDelay(userPR.getPartitionAttributes().getStartupRecoveryDelay());
        pfact.setRecoveryDelay(userPR.getPartitionAttributes().getRecoveryDelay());

        fact.setDiskStoreName(sender.getDiskStoreName());

        // if persistence is enabled, set the diskSyncronous to whatever user has set
        // else set it to false
        if (sender.isPersistenceEnabled()) {
          fact.setDiskSynchronous(sender.isDiskSynchronous());
        } else {
          fact.setDiskSynchronous(false);
        }

        // allow for no overflow directory
        EvictionAttributes ea = EvictionAttributes.createLIFOMemoryAttributes(
            sender.getMaximumQueueMemory(), EvictionAction.OVERFLOW_TO_DISK);

        fact.setEvictionAttributes(ea);
        fact.setPartitionAttributes(pfact.create());

        final RegionAttributes ra = fact.getCreateAttributes();

        if (logger.isDebugEnabled()) {
          logger.debug("{}: Attempting to create queue region: {}", this, prQName);
        }

        ParallelGatewaySenderQueueMetaRegion meta =
            metaRegionFactory.newMetataRegion(cache, prQName, ra, sender);

        fact.setInternalMetaRegion(meta);
        fact.setDestroyLockFlag(true);
        fact.setInternalRegion(true);
        fact.setSnapshotInputStream(null);
        fact.setImageTarget(null);
        prQ = (PartitionedRegion) fact.create(prQName);
        // at this point we should be able to assert prQ == meta;

        if (prQ == null) {
          return;
        }
        // TODO This should not be set on the PR but on the GatewaySender
        prQ.enableConflation(sender.isBatchConflationEnabled());
        if (isAccessor) {
          return; // return from here if accessor node
        }

        // Add the overflow statistics to the mbean
        addOverflowStatisticsToMBean(cache, prQ);

        // Wait for buckets to be recovered.
        prQ.shadowPRWaitForBucketRecovery();

        if (logger.isDebugEnabled()) {
          logger.debug("{}: Created queue region: {}", this, prQ);
        }

      } else {
        if (isAccessor) {
          return; // return from here if accessor node
        }
        // in case shadowPR exists already (can be possible when sender is
        // started from stop operation)
        // for first parallelGatewaySenderQueue only
        if (index == 0) {
          handleShadowPRExistsScenario(cache, prQ);
        }

      }

    } finally {
      if (prQ != null) {
        userRegionNameToShadowPRMap.put(userPR.getFullPath(), prQ);
        prQ.setSentGatewaySenderStoppedMessage(false);
        if (prQ.getDataStore() != null) {
          final Set<BucketRegion> buckets = prQ.getDataStore().getAllLocalBucketRegions();
          for (BucketRegion br : buckets) {
            br.setReceivedGatewaySenderStoppedMessage(false);
          }
        }
      }
      /*
       * Here, enqueueTempEvents need to be invoked when a sender is already running and userPR is
       * created later. When the flow comes here through start() method of sender i.e. userPR
       * already exists and sender is started later, the enqueueTempEvents is done in the start()
       * method of ParallelGatewaySender
       */
      if ((index == nDispatcher - 1) && sender.isRunning()) {
        sender.enqueueTempEvents();
      }
      afterRegionAdd(userPR);
      sender.getLifeCycleLock().writeLock().unlock();
    }
  }
Start the background batch removal thread.
geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java