BucketOperatorWrapper implements BucketOperator { @Override public boolean removeBucket(InternalDistributedMember targetMember, int i, Map<String, Long> colocatedRegionBytes) { boolean result = false; long elapsed = 0; long totalBytes = 0; if (stats != null) { stats.startBucketRemove(regionCount); } try { long start = System.nanoTime(); result = delegate.removeBucket(targetMember, i, colocatedRegionBytes); elapsed = System.nanoTime() - start; if (result) { if (logger.isDebugEnabled()) { logger.debug("Rebalancing {} redundant bucket {} removed from {}", leaderRegion, i, targetMember); } for (PartitionRebalanceDetailsImpl details : detailSet) { String regionPath = details.getRegionPath(); Long lrb = colocatedRegionBytes.get(regionPath); if (lrb != null) { long regionBytes = lrb.longValue(); details.incRemoves(regionBytes, details.getRegion().equals(leaderRegion) ? elapsed : 0); totalBytes += regionBytes; } } } else { if (logger.isDebugEnabled()) { logger.debug("Rebalancing {} redundant bucket {} failed removal o{}", leaderRegion, i, targetMember); } } } finally { if (stats != null) { stats.endBucketRemove(regionCount, result, totalBytes, elapsed); } } return result; } BucketOperatorWrapper(BucketOperator delegate,
      Set<PartitionRebalanceDetailsImpl> rebalanceDetails, ResourceManagerStats stats,
      PartitionedRegion leaderRegion); @Override boolean moveBucket(InternalDistributedMember sourceMember,
      InternalDistributedMember targetMember, int id, Map<String, Long> colocatedRegionBytes); @Override void createRedundantBucket(final InternalDistributedMember targetMember, final int i,
      final Map<String, Long> colocatedRegionBytes, final Completion completion); @Override boolean removeBucket(InternalDistributedMember targetMember, int i,
      Map<String, Long> colocatedRegionBytes); @Override boolean movePrimary(InternalDistributedMember source, InternalDistributedMember target,
      int bucketId); @Override void waitForOperations(); Set<PartitionRebalanceDetailsImpl> getDetailSet();  }
@Test public void bucketWrapperShouldDelegateRemoveBucketToEnclosedOperator() { wrapper.removeBucket(targetMember, bucketId, colocatedRegionBytes); verify(delegate, times(1)).removeBucket(targetMember, bucketId, colocatedRegionBytes); verify(stats, times(1)).startBucketRemove(anyInt()); verify(stats, times(1)).endBucketRemove(anyInt(), anyBoolean(), anyLong(), anyLong()); }
@Test public void bucketWrapperShouldRecordBucketRemovesPerRegionAfterRemoveBucketIsSuccessful() { doReturn(true).when(delegate).removeBucket(targetMember, bucketId, colocatedRegionBytes); wrapper.removeBucket(targetMember, bucketId, colocatedRegionBytes); for (PartitionRebalanceDetailsImpl details : rebalanceDetails) { if (details.getRegionPath().equalsIgnoreCase(PR_LEADER_REGION_NAME)) verify(details, times(1)).incRemoves((eq(colocatedRegionBytes.get(PR_LEADER_REGION_NAME))), anyLong()); else if (details.getRegionPath().equals(PR_COLOCATED_REGION_NAME)) verify(details, times(1)).incRemoves(colocatedRegionBytes.get(PR_COLOCATED_REGION_NAME), 0); } verify(stats, times(1)).startBucketRemove(anyInt()); verify(stats, times(1)).endBucketRemove(anyInt(), anyBoolean(), anyLong(), anyLong()); }
@Test public void bucketWrapperShouldDoNotRecordBucketRemovesIfMoveBucketFails() { doReturn(false).when(delegate).removeBucket(targetMember, bucketId, colocatedRegionBytes); wrapper.removeBucket(targetMember, bucketId, colocatedRegionBytes); for (PartitionRebalanceDetailsImpl details : rebalanceDetails) { verify(details, times(0)).incTransfers(anyLong(), anyLong()); } verify(stats, times(1)).startBucketRemove(anyInt()); verify(stats, times(1)).endBucketRemove(anyInt(), anyBoolean(), anyLong(), anyLong()); }