private InternalDistributedMember createBucketInstance(int bucketId, int newBucketSize,
      Collection<InternalDistributedMember> excludedMembers,
      Collection<InternalDistributedMember> alreadyUsed,
      ArrayListWithClearState<InternalDistributedMember> failedMembers, long timeOut,
      Set<InternalDistributedMember> allStores) {

    boolean isDebugEnabled = logger.isDebugEnabled();

    // Recalculate list of candidates
    Set<InternalDistributedMember> candidateMembers = new HashSet<>(allStores);
    candidateMembers.removeAll(alreadyUsed);
    candidateMembers.removeAll(excludedMembers);
    candidateMembers.removeAll(failedMembers);

    if (isDebugEnabled) {
      logger.debug("AllStores={} AlreadyUsed={} excluded={} failed={}", allStores, alreadyUsed,
          excludedMembers, failedMembers);
    }

    if (candidateMembers.isEmpty()) {
      partitionedRegion.checkReadiness();

      // Run out of candidates. Refetch?
      if (System.currentTimeMillis() > timeOut) {
        if (isDebugEnabled) {
          logger.debug("createBucketInstance: ran out of candidates and timed out");
        }
        // fail, let caller signal error
        return null;
      }

      // Recalculate
      candidateMembers = new HashSet<>(allStores);
      candidateMembers.removeAll(alreadyUsed);
      candidateMembers.removeAll(excludedMembers);
      failedMembers.clear();
    }

    if (isDebugEnabled) {
      logger.debug("createBucketInstance: candidateMembers = {}", candidateMembers);
    }

    // If there are no candidates, early out.
    if (candidateMembers.isEmpty()) {
      // no options
      if (isDebugEnabled) {
        logger.debug("createBucketInstance: no valid candidates");
      }
      // failure
      return null;
    }

    // In case of FixedPartitionedRegion, candidateMembers is the set of members on which
    // required fixed partition is defined.
    InternalDistributedMember candidate;
    if (partitionedRegion.isFixedPartitionedRegion()) {
      candidate = candidateMembers.iterator().next();
    } else {
      String colocatedWith =
          partitionedRegion.getAttributes().getPartitionAttributes().getColocatedWith();
      if (colocatedWith != null) {
        candidate = getColocatedDataStore(candidateMembers, alreadyUsed, bucketId, colocatedWith);
      } else {
        Collection<InternalDistributedMember> orderedCandidates = new ArrayList<>(candidateMembers);
        candidate = getPreferredDataStore(orderedCandidates, alreadyUsed);
      }
    }

    if (candidate == null) {
      failedMembers.addAll(candidateMembers);
      return null;
    }

    if (!partitionedRegion.isShadowPR() && !checkMembersColocation(partitionedRegion, candidate)) {
      if (isDebugEnabled) {
        logger.debug(
            "createBucketInstances - Member does not have all of the regions colocated with partitionedRegion {}",
            candidate);
      }
      failedMembers.add(candidate);
      return null;
    }

    if (!candidate.equals(partitionedRegion.getMyId())) {
      PartitionProfile profile =
          partitionedRegion.getRegionAdvisor().getPartitionProfile(candidate);
      if (profile == null) {
        if (isDebugEnabled) {
          logger.debug("createBucketInstance: {}: no partition profile for {}",
              partitionedRegion.getFullPath(), candidate);
        }
        failedMembers.add(candidate);
        return null;
      }
    }

    // Coordinate with any remote close occurring, causing it to wait until
    // this create bucket attempt has been made.
    ManageBucketRsp response =
        createBucketOnMember(bucketId, candidate, newBucketSize, failedMembers.wasCleared());

    // Add targetNode to bucketNodes if successful, else to failedNodeList
    if (response.isAcceptance()) {
      // success!
      return candidate;
    }

    if (isDebugEnabled) {
      logger.debug("createBucketInstance: {}: candidate {} declined to manage bucketId={}: {}",
          partitionedRegion.getFullPath(), candidate,
          partitionedRegion.bucketStringForLogs(bucketId),
          response);
    }
    if (response.equals(ManageBucketRsp.CLOSED)) {
      excludedMembers.add(candidate);
    } else {
      failedMembers.add(candidate);
    }

    return null;
  }
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