PRHARedundancyProvider { void startRedundancyRecovery() { partitionedRegion.getRegionAdvisor().addMembershipListener(new PRMembershipListener()); scheduleRedundancyRecovery(null); } PRHARedundancyProvider(PartitionedRegion partitionedRegion,
      InternalResourceManager resourceManager); @VisibleForTesting  PRHARedundancyProvider(PartitionedRegion partitionedRegion,
      InternalResourceManager resourceManager,
      BiFunction<PRHARedundancyProvider, Integer, PersistentBucketRecoverer> persistentBucketRecovererFunction); @VisibleForTesting  PRHARedundancyProvider(PartitionedRegion partitionedRegion,
      InternalResourceManager resourceManager,
      BiFunction<PRHARedundancyProvider, Integer, PersistentBucketRecoverer> persistentBucketRecovererFunction,
      PartitionedRegionRebalanceOpFactory rebalanceOpFactory,
      CompletableFuture<Void> providerStartupTask); static void timedOut(PartitionedRegion partitionedRegion,
      Set<InternalDistributedMember> allStores,
      Collection<InternalDistributedMember> alreadyUsed, String opString, long timeOut); PartitionedRegion getPartitionedRegion(); InternalDistributedMember createBucketAtomically(int bucketId, int newBucketSize,
      boolean finishIncompleteCreation, String partitionName); void endBucketCreationLocally(int bucketId, InternalDistributedMember newPrimary); void finishIncompleteBucketCreation(int bucketId); boolean createBackupBucketOnMember(int bucketId, InternalDistributedMember targetMember,
      boolean isRebalance, boolean replaceOfflineData, InternalDistributedMember fromMember,
      boolean forceCreation); boolean isRedundancyImpaired(); void shutdown(); InternalPRInfo buildPartitionedRegionInfo(boolean internal, LoadProbe loadProbe); OfflineMemberDetailsImpl fetchOfflineMembers(); InternalPartitionDetails buildPartitionMemberDetails(boolean internal,
      LoadProbe loadProbe); boolean isPersistentRecoveryComplete(); static final String TIMEOUT_MSG; static final String INSUFFICIENT_STORES_MSG; }
