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 { InitialImageAdvice advice = cacheDistributionAdvisor.adviseInitialImage(previousAdvice, true); if (hasReplicates(advice)) { if (hasDiskImageToRecoverFrom) { removeReplicatesIfWeAreEqualToAnyOrElseClearEqualMembers(advice.getReplicates()); } return advice; } if (hasNonPersistentMember(advice)) { updateMembershipViewFromAnyPeer(advice.getNonPersistent(), hasDiskImageToRecoverFrom); } 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 (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);   }
@Test public void clearsEqualMembers_ifHasDiskImageAndAllReplicatesAreUnequal() { persistenceInitialImageAdvisor = persistenceInitialImageAdvisorWithDiskImage(); InitialImageAdvice adviceFromCacheDistributionAdvisor = adviceWithReplicates(3); when(cacheDistributionAdvisor.adviseInitialImage(isNull(), anyBoolean())) .thenReturn(adviceFromCacheDistributionAdvisor); persistenceInitialImageAdvisor.getAdvice(null); verify(persistenceAdvisor, times(1)) .clearEqualMembers(); }
@Test public void adviceIncludesAllReplicates_ifHasDiskImageAndAllReplicatesAreUnequal() { persistenceInitialImageAdvisor = persistenceInitialImageAdvisorWithDiskImage(); InitialImageAdvice adviceFromCacheDistributionAdvisor = adviceWithReplicates(3); when(cacheDistributionAdvisor.adviseInitialImage(isNull(), anyBoolean())) .thenReturn(adviceFromCacheDistributionAdvisor); InitialImageAdvice result = persistenceInitialImageAdvisor.getAdvice(null); assertThat(result.getReplicates()) .isEqualTo(adviceFromCacheDistributionAdvisor.getReplicates()); }
@Test public void adviceIncludesNoReplicates_ifHasDiskImageAndAnyReplicateIsEqual() { persistenceInitialImageAdvisor = persistenceInitialImageAdvisorWithDiskImage(); when(cacheDistributionAdvisor.adviseInitialImage(isNull(), anyBoolean())) .thenReturn(adviceWithReplicates(9)); when(persistenceAdvisor.checkMyStateOnMembers(any())) .thenReturn(true); InitialImageAdvice result = persistenceInitialImageAdvisor.getAdvice(null); assertThat(result.getReplicates()) .isEmpty(); }
@Test public void obtainsFreshAdvice_ifAdviceIncludesNoReplicatesAndPreviousAdviceHasReplicates() { InitialImageAdvice previousAdviceWithReplicates = adviceWithReplicates(1); persistenceInitialImageAdvisor = persistenceInitialImageAdvisorWithDiskImage(); when(cacheDistributionAdvisor.adviseInitialImage(isNotNull(), anyBoolean())) .thenReturn(adviceWithReplicates(0)); persistenceInitialImageAdvisor.getAdvice(previousAdviceWithReplicates); InOrder inOrder = inOrder(cacheDistributionAdvisor); inOrder.verify(cacheDistributionAdvisor, times(1)) .adviseInitialImage(same(previousAdviceWithReplicates), anyBoolean()); inOrder.verify(cacheDistributionAdvisor, times(1)) .adviseInitialImage(isNull(), anyBoolean()); inOrder.verify(cacheDistributionAdvisor, times(0)) .adviseInitialImage(any(), anyBoolean()); }
@Test public void attemptsToUpdateMembershipViewFromEachNonPersistentReplicate_ifAllReplicatesAreNonPersistent() { persistenceInitialImageAdvisor = persistenceInitialImageAdvisorWithDiskImage(); InitialImageAdvice adviceWithNonPersistentReplicates = adviceWithNonPersistentReplicates(4); when(cacheDistributionAdvisor.adviseInitialImage(isNull(), anyBoolean())) .thenReturn(adviceWithNonPersistentReplicates, adviceWithReplicates(1)); doThrow(ReplyException.class) .when(persistenceAdvisor).updateMembershipView(any(), anyBoolean()); persistenceInitialImageAdvisor.getAdvice(null); for (InternalDistributedMember peer : adviceWithNonPersistentReplicates.getNonPersistent()) { verify(persistenceAdvisor, times(1)) .updateMembershipView(peer, true); } }
@Test public void updatesMembershipViewFromFirstNonPersistentReplicateThatReplies() { persistenceInitialImageAdvisor = persistenceInitialImageAdvisorWithDiskImage(); InitialImageAdvice adviceWithNonPersistentReplicates = adviceWithNonPersistentReplicates(4); when(cacheDistributionAdvisor.adviseInitialImage(isNull(), anyBoolean())) .thenReturn(adviceWithNonPersistentReplicates, adviceWithReplicates(1)); doThrow(ReplyException.class) .doNothing() .when(persistenceAdvisor).updateMembershipView(any(), anyBoolean()); persistenceInitialImageAdvisor.getAdvice(null); verify(persistenceAdvisor, times(2)) .updateMembershipView(any(), anyBoolean()); }
@Test public void propagatesException_ifCancelInProgress() { persistenceInitialImageAdvisor = persistenceInitialImageAdvisorWithDiskImage(); CancelCriterion cancelCriterion = mock(CancelCriterion.class); when(cacheDistributionAdvisor.getAdvisee().getCancelCriterion()) .thenReturn(cancelCriterion); doThrow(new CacheClosedException("test")) .when(cancelCriterion).checkCancelInProgress(any()); Throwable thrown = catchThrowable(() -> persistenceInitialImageAdvisor.getAdvice(null)); assertThat(thrown) .isInstanceOf(CacheClosedException.class); }
@Test public void returnsAdviceFromCacheDistributionAdvisor_ifNoOnlineOrPreviouslyOnlineMembers() { persistenceInitialImageAdvisor = persistenceInitialImageAdvisorWithDiskImage(); InitialImageAdvice adviceFromCacheDistributionAdvisor = new InitialImageAdvice(); when(cacheDistributionAdvisor.adviseInitialImage(isNull(), anyBoolean())) .thenReturn(adviceFromCacheDistributionAdvisor); when(persistenceAdvisor.getPersistedOnlineOrEqualMembers()) .thenReturn(new HashSet<>()); InitialImageAdvice result = persistenceInitialImageAdvisor.getAdvice(null); assertThat(result) .isSameAs(adviceFromCacheDistributionAdvisor); }
@Test public void adviceIncludesReplicatesThatAppearWhileAcquiringTieLock() { persistenceInitialImageAdvisor = persistenceInitialImageAdvisorWithDiskImage(); InitialImageAdvice adviceBeforeAcquiringTieLock = new InitialImageAdvice(); InitialImageAdvice adviceAfterAcquiringTieLock = adviceWithReplicates(4); when(persistenceAdvisor.getPersistedOnlineOrEqualMembers()) .thenReturn(persistentMemberIDs(1)); when(persistenceAdvisor.getMembersToWaitFor(any(), any())) .thenReturn(emptySet()); when(persistenceAdvisor.acquireTieLock()) .thenReturn(true); when(cacheDistributionAdvisor.adviseInitialImage(any(), anyBoolean())) .thenReturn(adviceBeforeAcquiringTieLock, adviceAfterAcquiringTieLock); InitialImageAdvice result = persistenceInitialImageAdvisor.getAdvice(null); assertThat(result.getReplicates()) .isEqualTo(adviceAfterAcquiringTieLock.getReplicates()); }
@Test public void propagatesException_ifIncompatibleWithReplicateThatAppearsWhileAcquiringTieLock() { persistenceInitialImageAdvisor = persistenceInitialImageAdvisorWithDiskImage(); InitialImageAdvice adviceBeforeAcquiringTieLock = new InitialImageAdvice(); InitialImageAdvice adviceAfterAcquiringTieLock = adviceWithReplicates(4); when(persistenceAdvisor.getPersistedOnlineOrEqualMembers()) .thenReturn(persistentMemberIDs(1)); when(persistenceAdvisor.getMembersToWaitFor(any(), any())) .thenReturn(emptySet()); when(persistenceAdvisor.acquireTieLock()) .thenReturn(true); when(cacheDistributionAdvisor.adviseInitialImage(any(), anyBoolean())) .thenReturn(adviceBeforeAcquiringTieLock, adviceAfterAcquiringTieLock); doThrow(ConflictingPersistentDataException.class) .when(persistenceAdvisor).checkMyStateOnMembers(any()); Throwable thrown = catchThrowable(() -> persistenceInitialImageAdvisor.getAdvice(null)); assertThat(thrown) .isInstanceOf(ConflictingPersistentDataException.class); }
@Test public void announcesProgressToPersistenceAdvisor_whenWaitingForMissingMembers() { persistenceInitialImageAdvisor = new PersistenceInitialImageAdvisor(persistenceAdvisor, "short disk store ID", "region path", cacheDistributionAdvisor, true, advisor -> new MembershipChangeListenerFactory() .setWarningDelay(ofSeconds(0)) .setPollDuration(ofSeconds(1)) .setCancelCondition(cancelCondition(advisor, mock(CancelCriterion.class))) .setWarning(advisor::logWaitingForMembers) .create()); HashSet<PersistentMemberID> offlineMembersToWaitFor = memberIDs("offline member", 1); Set<PersistentMemberID> membersToWaitFor = new HashSet<>(offlineMembersToWaitFor); when(persistenceAdvisor.getPersistedOnlineOrEqualMembers()) .thenReturn(offlineMembersToWaitFor); when(persistenceAdvisor.getMembersToWaitFor(any(), any())) .thenAnswer(invocation -> { Set<PersistentMemberID> offlineMembers = invocation.getArgument(1); offlineMembers.addAll(offlineMembersToWaitFor); return membersToWaitFor; }); when(cacheDistributionAdvisor.adviseInitialImage(null, true)) .thenReturn(adviceWithReplicates(0), adviceWithReplicates(1)); persistenceInitialImageAdvisor.getAdvice(null); InOrder inOrder = inOrder(persistenceAdvisor); inOrder.verify(persistenceAdvisor, times(1)) .beginWaitingForMembershipChange(membersToWaitFor); inOrder.verify(persistenceAdvisor, times(1)) .setWaitingOnMembers(isNotNull(), eq(offlineMembersToWaitFor)); inOrder.verify(persistenceAdvisor, times(1)) .endWaitingForMembershipChange(); inOrder.verify(persistenceAdvisor, times(1)) .setWaitingOnMembers(isNull(), isNull()); inOrder.verify(persistenceAdvisor, times(0)) .setWaitingOnMembers(any(), any()); }