class 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) { // region could have gone away - esp during shutdow
            long regionBytes = lrb.longValue();
            // Only add the elapsed time to the leader region.
            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;
  }

  public  BucketOperatorWrapper(BucketOperator delegate,
      Set<PartitionRebalanceDetailsImpl> rebalanceDetails, ResourceManagerStats stats,
      PartitionedRegion leaderRegion);

}