InternalDistributedMember createBucketOnDataStore(int bucketId, int size,
      RetryTimeKeeper retryTimeKeeper) {
    boolean isDebugEnabled = logger.isDebugEnabled();

    InternalDistributedMember primaryForFixedPartition = null;
    if (partitionedRegion.isFixedPartitionedRegion()) {
      primaryForFixedPartition =
          partitionedRegion.getRegionAdvisor().adviseFixedPrimaryPartitionDataStore(bucketId);
    }

    InternalDistributedMember memberHostingBucket;
    Collection<InternalDistributedMember> attempted = new HashSet<>();
    do {
      partitionedRegion.checkReadiness();
      Set<InternalDistributedMember> available =
          partitionedRegion.getRegionAdvisor().adviseInitializedDataStore();
      available.removeAll(attempted);
      InternalDistributedMember targetMember = null;
      for (InternalDistributedMember member : available) {
        if (available.contains(primaryForFixedPartition)) {
          targetMember = primaryForFixedPartition;
        } else {
          targetMember = member;
        }
        break;
      }
      if (targetMember == null) {
        if (shouldLogInsufficientStores()) {
          insufficientStores(available, Collections.emptySet(), true);
        }
        // this will always throw an exception
        insufficientStores(available, Collections.emptySet(), false);
      }
      try {
        if (isDebugEnabled) {
          logger.debug("Attempting to get data store {} to create the bucket {} for us",
              targetMember,
              partitionedRegion.bucketStringForLogs(bucketId));
        }
        CreateBucketMessage.NodeResponse response =
            CreateBucketMessage.send(targetMember, partitionedRegion, bucketId, size);
        memberHostingBucket = response.waitForResponse();
        if (memberHostingBucket != null) {
          return memberHostingBucket;
        }
      } catch (ForceReattemptException e) {
        // do nothing, we will already check again for a primary.
      }
      attempted.add(targetMember);
    } while ((memberHostingBucket =
        partitionedRegion.getNodeForBucketWrite(bucketId, retryTimeKeeper)) == null);
    return memberHostingBucket;
  }
Create a single copy of this bucket on one node. The bucket must already be locked.

@param bucketId The bucket we are working on
@param newBucketSize size to create it
@param alreadyUsed members who already seem to have the bucket
@param timeOut point at which to fail
@param allStores the set of data stores to choose from
@return the new member, null if it fails.
@throws PartitionedRegionStorageException if there are not enough data stores
geode-core/src/main/java/org/apache/geode/internal/cache/PRHARedundancyProvider.java