class BucketOperatorWrapper implements BucketOperator {

  @Override
  public void createRedundantBucket(final InternalDistributedMember targetMember, final int i,
      final Map<String, Long> colocatedRegionBytes, final Completion completion) {

    if (stats != null) {
      stats.startBucketCreate(regionCount);
    }

    final long start = System.nanoTime();
    delegate.createRedundantBucket(targetMember, i, colocatedRegionBytes, new Completion() {

      @Override
      public void onSuccess() {
        long totalBytes = 0;
        long elapsed = System.nanoTime() - start;
        if (logger.isDebugEnabled()) {
          logger.debug("Rebalancing {} redundant bucket {} created on {}", 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.incCreates(regionBytes, details.getRegion().equals(leaderRegion) ? elapsed : 0);
            totalBytes += regionBytes;
          }
        }

        if (stats != null) {
          stats.endBucketCreate(regionCount, true, totalBytes, elapsed);
        }

        // invoke onSuccess on the received completion callback
        completion.onSuccess();
      }

      @Override
      public void onFailure() {
        long elapsed = System.nanoTime() - start;

        if (logger.isDebugEnabled()) {
          logger.info("Rebalancing {} redundant bucket {} failed creation on {}", leaderRegion, i,
              targetMember);
        }

        if (stats != null) {
          stats.endBucketCreate(regionCount, false, 0, elapsed);
        }

        // invoke onFailure on the received completion callback
        completion.onFailure();
      }
    });
  }

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

}