public void addShadowPartitionedRegionForUserRR(DistributedRegion userRegion) {
    sender.getLifeCycleLock().writeLock().lock();
    PartitionedRegion prQ = null;

    if (logger.isDebugEnabled()) {
      logger.debug(
          "addShadowPartitionedRegionForUserRR: Going to create shadowpr for userRegion {}",
          userRegion.getFullPath());
    }

    try {
      String regionName = userRegion.getFullPath();

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

      InternalCache cache = sender.getCache();
      final String prQName = getQueueName(sender.getId(), userRegion.getFullPath());
      prQ = (PartitionedRegion) cache.getRegion(prQName);
      if (prQ == null) {
        InternalRegionFactory fact = cache.createInternalRegionFactory();
        // Fix for 48621 - don't enable concurrency checks
        // for queue buckets., event with persistence
        fact.setConcurrencyChecksEnabled(false);
        PartitionAttributesFactory pfact = new PartitionAttributesFactory();
        pfact.setTotalNumBuckets(sender.getMaxParallelismForReplicatedRegion());
        int localMaxMemory =
            userRegion.getDataPolicy().withStorage() ? sender.getMaximumQueueMemory() : 0;
        pfact.setLocalMaxMemory(localMaxMemory);
        pfact.setRedundantCopies(3); // TODO:Kishor : THis need to be handled nicely
        pfact.setPartitionResolver(new RREventIDResolver());
        if (sender.isPersistenceEnabled()) {
          fact.setDataPolicy(DataPolicy.PERSISTENT_PARTITION);
        }

        fact.setDiskStoreName(sender.getDiskStoreName());

        // if persistence is enabled, set the diskSyncronous to whatever user
        // has set
        // else set it to false
        // optimize with above check of enable persistence
        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 =
            new ParallelGatewaySenderQueueMetaRegion(prQName, ra, null, cache, sender,
                sender.getStatisticsClock());

        fact.setInternalMetaRegion(meta).setDestroyLockFlag(true)
            .setSnapshotInputStream(null).setImageTarget(null);
        prQ = (PartitionedRegion) fact.create(prQName);

        if (logger.isDebugEnabled()) {
          logger.debug("Region created  : {} partition Attributes : {}", prQ,
              prQ.getPartitionAttributes());
        }

        // TODO This should not be set on the PR but on the GatewaySender
        prQ.enableConflation(sender.isBatchConflationEnabled());

        // Before going ahead, make sure all the buckets of shadowPR are
        // loaded
        // and primary nodes have been decided.
        // This is required in case of persistent PR and sender.
        if (prQ.getLocalMaxMemory() != 0) {
          for (final Integer integer : prQ.getRegionAdvisor().getBucketSet()) {
          }
        }
        // In case of Replicated Region it may not be necessary.

        if (logger.isDebugEnabled()) {
          logger.debug("{}: Created queue region: {}", this, prQ);
        }
      } else {
        // in case shadowPR exists already (can be possible when sender is
        // started from stop operation)
        // for first processor only
        if (index == 0) {
          handleShadowPRExistsScenario(cache, prQ);
        }
      }
      /*
       * 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();
      }
    } finally {
      if (prQ != null) {
        userRegionNameToShadowPRMap.put(userRegion.getFullPath(), prQ);
      }
      sender.getLifeCycleLock().writeLock().unlock();
    }
  }
Start the background batch removal thread.
geode-core/src/main/java/org/apache/geode/internal/cache/wan/parallel/ParallelGatewaySenderQueue.java