/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.internal.cache.persistence;

import static java.lang.System.lineSeparator;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.TestOnly;

import org.apache.geode.CancelException;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MutableForTesting;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.persistence.ConflictingPersistentDataException;
import org.apache.geode.cache.persistence.RevokedPersistentDataException;
import org.apache.geode.distributed.DistributedLockService;
import org.apache.geode.distributed.internal.DistributionAdvisor.Profile;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.ProfileListener;
import org.apache.geode.distributed.internal.ReplyException;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.CopyOnWriteHashSet;
import org.apache.geode.internal.cache.BucketAdvisor;
import org.apache.geode.internal.cache.CacheDistributionAdvisor;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.CacheProfile;
import org.apache.geode.internal.cache.CacheDistributionAdvisor.InitialImageAdvice;
import org.apache.geode.internal.cache.DiskRegionStats;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.ProxyBucketRegion;
import org.apache.geode.internal.cache.persistence.PersistentMemberManager.MemberRevocationListener;
import org.apache.geode.internal.cache.persistence.PersistentStateQueryMessage.PersistentStateQueryReplyProcessor;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.process.StartupStatus;
import org.apache.geode.internal.util.TransformUtils;
import org.apache.geode.internal.util.TransformUtils.CollectionTransformer;
import org.apache.geode.internal.util.Transformer;
import org.apache.geode.logging.internal.log4j.api.LogService;

public class PersistenceAdvisorImpl implements InternalPersistenceAdvisor {

  private static final Logger logger = LogService.getLogger();

  @Immutable
  private static final PersistenceAdvisorObserver DEFAULT_PERSISTENCE_ADVISOR_OBSERVER = s -> {
  };

  @MutableForTesting
  private static PersistenceAdvisorObserver persistenceAdvisorObserver =
      DEFAULT_PERSISTENCE_ADVISOR_OBSERVER;

  protected final Object lock;

  protected final CacheDistributionAdvisor cacheDistributionAdvisor;
  protected final String regionPath;
  protected final PersistentMemberView persistentMemberView;
  private final DiskRegionStats diskRegionStats;
  private final PersistentMemberManager persistentMemberManager;
  private final ProfileChangeListener profileChangeListener;

  private final Set<PersistentMemberID> recoveredMembers;
  private final Set<PersistentMemberID> removedMembers = new HashSet<>();
  private final Set<PersistentMemberID> equalMembers;
  private final DistributedLockService distributedLockService;

  private volatile boolean holdingTieLock;

  private volatile boolean online;
  private volatile Set<PersistentStateListener> persistentStateListeners = Collections.emptySet();
  private volatile boolean initialized;
  private volatile boolean shouldUpdatePersistentView;
  protected volatile boolean isClosed;

  protected volatile Set<PersistentMemberID> allMembersWaitingFor;
  protected volatile Set<PersistentMemberID> offlineMembersWaitingFor;

  private final PersistentStateQueryMessageSenderFactory persistentStateQueryMessageSenderFactory;

  private final StartupStatus startupStatus;
  private final Transformer<PersistentMemberID, String> persistentMemberIdTransformer;
  private final CollectionTransformer<PersistentMemberID, String> collectionTransformer;

  public PersistenceAdvisorImpl(CacheDistributionAdvisor cacheDistributionAdvisor,
      DistributedLockService distributedLockService, PersistentMemberView persistentMemberView,
      String regionPath, DiskRegionStats diskRegionStats,
      PersistentMemberManager persistentMemberManager) {
    this(cacheDistributionAdvisor, distributedLockService, persistentMemberView, regionPath,
        diskRegionStats, persistentMemberManager, new StartupStatus(),
        TransformUtils.persistentMemberIdToLogEntryTransformer,
        TransformUtils::transform,
        new PersistentStateQueryMessageSenderFactory());
  }

