| /* |
| * 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 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.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.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.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.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; |
| |
| protected 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; |
| |
| public PersistenceAdvisorImpl(CacheDistributionAdvisor cacheDistributionAdvisor, |
| DistributedLockService distributedLockService, PersistentMemberView persistentMemberView, |
| String regionPath, DiskRegionStats diskRegionStats, |
| PersistentMemberManager persistentMemberManager) { |
| this(cacheDistributionAdvisor, distributedLockService, persistentMemberView, regionPath, |
| diskRegionStats, persistentMemberManager, new PersistentStateQueryMessageSenderFactory()); |
| } |
| |
| @VisibleForTesting |
| PersistenceAdvisorImpl(CacheDistributionAdvisor cacheDistributionAdvisor, |
| DistributedLockService distributedLockService, PersistentMemberView persistentMemberView, |
| String regionPath, DiskRegionStats diskRegionStats, |
| PersistentMemberManager persistentMemberManager, |
| PersistentStateQueryMessageSenderFactory persistentStateQueryMessageSenderFactory) { |
| this.cacheDistributionAdvisor = cacheDistributionAdvisor; |
| this.distributedLockService = distributedLockService; |
| this.regionPath = regionPath; |
| this.persistentMemberView = persistentMemberView; |
| this.diskRegionStats = diskRegionStats; |
| profileChangeListener = new ProfileChangeListener(); |
| this.persistentMemberManager = persistentMemberManager; |
| this.persistentStateQueryMessageSenderFactory = persistentStateQueryMessageSenderFactory; |
| |
| // 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<>(); |
| TransformUtils.transform(members, onlineMembers, |
| TransformUtils.persistentMemberIdToLogEntryTransformer); |
| |
| logger.info( |
| "The following persistent member has gone offline for region {}: {} Remaining participating members for the region include: {}", |
| new Object[] {regionPath, |
| TransformUtils.persistentMemberIdToLogEntryTransformer |
| .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 { |
| // Fix for 41100 - 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()); |
| |
| // Another fix for 41100 - 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. |
| * |
| * This method should be called after we have decided that there is no conflicting persistent |
| * exception. |
| * |
| * Fix for bug 44045. |
| */ |
| 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); |
| |
| persistenceAdvisorObserver.observe(regionPath); |
| |
| boolean equal = false; |
| for (Map.Entry<InternalDistributedMember, PersistentMemberState> entry : remoteStates |
| .getStateOnPeers() |
| .entrySet()) { |
| InternalDistributedMember member = entry.getKey(); |
| PersistentMemberID remoteId = remoteStates.getPersistentIds().get(member); |
| |
| final PersistentMemberID myId = getPersistentID(); |
| 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); |
| throw new ConflictingPersistentDataException(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; |
| } |
| |
| public static void setPersistenceAdvisorObserver(PersistenceAdvisorObserver o) { |
| persistenceAdvisorObserver = o == null ? DEFAULT_PERSISTENCE_ADVISOR_OBSERVER : o; |
| } |
| |
| @Override |
| public PersistentMemberID getPersistentIDIfOnline() { |
| if (online) { |
| return persistentMemberView.getMyPersistentID(); |
| } else { |
| 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); |
| } |
| } |
| } |
| |
| 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) { |
| Set<PersistentStateListener> tmpListeners = new HashSet<>(persistentStateListeners); |
| tmpListeners.remove(listener); |
| persistentStateListeners = Collections.unmodifiableSet(tmpListeners); |
| } |
| } |
| |
| private void notifyListenersMemberOnline(InternalDistributedMember member, |
| PersistentMemberID persistentID) { |
| for (PersistentStateListener listener : persistentStateListeners) { |
| listener.memberOnline(member, persistentID); |
| } |
| } |
| |
| private void notifyListenersMemberOffline(InternalDistributedMember member, |
| PersistentMemberID persistentID) { |
| for (PersistentStateListener listener : persistentStateListeners) { |
| listener.memberOffline(member, persistentID); |
| } |
| } |
| |
| private void notifyListenersMemberRemoved(PersistentMemberID persistentID, boolean revoked) { |
| for (PersistentStateListener listener : persistentStateListeners) { |
| listener.memberRemoved(persistentID, revoked); |
| } |
| } |
| |
| @Override |
| public HashSet<PersistentMemberID> getPersistedOnlineOrEqualMembers() { |
| HashSet<PersistentMemberID> members = new HashSet<>(persistentMemberView.getOnlineMembers()); |
| members.addAll(equalMembers); |
| return members; |
| } |
| |
| @Override |
| public void prepareNewMember(InternalDistributedMember sender, PersistentMemberID oldId, |
| PersistentMemberID newId) { |
| if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) { |
| logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE, |
| "{}-{}: Preparing new persistent id {}. Old id is {}", shortDiskStoreId(), regionPath, |
| newId, oldId); |
| } |
| synchronized (lock) { |
| // Don't prepare the ID if the advisor doesn't have a profile. This prevents a race with the |
| // advisor remove |
| if (!cacheDistributionAdvisor.containsId(sender)) { |
| if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) { |
| logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE, |
| "{}-{}: Refusing to prepare id because {} is not in our advisor", shortDiskStoreId(), |
| regionPath, sender); |
| } |
| return; |
| } |
| // Persist new members even if we are not online yet. Two members can become online at once. |
| // This way, they will know about each other. |
| persistentMemberView.memberOnline(newId); |
| |
| // The oldId and newId could be the same if the member is retrying a GII. See bug #42051 |
| if (oldId != null && !oldId.equals(newId)) { |
| memberRemoved(oldId, false); |
| } |
| } |
| } |
| |
| protected String shortDiskStoreId() { |
| DiskStoreID diskStoreID = getDiskStoreID(); |
| return diskStoreID == null ? "mem" : diskStoreID.abbrev(); |
| } |
| |
| @Override |
| public void removeMember(PersistentMemberID id) { |
| memberRemoved(id, false); |
| } |
| |
| @Override |
| public void markMemberOffline(InternalDistributedMember member, PersistentMemberID id) { |
| memberOffline(member, id); |
| } |
| |
| @Override |
| public CacheDistributionAdvisor getCacheDistributionAdvisor() { |
| return cacheDistributionAdvisor; |
| } |
| |
| @Override |
| public void setWaitingOnMembers(Set<PersistentMemberID> allMembersToWaitFor, |
| Set<PersistentMemberID> offlineMembersToWaitFor) { |
| allMembersWaitingFor = allMembersToWaitFor; |
| offlineMembersWaitingFor = offlineMembersToWaitFor; |
| } |
| |
| @Override |
| public boolean isClosed() { |
| return isClosed; |
| } |
| |
| |
| public void finishPendingDestroy() { |
| // send a message to peers indicating that they should remove this profile |
| long viewVersion = cacheDistributionAdvisor.startOperation(); |
| try { |
| RemovePersistentMemberMessage.send(cacheDistributionAdvisor.adviseProfileUpdate(), |
| cacheDistributionAdvisor.getDistributionManager(), regionPath, getPersistentID(), |
| getInitializingID()); |
| |
| persistentMemberView.finishPendingDestroy(); |
| } finally { |
| if (viewVersion != -1) { |
| cacheDistributionAdvisor.endOperation(viewVersion); |
| } |
| } |
| synchronized (lock) { |
| recoveredMembers.clear(); |
| recoveredMembers.addAll(getPersistedMembers()); |
| } |
| } |
| |
| /** |
| * Returns the member id of the member who has the latest copy of the persistent region. This may |
| * be the local member ID if this member has the latest known copy. |
| * |
| * This method will block until the latest member is online. |
| * |
| * @throws ConflictingPersistentDataException if there are active members which are not based on |
| * the state that is persisted in this member. |
| */ |
| @Override |
| public InitialImageAdvice getInitialImageAdvice(InitialImageAdvice previousAdvice, |
| boolean hasDiskImageToRecoverFrom) { |
| PersistenceInitialImageAdvisor piia = new PersistenceInitialImageAdvisor(this, |
| shortDiskStoreId(), regionPath, cacheDistributionAdvisor, hasDiskImageToRecoverFrom); |
| return piia.getAdvice(previousAdvice); |
| } |
| |
| /** |
| * @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. |
| */ |
| @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; |
| } |
| |
| /** |
| * In the event that the region was partially destroyed via DestroyDataStorage on the peer, |
| * we do not need to wait on that peer. Currently this state can be reached when a bucket region |
| * GII fails, which results in DestroyDataStorage on the region (as opposed to a DestroyRegion). |
| * See DiskRegion.destroyPartiallyInitializedRegion() which handles the failed GII on the image |
| * receiving side for more details. |
| */ |
| private void handlePartiallyDestroyedRegion(final Set<PersistentMemberID> offlineMembers, |
| final Set<PersistentMemberID> membersToWaitFor, |
| final PersistentMemberID persistentID, |
| final PersistentMemberID initializingID, |
| final DiskStoreID diskStoreID) { |
| /* |
| * When DestroyDataStorage is invoked on a peer for this region, we expect that its |
| * initializing and persistent IDs will be null, but the disk store ID will be non-null |
| * because the region was not fully destroyed. |
| */ |
| if (initializingID == null && persistentID == null & diskStoreID != null) { |
| removeByDiskStoreID(membersToWaitFor, diskStoreID, true); |
| removeByDiskStoreID(offlineMembers, diskStoreID, true); |
| } |
| } |
| |
| /** |
| * Given a set of persistent members, if the same member occurs more than once in the set but |
| * with different timestamps, remove the older ones leaving only the most recent. |
| * |
| * @param persistentMemberSet The set of persistent members, possibly modified by this method. |
| */ |
| protected void removeOlderMembers(Set<PersistentMemberID> persistentMemberSet) { |
| Map<DiskStoreID, PersistentMemberID> mostRecentMap = new HashMap<>(); |
| List<PersistentMemberID> idsToRemove = new ArrayList<>(); |
| for (PersistentMemberID persistentMember : persistentMemberSet) { |
| DiskStoreID diskStoreId = persistentMember.getDiskStoreId(); |
| PersistentMemberID mostRecent = mostRecentMap.get(diskStoreId); |
| if (mostRecent == null) { |
| mostRecentMap.put(diskStoreId, persistentMember); |
| } else { |
| PersistentMemberID older = persistentMember; |
| boolean persistentMemberIsNewer = |
| !persistentMember.isOlderOrEqualVersionOf(mostRecent); |
| if (persistentMemberIsNewer) { |
| older = mostRecent; |
| mostRecentMap.put(diskStoreId, persistentMember); |
| } |
| idsToRemove.add(older); |
| } |
| } |
| persistentMemberSet.removeAll(idsToRemove); |
| } |
| |
| /** |
| * Remove all members with a given disk store id from the set of members to wait for, who is newer |
| * than the real one. The reason is: A is waiting for B2, but B sends B1<=A to A. That means A |
| * knows more than B in both B1 and B2. B itself knows nothing about B2. So we don't need to wait |
| * for B2 (since we don't need to wait for B1). |
| */ |
| private void removeNewerPersistentID(Set<PersistentMemberID> membersToWaitFor, |
| PersistentMemberID persistentID) { |
| for (Iterator<PersistentMemberID> itr = membersToWaitFor.iterator(); itr.hasNext();) { |
| PersistentMemberID id = itr.next(); |
| if (persistentID.isOlderOrEqualVersionOf(id)) { |
| if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) { |
| logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE, |
| "{}-{}: Not waiting for {} because local member knows more about it", |
| shortDiskStoreId(), regionPath, id); |
| } |
| itr.remove(); |
| } |
| } |
| } |
| |
| /** |
| * Remove all members with a given disk store id from the set of members to wait for. |
| */ |
| private void removeByDiskStoreID(Set<PersistentMemberID> membersToWaitFor, |
| DiskStoreID diskStoreID, boolean updateAdvisor) { |
| for (Iterator<PersistentMemberID> itr = membersToWaitFor.iterator(); itr.hasNext();) { |
| PersistentMemberID id = itr.next(); |
| if (id.getDiskStoreId().equals(diskStoreID)) { |
| if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) { |
| logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE, |
| "{}-{}: Not waiting for {} because it no longer has this region in its disk store", |
| shortDiskStoreId(), regionPath, id); |
| } |
| itr.remove(); |
| if (updateAdvisor) { |
| memberRemoved(id, false); |
| } |
| } |
| } |
| } |
| |
| @Override |
| public boolean wasHosting() { |
| return getPersistentID() != null || getInitializingID() != null; |
| } |
| |
| protected String getRegionPathForOfflineMembers() { |
| return regionPath; |
| } |
| |
| /** |
| * Returns the set of missing members that we report back to the any admin DS looking for missing |
| * members. |
| */ |
| protected Set<PersistentMemberID> getMissingMembers() { |
| return offlineMembersWaitingFor; |
| } |
| |
| /** |
| * Returns the set of missing members that we report back to the any admin DS looking for missing |
| * members. |
| */ |
| public Set<PersistentMemberID> getAllMembersToWaitFor() { |
| return allMembersWaitingFor; |
| } |
| |
| @Override |
| public void logWaitingForMembers() { |
| Set<String> membersToWaitForLogEntries = new HashSet<>(); |
| |
| if (offlineMembersWaitingFor != null && !offlineMembersWaitingFor.isEmpty()) { |
| TransformUtils.transform(offlineMembersWaitingFor, membersToWaitForLogEntries, |
| TransformUtils.persistentMemberIdToLogEntryTransformer); |
| |
| StartupStatus.startup( |
| String.format( |
| "Region %s has potentially stale data. It is waiting for another member to recover the latest data.My persistent id:%sMembers with potentially new data:%sUse the gfsh show missing-disk-stores command to see all disk stores that are being waited on by other members.", |
| regionPath, |
| TransformUtils.persistentMemberIdToLogEntryTransformer.transform(getPersistentID()), |
| membersToWaitForLogEntries)); |
| } else { |
| TransformUtils.transform(allMembersWaitingFor, membersToWaitForLogEntries, |
| TransformUtils.persistentMemberIdToLogEntryTransformer); |
| |
| StartupStatus.startup( |
| String.format( |
| "Region %s has potentially stale data. It is waiting for another online member to recover the latest data.My persistent id:%sMembers with potentially new data:%sUse the gfsh show missing-disk-stores command to see all disk stores that are being waited on by other members.", |
| regionPath, |
| TransformUtils.persistentMemberIdToLogEntryTransformer.transform(getPersistentID()), |
| membersToWaitForLogEntries)); |
| } |
| } |
| |
| @Override |
| public void clearEqualMembers() { |
| synchronized (lock) { |
| equalMembers.clear(); |
| } |
| } |
| |
| @Override |
| public void checkInterruptedByShutdownAll() {} |
| |
| @Override |
| public void close() { |
| isClosed = true; |
| persistentMemberManager.removeRevocationListener(profileChangeListener); |
| cacheDistributionAdvisor.removeProfileChangeListener(profileChangeListener); |
| releaseTieLock(); |
| } |
| |
| |
| /** |
| * Try to acquire the distributed lock which members must grab for in the case of a tie. Whoever |
| * gets the lock initializes first. |
| */ |
| @Override |
| public boolean acquireTieLock() { |
| // We're tied for the latest copy of the data. try to get the distributed lock. |
| holdingTieLock = distributedLockService.lock("PERSISTENCE_" + regionPath, 0, -1); |
| if (!holdingTieLock) { |
| if (logger.isDebugEnabled(LogMarker.PERSIST_ADVISOR_VERBOSE)) { |
| logger.debug(LogMarker.PERSIST_ADVISOR_VERBOSE, "{}-{}: Failed to acquire the lock.", |
| shortDiskStoreId(), regionPath); |
| } |
| } |
| return holdingTieLock; |
| } |
| |
| @Override |
| public void releaseTieLock() { |
| if (holdingTieLock) { |
| distributedLockService.unlock("PERSISTENCE_" + regionPath); |
| holdingTieLock = false; |
| } |
| } |
| |
| private boolean wasAboutToDestroy() { |
| return persistentMemberView.wasAboutToDestroy() |
| || persistentMemberView.wasAboutToDestroyDataStorage(); |
| } |
| |
| protected synchronized void resetState() { |
| online = false; |
| removedMembers.clear(); |
| } |
| |
| public void flushMembershipChanges() { |
| try { |
| cacheDistributionAdvisor.waitForCurrentOperations(); |
| } catch (RegionDestroyedException ignored) { |
| } |
| } |
| |
| @Override |
| public void persistMembersOfflineAndEqual( |
| Map<InternalDistributedMember, PersistentMemberID> map) { |
| for (PersistentMemberID persistentID : map.values()) { |
| persistentMemberView.memberOfflineAndEqual(persistentID); |
| } |
| } |
| |
| @Override |
| public DiskStoreID getDiskStoreID() { |
| return persistentMemberView.getDiskStoreID(); |
| } |
| |
| @Override |
| public boolean isOnline() { |
| return online; |
| } |
| |
| public interface PersistenceAdvisorObserver { |
| void observe(String regionPath); |
| } |
| |
| private class ProfileChangeListener implements ProfileListener, MemberRevocationListener { |
| |
| @Override |
| public void profileCreated(Profile profile) { |
| profileUpdated(profile); |
| } |
| |
| @Override |
| public void profileRemoved(Profile profile, boolean destroyed) { |
| CacheProfile cp = (CacheProfile) profile; |
| if (cp.persistentID != null) { |
| if (destroyed) { |
| memberRemoved(cp.persistentID, false); |
| } else { |
| memberOffline(profile.getDistributedMember(), cp.persistentID); |
| } |
| } |
| } |
| |
| @Override |
| public void profileUpdated(Profile profile) { |
| CacheProfile cp = (CacheProfile) profile; |
| if (cp.persistentID != null && cp.persistenceInitialized) { |
| memberOnline(profile.getDistributedMember(), cp.persistentID); |
| } |
| } |
| |
| @Override |
| public void revoked(PersistentMemberPattern pattern) { |
| memberRevoked(pattern); |
| } |
| |
| @Override |
| public Set<PersistentMemberID> getMissingMemberIds() { |
| return getMissingMembers(); |
| } |
| |
| @Override |
| public String getRegionPath() { |
| return getRegionPathForOfflineMembers(); |
| } |
| |
| @Override |
| public boolean matches(PersistentMemberPattern pattern) { |
| return pattern.matches(getPersistentID()) || pattern.matches(getInitializingID()); |
| } |
| } |
| } |
