class PersistenceInitialImageAdvisor {

  @VisibleForTesting
  InitialImageAdvice getAdvice(InitialImageAdvice previousAdvice) {
    final boolean isPersistAdvisorDebugEnabled =
        logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE);

    MembershipChangeListener listener = membershipChangeListenerProvider.apply(persistenceAdvisor);
    cacheDistributionAdvisor.addMembershipAndProxyListener(listener);
    persistenceAdvisor.addListener(listener);
    try {
      while (true) {
        cacheDistributionAdvisor.getAdvisee().getCancelCriterion().checkCancelInProgress(null);
        try {
          // On first pass, previous advice is null. On subsequent passes, it's the advice
          // from the previous iteration.
          InitialImageAdvice advice =
              cacheDistributionAdvisor.adviseInitialImage(previousAdvice, true);

          if (hasReplicates(advice)) {
            if (hasDiskImageToRecoverFrom) {
              removeReplicatesIfWeAreEqualToAnyOrElseClearEqualMembers(advice.getReplicates());
            }
            return advice;
          }
          if (hasNonPersistentMember(advice)) {
            updateMembershipViewFromAnyPeer(advice.getNonPersistent(), hasDiskImageToRecoverFrom);
          }

          // If there are online members that we previously failed to get a GII from, retry those
          // members rather than wait for new persistent members to recover.
          if (hasReplicates(previousAdvice)) {
            logger.info(
                "GII failed from all sources, but members are still online. Retrying the GII.");
            previousAdvice = null;
            continue;
          }

          Set<PersistentMemberID> previouslyOnlineMembers =
              persistenceAdvisor.getPersistedOnlineOrEqualMembers();

          // If there are no currently online members, and no previously online members, this member
          // should just go with what's on its own disk
          if (previouslyOnlineMembers.isEmpty()) {
            if (isPersistAdvisorDebugEnabled()) {
              logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
                  "{}-{}: No previously online members. Recovering with the data from the local disk",
                  shortDiskStoreID, regionPath);
            }
            return advice;
          }

          Set<PersistentMemberID> offlineMembers = new HashSet<>();
          Set<PersistentMemberID> membersToWaitFor =
              persistenceAdvisor.getMembersToWaitFor(previouslyOnlineMembers, offlineMembers);

          if (membersToWaitFor.isEmpty()) {
            if (isPersistAdvisorDebugEnabled()) {
              logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
                  "{}-{}: All of the previously online members are now online and waiting for us. Acquiring tie lock. Previously online members {}",
                  shortDiskStoreID, regionPath, advice.getReplicates());
            }
            if (persistenceAdvisor.acquireTieLock()) {
              return refreshInitialImageAdviceAndThenCheckMyStateWithReplicates(previousAdvice);
            }
          } else {
            if (isPersistAdvisorDebugEnabled) {
              logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
                  "{}-{}: Going to wait for these member ids: {}", shortDiskStoreID, regionPath,
                  membersToWaitFor);
            }
          }

          waitForMembershipChangeForMissingDiskStores(listener, offlineMembers, membersToWaitFor);
        } catch (InterruptedException e) {
          logger.debug("Interrupted while trying to determine latest persisted copy", e);
        }
      }
    } finally {
      persistenceAdvisor.setWaitingOnMembers(null, null);
      cacheDistributionAdvisor.removeMembershipAndProxyListener(listener);
      persistenceAdvisor.removeListener(listener);
    }
  }

    PersistenceInitialImageAdvisor(InternalPersistenceAdvisor persistenceAdvisor,
      String shortDiskStoreID, String regionPath, CacheDistributionAdvisor cacheDistributionAdvisor,
      boolean hasDiskImageToRecoverFrom);
  @VisibleForTesting  PersistenceInitialImageAdvisor(InternalPersistenceAdvisor persistenceAdvisor,
      String shortDiskStoreID, String regionPath, CacheDistributionAdvisor cacheDistributionAdvisor,
      boolean hasDiskImageToRecoverFrom,
      Function<InternalPersistenceAdvisor, MembershipChangeListener> membershipChangeListenerProvider);

}