  @VisibleForTesting
  PersistenceAdvisorImpl(CacheDistributionAdvisor cacheDistributionAdvisor,
      DistributedLockService distributedLockService, PersistentMemberView persistentMemberView,
      String regionPath, DiskRegionStats diskRegionStats,
      PersistentMemberManager persistentMemberManager,
      StartupStatus startupStatus,
      Transformer<PersistentMemberID, String> persistentMemberIdTransformer,
      CollectionTransformer<PersistentMemberID, String> collectionTransformer,
      PersistentStateQueryMessageSenderFactory persistentStateQueryMessageSenderFactory) {
    this.cacheDistributionAdvisor = cacheDistributionAdvisor;
    this.distributedLockService = distributedLockService;
    this.regionPath = regionPath;
    this.persistentMemberView = persistentMemberView;
    this.diskRegionStats = diskRegionStats;
    this.persistentMemberManager = persistentMemberManager;
    this.startupStatus = startupStatus;
    this.persistentMemberIdTransformer = persistentMemberIdTransformer;
    this.collectionTransformer = collectionTransformer;
    this.persistentStateQueryMessageSenderFactory = persistentStateQueryMessageSenderFactory;

    profileChangeListener = new ProfileChangeListener();

    // Prevent membership changes while we are persisting the membership view online. If we
    // synchronize on something else, we need to be careful about lock ordering because the
    // membership notifications are called with the advisor lock held.
    lock = cacheDistributionAdvisor;

    // Remember which members we know about because of what we have persisted. We will later use
    // this to handle updates from peers.
    recoveredMembers = getPersistedMembers();

    // To prevent races if we crash during initialization, mark equal members as online before we
    // initialize. We will still report these members as equal, but if we crash and recover they
    // will no longer be considered equal.
    equalMembers = new CopyOnWriteHashSet<>(persistentMemberView.getOfflineAndEqualMembers());
    for (PersistentMemberID id : equalMembers) {
      persistentMemberView.memberOnline(id);
    }
  }

  @Override
  public void initialize() {
    if (initialized) {
      return;
    }

    if (wasAboutToDestroy()) {
      logger.info("Region {} crashed during a region destroy. Finishing the destroy.", regionPath);
      finishPendingDestroy();
    }

    cacheDistributionAdvisor.addProfileChangeListener(profileChangeListener);

    Set<PersistentMemberPattern> revokedMembers = persistentMemberManager
        .addRevocationListener(profileChangeListener, persistentMemberView.getRevokedMembers());

    for (PersistentMemberPattern pattern : revokedMembers) {
      memberRevoked(pattern);
    }

    // Start logging changes to the persistent view
    startMemberLogging();

    initialized = true;
  }

  /**
   * Adds a PersistentStateListener whose job is to log changes in the persistent view.
   */
  private void startMemberLogging() {
    addListener(new PersistentStateListener.PersistentStateAdapter() {
      /**
       * A persistent member has gone offline. Log the offline member and log which persistent
       * members are still online (the current persistent view).
       */
      @Override
      public void memberOffline(InternalDistributedMember member, PersistentMemberID persistentID) {
        if (logger.isDebugEnabled()) {
          Set<PersistentMemberID> members =
              new HashSet<>(cacheDistributionAdvisor.adviseInitializedPersistentMembers().values());
          members.remove(persistentID);

          Set<String> onlineMembers = new HashSet<>();
          collectionTransformer.transform(members, onlineMembers, persistentMemberIdTransformer);

          logger.info(
              "The following persistent member has gone offline for region {}: {}  Remaining participating members for the region include: {}",
              regionPath, persistentMemberIdTransformer.transform(persistentID), onlineMembers);
        }
      }
    });
  }

  @Override
  public PersistentStateQueryResults getMyStateOnMembers(Set<InternalDistributedMember> members)
      throws ReplyException {
    return fetchPersistentStateQueryResults(members,
        cacheDistributionAdvisor.getDistributionManager(), persistentMemberView.getMyPersistentID(),
        persistentMemberView.getMyInitializingID());
  }

