class BucketOperatorWrapper implements BucketOperator {

  @Override
  public boolean moveBucket(InternalDistributedMember sourceMember,
      InternalDistributedMember targetMember, int id, Map<String, Long> colocatedRegionBytes) {
    long start = System.nanoTime();
    boolean result = false;
    long elapsed = 0;
    long totalBytes = 0;

    if (stats != null) {
      stats.startBucketTransfer(regionCount);
    }
    try {
      result = delegate.moveBucket(sourceMember, targetMember, id, colocatedRegionBytes);
      elapsed = System.nanoTime() - start;
      if (result) {
        if (logger.isDebugEnabled()) {
          logger.debug("Rebalancing {} bucket {} moved from {} to {}", leaderRegion, id,
              sourceMember, targetMember);
        }
        for (PartitionRebalanceDetailsImpl details : detailSet) {
          String regionPath = details.getRegionPath();
          Long regionBytes = colocatedRegionBytes.get(regionPath);
          if (regionBytes != null) {
            // only increment the elapsed time for the leader region
            details.incTransfers(regionBytes.longValue(),
                details.getRegion().equals(leaderRegion) ? elapsed : 0);
            totalBytes += regionBytes.longValue();
          }
        }
      } else {
        if (logger.isDebugEnabled()) {
          logger.debug("Rebalancing {} bucket {} moved failed from {} to {}", leaderRegion, id,
              sourceMember, targetMember);
        }
      }
    } finally {
      if (stats != null) {
        stats.endBucketTransfer(regionCount, result, totalBytes, elapsed);
      }
    }

    return result;
  }

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

}