@Override
  public Set<PersistentMemberID> getMembersToWaitFor(
      Set<PersistentMemberID> previouslyOnlineMembers, Set<PersistentMemberID> offlineMembers)
      throws ReplyException {
    PersistentMemberID myPersistentID = getPersistentID();
    PersistentMemberID myInitializingId = getInitializingID();

    // This is the set of members that are currently waiting for this member
    // to come online.
    Set<PersistentMemberID> membersToWaitFor = new HashSet<>(previouslyOnlineMembers);
    offlineMembers.addAll(previouslyOnlineMembers);

    // If our persistent ID is null, we need to wait for all of the previously online members.
    if (myPersistentID != null || myInitializingId != null) {
      Set<InternalDistributedMember> members = cacheDistributionAdvisor.adviseProfileUpdate();
      Set<InternalDistributedMember> membersHostingThisRegion =
          cacheDistributionAdvisor.adviseGeneric();

      // Fetch the persistent view from all of our peers.
      PersistentStateQueryResults results = fetchPersistentStateQueryResults(members,
          cacheDistributionAdvisor.getDistributionManager(), myPersistentID, myInitializingId);

      // iterate through all of the peers. For each peer: if the member was previously online
      // according
      // to us, grab its online members and add them to the members to wait for set. We may need to
      // do this several times until we discover all of the members that may have newer data than
      // us.
      boolean addedMembers = true;
      while (addedMembers) {
        addedMembers = false;
        for (Entry<InternalDistributedMember, Set<PersistentMemberID>> entry : results
            .getOnlineMemberMap()
            .entrySet()) {
          InternalDistributedMember memberId = entry.getKey();
          Set<PersistentMemberID> peersOnlineMembers = entry.getValue();
          PersistentMemberID persistentID = results.getPersistentIds().get(memberId);
          PersistentMemberID initializingID = results.getInitializingIds().get(memberId);
          if (membersToWaitFor.contains(persistentID)
              || membersToWaitFor.contains(initializingID)) {
            for (PersistentMemberID peerOnlineMember : peersOnlineMembers) {
              if (!isRevoked(peerOnlineMember)
                  && !peerOnlineMember.getDiskStoreId().equals(getDiskStoreID())
                  && !persistentMemberView.getOfflineMembers().contains(peerOnlineMember)) {
                if (membersToWaitFor.add(peerOnlineMember)) {
                  addedMembers = true;
                  // Make sure we also persist that this member is online.
                  persistentMemberView.memberOnline(peerOnlineMember);
                  if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
                    logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
                        "{}-{}: Adding {} to the list of members we're wait for, because {} has newer or equal data than is and is waiting for that member",
                        shortDiskStoreId(), regionPath, peerOnlineMember, memberId);
                  }
                }
              }
            }
          }
        }
      }
      removeOlderMembers(membersToWaitFor);
      if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
        logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
            "{}-{}: Initial state of membersToWaitFor, before pruning {}", shortDiskStoreId(),
            regionPath, membersToWaitFor);
      }

      // For each of our peers, see what our state is according to their view.
      for (Map.Entry<InternalDistributedMember, PersistentMemberState> entry : results
          .getStateOnPeers().entrySet()) {
        InternalDistributedMember memberId = entry.getKey();
        PersistentMemberID persistentID = results.getPersistentIds().get(memberId);
        PersistentMemberID initializingID = results.getInitializingIds().get(memberId);
        DiskStoreID diskStoreID = results.getDiskStoreIds().get(memberId);
        PersistentMemberState state = entry.getValue();

        if (PersistentMemberState.REVOKED.equals(state)) {
          throw new RevokedPersistentDataException(
              String.format(
                  "The persistent member id %s has been revoked in this distributed system. You cannot recover from disk files which have been revoked.",
                  myPersistentID));
        }

        // If the peer thinks we are newer or equal to them, we don't need to wait for this peer.
        if (membersHostingThisRegion.contains(memberId) && persistentID != null && state != null
            && myInitializingId == null && (state.equals(PersistentMemberState.ONLINE)
                || state.equals(PersistentMemberState.EQUAL))) {
          if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
            logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
                "{}-{}: Not waiting for {} because it thinks our state was {}", shortDiskStoreId(),
                regionPath, persistentID, state);
          }
          removeNewerPersistentID(membersToWaitFor, persistentID);
        }

        // If the peer has an initialized ID, they are no longer offline.
        if (persistentID != null) {
          removeNewerPersistentID(offlineMembers, persistentID);
        }

        // If the peer thinks we are newer or equal to them, we don't need to wait for this peer.
        if (membersHostingThisRegion.contains(memberId) && initializingID != null && state != null
            && (state.equals(PersistentMemberState.ONLINE)
                || state.equals(PersistentMemberState.EQUAL))) {
          if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
            logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
                "{}-{}: Not waiting for {} because it thinks our state was {}", shortDiskStoreId(),
                regionPath, initializingID, state);
          }
          removeByDiskStoreID(membersToWaitFor, diskStoreID, false);
        }

        // If the peer has an initializing id, they are also not online.
        if (initializingID != null) {
          removeNewerPersistentID(offlineMembers, initializingID);
        }

        handlePartiallyDestroyedRegion(offlineMembers, membersToWaitFor, persistentID,
            initializingID, diskStoreID);
      }
    }
    return membersToWaitFor;
  }
@param previouslyOnlineMembers the members we have persisted online in our persistence files
@param offlineMembers This method will populate this set with any members that we are waiting
       for which are actually not running right now. This is different from the set of members
       we need to wait for - this member may end up waiting on a member that is actually
       running.
@return the list of members that this member needs to wait for before it can initialize.
geode-core/src/main/java/org/apache/geode/internal/cache/persistence/PersistenceAdvisorImpl.java