  private PersistentStateQueryResults fetchPersistentStateQueryResults(
      Set<InternalDistributedMember> members, DistributionManager dm,
      PersistentMemberID persistentMemberID, PersistentMemberID initializingMemberId) {
    PersistentStateQueryReplyProcessor replyProcessor = persistentStateQueryMessageSenderFactory
        .createPersistentStateQueryReplyProcessor(dm, members);
    PersistentStateQueryMessage message =
        persistentStateQueryMessageSenderFactory.createPersistentStateQueryMessage(regionPath,
            persistentMemberID, initializingMemberId, replyProcessor.getProcessorId());
    return message.send(members, dm, replyProcessor);
  }

  /**
   * Return what state we have persisted for a given peer's id.
   */
  @Override
  public PersistentMemberState getPersistedStateOfMember(PersistentMemberID id) {
    if (isRevoked(id)) {
      return PersistentMemberState.REVOKED;
    }

    // If the peer is marked as equal, indicate they are equal
    if (equalMembers != null && equalMembers.contains(id)) {
      return PersistentMemberState.EQUAL;
    }

    // If we have a member that is marked as online that is an older version of the peers id, tell
    // them they are online
    for (PersistentMemberID onlineMember : persistentMemberView.getOnlineMembers()) {
      if (onlineMember.isOlderOrEqualVersionOf(id)) {
        return PersistentMemberState.ONLINE;
      }
    }

    // If we have a member that is marked as offline that is a newer version of the peers id, tell
    // them they are online
    for (PersistentMemberID offline : persistentMemberView.getOfflineMembers()) {
      if (id.isOlderOrEqualVersionOf(offline)) {
        return PersistentMemberState.OFFLINE;
      }
    }
    return null;
  }

