private ManageBucketRsp createBucketOnMember(int bucketId, InternalDistributedMember targetMember,
      int newBucketSize, boolean forceCreation) {
    if (logger.isDebugEnabled()) {
      logger.debug("createBucketOnMember for bucketId={} member: {}{}",
          partitionedRegion.bucketStringForLogs(bucketId), targetMember,
          forceCreation ? " forced" : "");
    }

    if (!targetMember.equals(partitionedRegion.getMyId())) {
      PartitionProfile profile =
          partitionedRegion.getRegionAdvisor().getPartitionProfile(targetMember);
      if (profile == null) {
        return ManageBucketRsp.NO;
      }

      try {
        NodeResponse response = ManageBucketMessage.send(targetMember, partitionedRegion, bucketId,
            newBucketSize, forceCreation);

        if (response.waitForAcceptance()) {
          if (logger.isDebugEnabled()) {
            logger.debug(
                "createBucketOnMember: Bucket creation succeed for bucketId={} on member = {}",
                partitionedRegion.bucketStringForLogs(bucketId), targetMember);
          }

          return ManageBucketRsp.YES;
        }

        if (logger.isDebugEnabled()) {
          logger.debug(
              "createBucketOnMember: Bucket creation failed for bucketId={} on member = {}",
              partitionedRegion.bucketStringForLogs(bucketId), targetMember);
        }

        return response.rejectedDueToInitialization() ? ManageBucketRsp.NO_INITIALIZING
            : ManageBucketRsp.NO;

      } catch (PartitionOfflineException e) {
        throw e;
      } catch (VirtualMachineError err) {
        SystemFailure.initiateFailure(err);
        // If this ever returns, rethrow the error. We're poisoned
        // now, so don't let this thread continue.
        throw err;
      } catch (Throwable e) {
        // Whenever you catch Error or Throwable, you must also
        // catch VirtualMachineError (see above). However, there is
        // _still_ a possibility that you are dealing with a cascading
        // error condition, so you also need to check to see if the JVM
        // is still usable:
        SystemFailure.checkFailure();
        if (e instanceof CancelException
            || e.getCause() != null && e.getCause() instanceof CancelException) {
          // no need to log exceptions caused by cache closure
          return ManageBucketRsp.CLOSED;
        }
        if (e instanceof ForceReattemptException) {
          // no log needed
        } else {
          logger.warn("Exception creating partition on {}", targetMember, e);
        }
        return ManageBucketRsp.NO;
      }
    }

    PartitionedRegionDataStore dataStore = partitionedRegion.getDataStore();
    boolean bucketManaged = dataStore != null && dataStore.handleManageBucketRequest(bucketId,
        newBucketSize, partitionedRegion.getMyId(), forceCreation);
    if (!bucketManaged) {
      if (logger.isDebugEnabled()) {
        logger.debug(
            "createBucketOnMember: Local data store not able to accommodate the data for bucketId={}",
            partitionedRegion.bucketStringForLogs(bucketId));
      }
    }
    return ManageBucketRsp.valueOf(bucketManaged);
  }
Creates bucket with ID bucketId on targetNode.

@param forceCreation inform the targetMember it must attempt host the bucket, appropriately
       ignoring it's maximums
@return a response object
geode-core/src/main/java/org/apache/geode/internal/cache/PRHARedundancyProvider.java