private InternalDistributedMember getPreferredDataStore(
      Collection<InternalDistributedMember> candidates,
      Collection<InternalDistributedMember> alreadyUsed) {
    // has a primary already been chosen?
    boolean forPrimary = alreadyUsed.isEmpty();

    if (forPrimary && getForceLocalPrimaries()) {
      PartitionedRegionDataStore localDataStore = partitionedRegion.getDataStore();
      if (localDataStore != null) {
        return partitionedRegion.getMyId();
      }
    }

    if (candidates.size() == 1) {
      return candidates.iterator().next();
    }
    Assert.assertTrue(candidates.size() > 1);

    // Convert peers to DataStoreBuckets
    List<DataStoreBuckets> stores = partitionedRegion.getRegionAdvisor()
        .adviseFilteredDataStores(new HashSet<>(candidates));

    DistributionManager dm = partitionedRegion.getDistributionManager();

    // Add local member as a candidate, if appropriate
    InternalDistributedMember localMember = dm.getId();
    PartitionedRegionDataStore localDataStore = partitionedRegion.getDataStore();
    if (localDataStore != null && candidates.contains(localMember)) {
      int bucketCount = localDataStore.getBucketsManaged();
      int priCount = localDataStore.getNumberOfPrimaryBucketsManaged();
      int localMaxMemory = partitionedRegion.getLocalMaxMemory();
      stores.add(new DataStoreBuckets(localMember, bucketCount, priCount, localMaxMemory));
    }

    if (stores.isEmpty()) {
      return null;
    }

    // ---------------------------------------------
    // Calculate all hosts who already have this bucket
    Collection<InternalDistributedMember> existingHosts = new HashSet<>();
    for (InternalDistributedMember mem : alreadyUsed) {
      existingHosts.addAll(dm.getMembersInSameZone(mem));
    }

    Comparator<DataStoreBuckets> comparator = (d1, d2) -> {
      boolean host1Used = existingHosts.contains(d1.memberId());
      boolean host2Used = existingHosts.contains(d2.memberId());

      if (!host1Used && host2Used) {
        // host1 preferred
        return -1;
      }
      if (host1Used && !host2Used) {
        // host2 preferred
        return 1;
      }

      // Look for least loaded
      float load1;
      float load2;
      if (forPrimary) {
        load1 = d1.numPrimaries() / (float) d1.localMaxMemoryMB();
        load2 = d2.numPrimaries() / (float) d2.localMaxMemoryMB();
      } else {
        load1 = d1.numBuckets() / (float) d1.localMaxMemoryMB();
        load2 = d2.numBuckets() / (float) d2.localMaxMemoryMB();
      }

      int result = Float.compare(load1, load2);
      if (result == 0) {
        // if they have the same load, choose the member with the higher localMaxMemory
        result = d2.localMaxMemoryMB() - d1.localMaxMemoryMB();
      }
      return result;
    };

    // ---------------------------------------------
    // First step is to sort datastores first by those whose hosts don't
    // hold this bucket, and then secondarily by loading.
    stores.sort(comparator);

    if (logger.isDebugEnabled()) {
      logger.debug(fancyFormatBucketAllocation("Sorted ", stores, existingHosts));
    }

    // ---------------------------------------------
    // Always add the first datastore and note just how good it is.
    DataStoreBuckets bestDataStore = stores.get(0);
    List<DataStoreBuckets> bestStores = new ArrayList<>();
    bestStores.add(bestDataStore);

    boolean allStoresInUse = alreadyUsed.contains(bestDataStore.memberId());

    // ---------------------------------------------
    // Collect all of the other hosts in this sorted list that are as good as the very first one.
    for (int i = 1; i < stores.size(); i++) {
      DataStoreBuckets aDataStore = stores.get(i);
      if (!allStoresInUse && alreadyUsed.contains(aDataStore.memberId())) {
        // Only choose between the ones not in use.
        break;
      }

      if (comparator.compare(bestDataStore, aDataStore) != 0) {
        break;
      }
      bestStores.add(aDataStore);
    }

    if (logger.isDebugEnabled()) {
      logger.debug(fancyFormatBucketAllocation("Best Stores ", bestStores, existingHosts));
    }

    // ---------------------------------------------
    int chosen;
    if (DISABLE_CREATE_BUCKET_RANDOMNESS) {
      chosen = 0;
    } else {
      // Pick one (at random)
      chosen = PartitionedRegion.RANDOM.nextInt(bestStores.size());
    }

    return bestStores.get(chosen).memberId();
  }
Select the member with the fewest buckets, among those with the fewest randomly select one.

Under concurrent access, the data that this method uses, may be somewhat volatile, note that
createBucketAtomically synchronizes to enhance the consistency of the data used in this method.

@param candidates collection of InternalDistributedMember, potential datastores
@param alreadyUsed data stores already in use
@return a member with the fewest buckets or null if no datastores
geode-core/src/main/java/org/apache/geode/internal/cache/PRHARedundancyProvider.java