  @Override
  public void updateMembershipView(InternalDistributedMember peer, boolean targetReinitializing) {
    beginUpdatingPersistentView();
    DistributionManager dm = cacheDistributionAdvisor.getDistributionManager();
    PersistentMembershipView peersPersistentMembershipView =
        MembershipViewRequest.send(peer, dm, regionPath, targetReinitializing);
    if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
      logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE, "{}-{}: Updating persistent view from {}",
          shortDiskStoreId(), regionPath, peer);
    }

    synchronized (lock) {
      PersistentMemberID myId = getPersistentID();
      Map<InternalDistributedMember, PersistentMemberID> peersOnlineMembers =
          peersPersistentMembershipView.getOnlineMembers();
      Set<PersistentMemberID> peersOfflineMembers =
          peersPersistentMembershipView.getOfflineMembers();

      for (PersistentMemberID id : peersOnlineMembers.values()) {
        if (!isRevoked(id) && !removedMembers.contains(id)) {
          if (!id.equals(myId) && !recoveredMembers.remove(id)
              && !id.getDiskStoreId().equals(getDiskStoreID())) {
            if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
              logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
                  "{}-{}: Processing membership view from peer. Marking {} as online because {} says its online",
                  shortDiskStoreId(), regionPath, id, peer);
            }
            persistentMemberView.memberOnline(id);
          }
        }
      }

      for (PersistentMemberID id : peersOfflineMembers) {
        if (!isRevoked(id) && !removedMembers.contains(id)) {
          // This method is called before the current member is online. if the peer knows about a
          // member that the current member doesn't know about, that means that member must have
          // been added to the DS after the current member went offline. Therefore, that member is
          // *newer* than the current member. So mark that member as online (meaning, online later
          // than the current member).
          if (!id.equals(myId) && !recoveredMembers.remove(id)
              && !id.getDiskStoreId().equals(getDiskStoreID())) {
            if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
              logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
                  "{}-{}: Processing membership view from peer. Marking {} as online because {} says its offline, but we have never seen it",
                  shortDiskStoreId(), regionPath, id, peer);
            }
            persistentMemberView.memberOnline(id);
          }
        }
      }

      for (PersistentMemberID id : recoveredMembers) {
        if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
          logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
              "{}-{}: Processing membership view from peer. Removing {} because {} doesn't have it",
              shortDiskStoreId(), regionPath, id, peer);
        }
        persistentMemberView.memberRemoved(id);
      }
    }

    // Update the set of revoked members from the peer. This should be called without holding the
    // lock to avoid deadlocks
    Set<PersistentMemberPattern> revokedMembers = peersPersistentMembershipView.getRevokedMembers();
    for (PersistentMemberPattern revoked : revokedMembers) {
      persistentMemberManager.revokeMember(revoked);
    }
  }

  private boolean isRevoked(PersistentMemberID id) {
    return persistentMemberManager.isRevoked(regionPath, id);
  }

  @Override
  public void setOnline(boolean didGII, boolean atomicCreation, PersistentMemberID newId)
      throws ReplyException {
    if (online) {
      return;
    }

    if (!didGII) {
      setInitializing(newId);
    }

    synchronized (lock) {

      // Transition any members that are marked as online, but not actually currently running, to
      // offline.
      Set<PersistentMemberID> membersToMarkOffline =
          new HashSet<>(persistentMemberView.getOnlineMembers());
      Map<InternalDistributedMember, PersistentMemberID> onlineMembers;
      if (!atomicCreation) {
        onlineMembers = cacheDistributionAdvisor.adviseInitializedPersistentMembers();
      } else {
        // If this is an atomic bucket creation, don't mark our peers, which are concurrently
        // initializing, as offline they have the exact same data as we do (none), so we are not
        // technically "newer," and this avoids a race where both members can think the other is
        // offline ("older").
        onlineMembers = cacheDistributionAdvisor.advisePersistentMembers();
      }
      membersToMarkOffline.removeAll(onlineMembers.values());

      // Don't mark equal members as offline if that are currently running. We don't have newer data
      // than these members so this is safe, and it it avoids a race where we mark them offline at
      // this point, and then later they mark us as offline.
      if (equalMembers != null && !equalMembers.isEmpty()) {

        // This is slightly hacky. We're looking for a running member that has the same disk store
        // as our equal members, because all have is a persistent id of the equal members. The
        // persistent id of the running member may be different than what we have marked as equal,
        // because the id in the profile is the new id for the member.
        Collection<PersistentMemberID> allMembers =
            cacheDistributionAdvisor.advisePersistentMembers().values();
        Set<DiskStoreID> runningDiskStores = new HashSet<>();
        for (PersistentMemberID mem : allMembers) {
          runningDiskStores.add(mem.getDiskStoreId());
        }
        // Remove any equal members which are not actually running right now.
        for (PersistentMemberID id : equalMembers) {
          if (!runningDiskStores.contains(id.getDiskStoreId())) {
            equalMembers.remove(id);
          }
        }
        membersToMarkOffline.removeAll(equalMembers);
      }
      for (PersistentMemberID id : membersToMarkOffline) {
        persistentMemberView.memberOffline(id);
      }
      if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
        logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
            "{}-{}: Persisting the new membership view and ID as online. Online members {}. Offline members {}. Equal memebers {}.",
            shortDiskStoreId(), regionPath, persistentMemberView.getOnlineMembers(),
            persistentMemberView.getOfflineMembers(), equalMembers);
      }

      persistentMemberView.setInitialized();
      online = true;
      removedMembers.clear();
    }
    if (diskRegionStats != null) {
      diskRegionStats.incInitializations(!didGII);
    }
  }

  /**
   * Start listening for persistent view updates and apply any updates that have already happened.
   *
   * <p>
   * This method should be called after we have decided that there is no conflicting persistent
   * exception.
   */
  protected void beginUpdatingPersistentView() {
    synchronized (lock) {
      // Only update the view if it is has not already happened.
      if (!shouldUpdatePersistentView) {
        shouldUpdatePersistentView = true;
        Map<InternalDistributedMember, PersistentMemberID> onlineMembers =
            cacheDistributionAdvisor.adviseInitializedPersistentMembers();
        for (Map.Entry<InternalDistributedMember, PersistentMemberID> entry : onlineMembers
            .entrySet()) {
          memberOnline(entry.getKey(), entry.getValue());
        }
      }
    }
  }

  @Override
  public void setInitializing(PersistentMemberID newId) {
    beginUpdatingPersistentView();

    DistributionManager dm = cacheDistributionAdvisor.getDistributionManager();

    PersistentMemberID oldId = getPersistentID();
    PersistentMemberID initializingId = getInitializingID();

    Set<InternalDistributedMember> profileUpdateRecipients =
        cacheDistributionAdvisor.adviseProfileUpdate();
    if (newId == null || !newId.equals(oldId) && !newId.equals(initializingId)) {
      // If we have not yet prepared the old id, prepare it now.

      // This will only be the case if we crashed while initializing previously. In the case, we are
      // essentially finishing what we started by preparing that ID first. This will remove that ID
      // from the peers.
      if (initializingId != null) {
        if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
          logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
              "{}-{}: We still have an initializing id: {}. Telling peers to remove the old id {} and transitioning this initializing id to old id. recipients {}",
              shortDiskStoreId(), regionPath, initializingId, oldId, profileUpdateRecipients);
        }
        long viewVersion = cacheDistributionAdvisor.startOperation();
        try {
          PrepareNewPersistentMemberMessage.send(profileUpdateRecipients, dm, regionPath, oldId,
              initializingId);
        } finally {
          if (viewVersion != -1) {
            cacheDistributionAdvisor.endOperation(viewVersion);
          }
        }
        oldId = initializingId;
      }

      if (logger.isDebugEnabled()) {
        logger.debug("Persisting my new persistent ID {}", newId);
      }
      persistentMemberView.setInitializing(newId);
    }

    profileUpdateRecipients = cacheDistributionAdvisor.adviseProfileUpdate();
    if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
      logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
          "{}-{}: Sending the new ID to peers. They should remove the old id {}. Recipients: {}",
          shortDiskStoreId(), regionPath, oldId, profileUpdateRecipients);
    }
    if (newId != null) {
      PrepareNewPersistentMemberMessage.send(profileUpdateRecipients, dm, regionPath, oldId, newId);
    }
  }

  @Override
  public PersistentMemberID generatePersistentID() {
    return persistentMemberView.generatePersistentID();
  }

  @Override
  public PersistentMembershipView getMembershipView() {
    if (!initialized) {
      return null;
    }
    Set<PersistentMemberID> offlineMembers = getPersistedMembers();
    Map<InternalDistributedMember, PersistentMemberID> onlineMembers =
        cacheDistributionAdvisor.adviseInitializedPersistentMembers();
    offlineMembers.removeAll(onlineMembers.values());

    PersistentMemberID myId = getPersistentID();
    if (myId != null) {
      onlineMembers
          .put(cacheDistributionAdvisor.getDistributionManager().getDistributionManagerId(), myId);
    }

    return new PersistentMembershipView(offlineMembers, onlineMembers,
        persistentMemberManager.getRevokedMembers());
  }

  @Override
  public Set<PersistentMemberID> getPersistedMembers() {
    Set<PersistentMemberID> persistentMembers = new HashSet<>();
    persistentMembers.addAll(persistentMemberView.getOfflineMembers());
    persistentMembers.addAll(persistentMemberView.getOfflineAndEqualMembers());
    persistentMembers.addAll(persistentMemberView.getOnlineMembers());
    return persistentMembers;
  }

  @Override
  public boolean checkMyStateOnMembers(Set<InternalDistributedMember> replicates)
      throws ReplyException {
    PersistentStateQueryResults remoteStates = getMyStateOnMembers(replicates);
    Set<InternalDistributedMember> copyOfReplicates = null;

    persistenceAdvisorObserver.observe(regionPath);

    boolean equal = false;
    PersistentMemberID myId = getPersistentID();
    for (Map.Entry<InternalDistributedMember, PersistentMemberState> entry : remoteStates
        .getStateOnPeers().entrySet()) {
      InternalDistributedMember member = entry.getKey();
      PersistentMemberID remoteId = remoteStates.getPersistentIds().get(member);

      PersistentMemberState stateOnPeer = entry.getValue();
      if (PersistentMemberState.REVOKED.equals(stateOnPeer)) {
        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.",
                myId));
      }

      if (myId != null && stateOnPeer == null) {
        String message = String.format(
            "Region %s remote member %s with persistent data %s was not part of the same distributed system as the local data from %s",
            regionPath, member, remoteId, myId);
        // Conceptually the removed member due to not knowing current member, should be equal to
        // existing replicates.
        // It can still be used as GII provider candidate. Use a copyOfReplicates to avoid modifying
        // the replicates.
        if (copyOfReplicates == null) {
          copyOfReplicates = new HashSet<>(replicates);
        }

        boolean gwRegion = false;

        if (cacheDistributionAdvisor instanceof BucketAdvisor) {
          BucketAdvisor ba = (BucketAdvisor) cacheDistributionAdvisor;
          if (ba.getAdvisee() instanceof ProxyBucketRegion) {
            ProxyBucketRegion pbr = (ProxyBucketRegion) ba.getAdvisee();
            PartitionedRegion pr = pbr.getPartitionedRegion();
            if (pr != null) {
              gwRegion = pr.isShadowPR();
            }
          }
        }

        copyOfReplicates.remove(member);
        if (copyOfReplicates.isEmpty() && !gwRegion) {
          throw new ConflictingPersistentDataException(message);
        } else {
          logger.info(message);
        }
      }

      if (myId != null && stateOnPeer == PersistentMemberState.EQUAL) {
        equal = true;
      }

      // The other member changes its ID when it comes back online.
      if (remoteId != null) {
        PersistentMemberState remoteState = getPersistedStateOfMember(remoteId);
        if (remoteState == PersistentMemberState.OFFLINE) {
          String message =
              String.format(
                  "Region %s refusing to initialize from member %s with persistent data %s which was offline when the local data from %s was last online",
                  regionPath, member, remoteId, myId);
          throw new ConflictingPersistentDataException(message);
        }
      }
    }
    return equal;
  }

  @VisibleForTesting
  public static void setPersistenceAdvisorObserver(PersistenceAdvisorObserver o) {
    persistenceAdvisorObserver = o == null ? DEFAULT_PERSISTENCE_ADVISOR_OBSERVER : o;
  }

  @Override
  public PersistentMemberID getPersistentIDIfOnline() {
    if (online) {
      return persistentMemberView.getMyPersistentID();
    }
    return null;
  }

  private void memberOffline(InternalDistributedMember distributedMember,
      PersistentMemberID persistentID) {
    if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
      logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
          "{}-{}: Member offine. id={}, persistentID={}", shortDiskStoreId(), regionPath,
          distributedMember, persistentID);
    }
    synchronized (lock) {
      boolean foundMember = recoveredMembers.remove(persistentID);
      foundMember |= equalMembers.remove(persistentID);
      foundMember |= getPersistedMembers().contains(persistentID);
      // Don't persist members as offline until we are online. Otherwise, we may think we have later
      // data than them during recovery.
      if (shouldUpdatePersistentView && online) {
        try {
          // Don't persistent members as offline if we have already persisted them as equal.
          if (persistentMemberView.getOfflineAndEqualMembers().contains(persistentID)) {
            return;
          }
          // Don't mark the member as offline if we have never seen it. If we haven't seen it that
          // means it's not done initializing yet.
          if (foundMember) {
            if (PersistenceObserverHolder.getInstance().memberOffline(regionPath, persistentID)) {
              persistentMemberView.memberOffline(persistentID);
            }
            PersistenceObserverHolder.getInstance().afterPersistedOffline(regionPath, persistentID);
          }
        } catch (DiskAccessException e) {
          logger.warn("Unable to persist membership change", e);
        }
      }
      notifyListenersMemberOffline(distributedMember, persistentID);
    }
  }

  private void memberOnline(InternalDistributedMember distributedMember,
      PersistentMemberID persistentID) {
    if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
      logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
          "{}-{}: Sending the new ID to peers.  Member online. id={}, persistentID={}",
          shortDiskStoreId(), regionPath, distributedMember, persistentID);
    }
    synchronized (lock) {
      if (shouldUpdatePersistentView) {
        recoveredMembers.remove(persistentID);
        try {
          if (PersistenceObserverHolder.getInstance().memberOnline(regionPath, persistentID)) {
            persistentMemberView.memberOnline(persistentID);
          }
          PersistenceObserverHolder.getInstance().afterPersistedOnline(regionPath, persistentID);
        } catch (DiskAccessException e) {
          logger.warn("Unable to persist membership change", e);
        }
      } else {
        if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
          logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE,
              "{}-{}: Not marking member online in persistent view because we're still in initialization",
              shortDiskStoreId(), regionPath);
        }
      }

      notifyListenersMemberOnline(distributedMember, persistentID);
    }
  }

  private void memberRevoked(PersistentMemberPattern pattern) {
    // Persist the revoked member, so if we recover later we will remember that they were revoked.
    persistentMemberView.memberRevoked(pattern);

    // Remove the revoked member from our view.
    for (PersistentMemberID id : persistentMemberView.getOfflineMembers()) {
      if (pattern.matches(id)) {
        memberRemoved(id, true);
      }
    }
    for (PersistentMemberID id : persistentMemberView.getOnlineMembers()) {
      if (pattern.matches(id)) {
        memberRemoved(id, true);
      }
    }
    for (PersistentMemberID id : persistentMemberView.getOfflineAndEqualMembers()) {
      if (pattern.matches(id)) {
        memberRemoved(id, true);
      }
    }
  }

  @VisibleForTesting
  void memberRemoved(PersistentMemberID id, boolean revoked) {
    if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) {
      logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE, "{}-{}: Member removed. persistentID={}",
          shortDiskStoreId(), regionPath, id);
    }

    synchronized (lock) {
      recoveredMembers.remove(id);
      equalMembers.remove(id);
      if (!online) {
        removedMembers.add(id);
      }
      try {
        if (PersistenceObserverHolder.getInstance().memberRemoved(regionPath, id)) {
          persistentMemberView.memberRemoved(id);
        }

        // Purge any IDs that are old versions of the the id that we just removed
        for (PersistentMemberID persistedId : getPersistedMembers()) {
          if (persistedId.isOlderOrEqualVersionOf(id)) {
            persistentMemberView.memberRemoved(persistedId);
          }
        }
        PersistenceObserverHolder.getInstance().afterRemovePersisted(regionPath, id);
      } catch (DiskAccessException e) {
        logger.warn("Unable to persist membership change", e);
      }
      notifyListenersMemberRemoved(id, revoked);
    }
  }

  @Override
  public PersistentMemberID getPersistentID() {
    return persistentMemberView.getMyPersistentID();
  }

  @Override
  public PersistentMemberID getInitializingID() {
    return persistentMemberView.getMyInitializingID();
  }

  @Override
  public void addListener(PersistentStateListener listener) {
    synchronized (this) {
      Set<PersistentStateListener> tmpListeners = new HashSet<>(persistentStateListeners);
      tmpListeners.add(listener);
      persistentStateListeners = Collections.unmodifiableSet(tmpListeners);
    }
  }

  @Override
  public void removeListener(PersistentStateListener listener) {
    synchronized (this) {
      if (!persistentStateListeners.contains(listener)) {
        return;
      }
      Set<PersistentStateListener> tmpListeners = new HashSet<>(persistentStateListeners);
      tmpListeners.remove(listener);
      persistentStateListeners = Collections.unmodifiableSet(tmpListeners);
    }
  }

  @TestOnly
  Set<PersistentStateListener> getPersistentStateListenerSet() {
    return persistentStateListeners;
  }

  private void notifyListenersMemberOnline(InternalDistributedMember member,
      PersistentMemberID persistentID) {
    for (PersistentStateListener listener : persistentStateListeners) {
      listener.memberOnline(member, persistentID);
    }
  }

  private void notifyListenersMemberOffline(InternalDistributedMember member,
      PersistentMemberID persistentID) {
