class BucketOperatorWrapper implements BucketOperator {

  @Override
  public boolean movePrimary(InternalDistributedMember source, InternalDistributedMember target,
      int bucketId) {
    boolean result = false;
    long elapsed = 0;

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

    try {
      long start = System.nanoTime();
      result = delegate.movePrimary(source, target, bucketId);
      elapsed = System.nanoTime() - start;
      if (result) {
        if (logger.isDebugEnabled()) {
          logger.debug("Rebalancing {} primary bucket {} moved from {} to {}", leaderRegion,
              bucketId, source, target);
        }
        for (PartitionRebalanceDetailsImpl details : detailSet) {
          details.incPrimaryTransfers(details.getRegion().equals(leaderRegion) ? elapsed : 0);
        }
      } else {
        if (logger.isDebugEnabled()) {
          logger.debug("Rebalancing {} primary bucket {} failed to move from {} to {}",
              leaderRegion, bucketId, source, target);
        }
      }
    } finally {
      if (stats != null) {
        stats.endPrimaryTransfer(regionCount, result, elapsed);
      }
    }

    return result;
  }

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

}