boolean recoverPersistentBuckets() {
    /*
     * To handle a case where ParallelGatewaySender is persistent but userPR is not First recover
     * the GatewaySender buckets for ParallelGatewaySender irrespective of whether colocation is
     * complete or not.
     */
    PartitionedRegion leaderRegion = ColocationHelper.getLeaderRegion(partitionedRegion);


    // Check if the leader region or some child shadow PR region is persistent
    // and return the first persistent region found
    PartitionedRegion persistentLeader = getPersistentLeader();

    // If there is no persistent region in the colocation chain, no need to recover.
    if (persistentLeader == null) {
      return true;
    }

    if (!checkMembersColocation(leaderRegion, leaderRegion.getDistributionManager().getId())) {
      if (logger.isDebugEnabled()) {
        logger.debug("Skipping persistent recovery of {} because colocation is not complete for {}",
            partitionedRegion, leaderRegion);
      }
      return false;
    }

    ProxyBucketRegion[] proxyBucketArray =
        persistentLeader.getRegionAdvisor().getProxyBucketArray();
    if (proxyBucketArray.length == 0) {
      throw new IllegalStateException("Unexpected empty proxy bucket array");
    }
    for (ProxyBucketRegion proxyBucket : proxyBucketArray) {
      proxyBucket.initializePersistenceAdvisor();
    }

    Set<InternalDistributedMember> peers = partitionedRegion.getRegionAdvisor().adviseGeneric();

    // We need to make sure here that we don't run into this race condition:
    // 1) We get a membership view from member A
    // 2) Member B removes itself, and distributes to us and A. We don't remove B
    // 3) We apply the membership view from A, which includes B.
    // That will add B back into the set.
    // This state flush will make sure that any membership changes
    // That are in progress on the peers are finished.
    MembershipFlushRequest.send(peers, partitionedRegion.getDistributionManager(),
        partitionedRegion.getFullPath());

    List<ProxyBucketRegion> bucketsNotHostedLocally = new ArrayList<>(proxyBucketArray.length);
    List<ProxyBucketRegion> bucketsHostedLocally = new ArrayList<>(proxyBucketArray.length);

    createPersistentBucketRecoverer(proxyBucketArray.length);

    /*
     * Spawn a separate thread for bucket that we previously hosted to recover that bucket.
     *
     * That thread will get to the point at which it has determined that at least one member
     * (possibly the local member) has fully initialized the bucket, at which it will count down the
     * someMemberRecoveredLatch latch on the bucket.
     *
     * Once at least one copy of each bucket has been created in the distributed system, the
     * initPRInternals method will exit. Some of the threads spawned here will still be doing GII's
     * in the background. This allows the system to become usable as fast as possible.
     *
     * If we used a bounded thread pool here, we end up waiting for some buckets to finish there GII
     * before returning from initPRInternals. In the future maybe we could let the create bucket
     * return and pass the GII task to a separate thread pool.
     *
     */
    for (ProxyBucketRegion proxyBucket : proxyBucketArray) {
      if (proxyBucket.getPersistenceAdvisor().wasHosting()) {
        RecoveryRunnable recoveryRunnable = new RecoveryRunnable(this) {

          @Override
          public void run() {
            // make sure that we always count down this latch, even if the region was destroyed.
            try {
              super.run();
            } finally {
              if (getPersistentBucketRecoverer() != null) {
                getPersistentBucketRecoverer().countDown();
              }
            }
          }

          @Override
          public void run2() {
            proxyBucket.recoverFromDiskRecursively();
          }
        };

        Thread recoveryThread =
            new LoggingThread("Recovery thread for bucket " + proxyBucket.getName(),
                false, recoveryRunnable);
        recoveryThread.start();
        bucketsHostedLocally.add(proxyBucket);
      } else {
        bucketsNotHostedLocally.add(proxyBucket);
      }
    }

    try {
      // try to recover the local buckets before the proxy buckets. This will allow us to detect
      // any ConflictingDataException before the proxy buckets update their membership view.
      for (ProxyBucketRegion proxyBucket : bucketsHostedLocally) {
        proxyBucket.waitForPrimaryPersistentRecovery();
      }

      if (!partitionedRegion.isInternalRegion() && !bucketsNotHostedLocally.isEmpty()) {
        partitionedRegion.notifyRegionCreated();
      }

      for (ProxyBucketRegion proxyBucket : bucketsNotHostedLocally) {
        proxyBucket.recoverFromDiskRecursively();
      }
    } finally {
      if (getPersistentBucketRecoverer() != null) {
        getPersistentBucketRecoverer().countDown(bucketsNotHostedLocally.size());
      }
    }

    return true;
  }
Schedule a task to perform redundancy recovery for a new node or for the node departed.
geode-core/src/main/java/org/apache/geode/internal/cache/PRHARedundancyProvider.java