@Test public void reportsStartupTaskToResourceManager() { @SuppressWarnings("unchecked") CompletableFuture<Void> providerStartupTask = mock(CompletableFuture.class); when(partitionedRegion.getRegionAdvisor()).thenReturn(mock(RegionAdvisor.class)); when(partitionedRegion.getPartitionAttributes()).thenReturn(mock(PartitionAttributes.class)); when(partitionedRegion.isDataStore()).thenReturn(true); when(resourceManager.getExecutor()).thenReturn(mock(ScheduledExecutorService.class)); prHaRedundancyProvider = new PRHARedundancyProvider(partitionedRegion, resourceManager, (a, b) -> mock(PersistentBucketRecoverer.class), PRHARedundancyProviderTest::createRebalanceOp, providerStartupTask); prHaRedundancyProvider.startRedundancyRecovery(); verify(resourceManager).addStartupTask(same(providerStartupTask)); }
@Test public void doNotNotReportStartupTaskIfRecoveryDelayIsNegative() { when(partitionedRegion.getRegionAdvisor()).thenReturn(mock(RegionAdvisor.class)); PartitionAttributes partitionAttributes = mock(PartitionAttributes.class); when(partitionAttributes.getStartupRecoveryDelay()).thenReturn(-1L); when(partitionedRegion.getPartitionAttributes()).thenReturn(partitionAttributes); when(resourceManager.getExecutor()).thenReturn(mock(ScheduledExecutorService.class)); prHaRedundancyProvider = new PRHARedundancyProvider(partitionedRegion, resourceManager, (a, b) -> mock(PersistentBucketRecoverer.class), PRHARedundancyProviderTest::createRebalanceOp, null); prHaRedundancyProvider.startRedundancyRecovery(); verify(resourceManager, never()).addStartupTask(any()); }
@Test public void doNotNotReportStartupTaskIfPartitionIsNotDataStore() { when(partitionedRegion.getRegionAdvisor()).thenReturn(mock(RegionAdvisor.class)); PartitionAttributes partitionAttributes = mock(PartitionAttributes.class); when(partitionedRegion.getPartitionAttributes()).thenReturn(partitionAttributes); when(partitionedRegion.isDataStore()).thenReturn(false); when(resourceManager.getExecutor()).thenReturn(mock(ScheduledExecutorService.class)); prHaRedundancyProvider = new PRHARedundancyProvider(partitionedRegion, resourceManager, (a, b) -> mock(PersistentBucketRecoverer.class), PRHARedundancyProviderTest::createRebalanceOp, null); prHaRedundancyProvider.startRedundancyRecovery(); verify(resourceManager, never()).addStartupTask(any()); }
@Test public void doNotNotReportStartupTaskIfExecutorRejectsRebalanceTask() { when(partitionedRegion.getRegionAdvisor()).thenReturn(mock(RegionAdvisor.class)); PartitionAttributes partitionAttributes = mock(PartitionAttributes.class); when(partitionedRegion.getPartitionAttributes()).thenReturn(partitionAttributes); when(partitionedRegion.isDataStore()).thenReturn(true); ScheduledExecutorService executor = mock(ScheduledExecutorService.class); when(executor.schedule(any(Runnable.class), anyLong(), any())) .thenThrow(new RejectedExecutionException("Rejected for the test")); when(resourceManager.getExecutor()).thenReturn(executor); prHaRedundancyProvider = new PRHARedundancyProvider(partitionedRegion, resourceManager, (a, b) -> mock(PersistentBucketRecoverer.class), PRHARedundancyProviderTest::createRebalanceOp, null); prHaRedundancyProvider.startRedundancyRecovery(); verify(resourceManager, never()).addStartupTask(any()); }
@Test public void completesStartupTaskWhenRedundancyRecovered() { DistributedSystem distributedSystem = mock(DistributedSystem.class); when(distributedSystem.getCancelCriterion()).thenReturn(mock(CancelCriterion.class)); InternalCache cache = mock(InternalCache.class); when(cache.getDistributedSystem()).thenReturn(distributedSystem); when(partitionedRegion.getGemFireCache()).thenReturn(cache); when(partitionedRegion.getRegionAdvisor()).thenReturn(mock(RegionAdvisor.class)); when(partitionedRegion.getPartitionAttributes()).thenReturn(mock(PartitionAttributes.class)); when(partitionedRegion.isDataStore()).thenReturn(true); when(partitionedRegion.getPrStats()).thenReturn(mock(PartitionedRegionStats.class)); ScheduledExecutorService executorService = mock(ScheduledExecutorService.class); when(resourceManager.getExecutor()).thenReturn(executorService); when(executorService.schedule(any(Runnable.class), anyLong(), any())) .thenAnswer(runTheRunnable()); @SuppressWarnings("unchecked") CompletableFuture<Void> providerStartupTask = mock(CompletableFuture.class); prHaRedundancyProvider = new PRHARedundancyProvider(partitionedRegion, resourceManager, (a, b) -> mock(PersistentBucketRecoverer.class), PRHARedundancyProviderTest::createRebalanceOp, providerStartupTask); prHaRedundancyProvider.startRedundancyRecovery(); verify(providerStartupTask).complete(any()); }
@Test @Parameters({"RUNTIME", "CANCEL", "REGION_DESTROYED"}) @TestCaseName("{method}[{index}]: {params}") public void startTaskCompletesExceptionallyIfExceptionIsThrown( ExceptionToThrow exceptionToThrow) { DistributedSystem distributedSystem = mock(DistributedSystem.class); when(distributedSystem.getCancelCriterion()).thenReturn(mock(CancelCriterion.class)); InternalCache cache = mock(InternalCache.class); when(cache.getDistributedSystem()).thenReturn(distributedSystem); when(partitionedRegion.getGemFireCache()).thenReturn(cache); when(partitionedRegion.getRegionAdvisor()).thenReturn(mock(RegionAdvisor.class)); when(partitionedRegion.getPartitionAttributes()).thenReturn(mock(PartitionAttributes.class)); when(partitionedRegion.isDataStore()).thenReturn(true); when(partitionedRegion.getPrStats()).thenReturn(mock(PartitionedRegionStats.class)); ScheduledExecutorService executorService = mock(ScheduledExecutorService.class); when(resourceManager.getExecutor()).thenReturn(executorService); when(executorService.schedule(any(Runnable.class), anyLong(), any())) .thenAnswer(runTheRunnable()); @SuppressWarnings("unchecked") CompletableFuture<Void> providerStartupTask = mock(CompletableFuture.class); Exception exception = exceptionToThrow.getException(); prHaRedundancyProvider = new PRHARedundancyProvider(partitionedRegion, resourceManager, (a, b) -> mock(PersistentBucketRecoverer.class), (a, b, c, d, e) -> createThrowingRebalanceOp(exception), providerStartupTask); prHaRedundancyProvider.startRedundancyRecovery(); verify(providerStartupTask, never()).complete(any()); verify(providerStartupTask).completeExceptionally(exception); }