public InternalDistributedMember createBucketAtomically(int bucketId, int newBucketSize,
      boolean finishIncompleteCreation, String partitionName)
      throws PartitionedRegionStorageException, PartitionedRegionException,
      PartitionOfflineException {
    boolean isDebugEnabled = logger.isDebugEnabled();

    partitionedRegion.checkPROffline();

    // If there are insufficient stores throw *before* we try acquiring the
    // (very expensive) bucket lock or the (somewhat expensive) monitor on this
    earlySufficientStoresCheck(partitionName);

    synchronized (this) {
      if (partitionedRegion.getCache().isCacheAtShutdownAll()) {
        throw partitionedRegion.getCache().getCacheClosedException("Cache is shutting down");
      }

      if (isDebugEnabled) {
        logger.debug("Starting atomic creation of bucketId={}",
            partitionedRegion.bucketStringForLogs(bucketId));
      }

      long timeOut = System.currentTimeMillis() + computeTimeout();
      BucketMembershipObserver observer = null;
      boolean needToElectPrimary = true;
      InternalDistributedMember bucketPrimary = null;

      try {
        partitionedRegion.checkReadiness();

        Bucket toCreate = partitionedRegion.getRegionAdvisor().getBucket(bucketId);

        if (!finishIncompleteCreation) {
          bucketPrimary = partitionedRegion.getBucketPrimary(bucketId);
          if (bucketPrimary != null) {
            if (isDebugEnabled) {
              logger.debug(
                  "during atomic creation, discovered that the primary already exists {} returning early",
                  bucketPrimary);
            }
            needToElectPrimary = false;
            return bucketPrimary;
          }
        }

        observer = new BucketMembershipObserver(toCreate).beginMonitoring();
        ArrayListWithClearState<InternalDistributedMember> failedMembers =
            new ArrayListWithClearState<>();
        Set<InternalDistributedMember> excludedMembers = new HashSet<>();
        Collection<InternalDistributedMember> acceptedMembers = new ArrayList<>();

        for (boolean loggedInsufficientStores = false;;) {
          partitionedRegion.checkReadiness();
          if (partitionedRegion.getCache().isCacheAtShutdownAll()) {
            if (isDebugEnabled) {
              logger.debug("Aborted createBucketAtomically due to ShutdownAll");
            }
            throw partitionedRegion.getCache().getCacheClosedException("Cache is shutting down");
          }

          long timeLeft = timeOut - System.currentTimeMillis();
          if (timeLeft < 0) {
            // It took too long.
            timedOut(partitionedRegion, getAllStores(partitionName), acceptedMembers,
                ALLOCATE_ENOUGH_MEMBERS_TO_HOST_BUCKET, computeTimeout());
          }

          if (isDebugEnabled) {
            logger.debug("createBucketAtomically: have {} ms left to finish this", timeLeft);
          }

          // Always go back to the advisor, see if any fresh data stores are present.
          Set<InternalDistributedMember> allStores = getAllStores(partitionName);

          loggedInsufficientStores = checkSufficientStores(allStores, loggedInsufficientStores);

          InternalDistributedMember candidate = createBucketInstance(bucketId, newBucketSize,
              excludedMembers, acceptedMembers, failedMembers, timeOut, allStores);
          if (candidate != null) {
            if (partitionedRegion.getDistributionManager().enforceUniqueZone()) {
              // enforceUniqueZone property has no effect for a loner
              if (!(partitionedRegion
                  .getDistributionManager() instanceof LonerDistributionManager)) {
                Set<InternalDistributedMember> exm = getBuddyMembersInZone(candidate, allStores);
                exm.remove(candidate);
                exm.removeAll(acceptedMembers);
                excludedMembers.addAll(exm);
              } else {
                // log a warning if Loner
                logger.warn(
                    "enforce-unique-host and redundancy-zone properties have no effect for a LonerDistributedSystem.");
              }
            }
          }

          // Get an updated list of bucket owners, which should include
          // buckets created concurrently with this createBucketAtomically call
          acceptedMembers = partitionedRegion.getRegionAdvisor().getBucketOwners(bucketId);

          if (isDebugEnabled) {
            logger.debug("Accepted members: {}", acceptedMembers);
          }

          // set the primary as the candidate in the first iteration if the candidate has accepted
          if (bucketPrimary == null && acceptedMembers.contains(candidate)) {
            bucketPrimary = candidate;
          }

          // prune out the stores that have left
          verifyBucketNodes(excludedMembers, partitionName);

          // Note - we used to wait for the created bucket to become primary here
          // if this is a colocated region. We no longer need to do that, because
          // the EndBucketMessage is sent out after bucket creation completes to
          // select the primary.

          // Have we exhausted all candidates?
          int potentialCandidateCount = allStores.size()
              - (excludedMembers.size() + acceptedMembers.size() + failedMembers.size());

          // Determining exhausted members competes with bucket balancing; it's
          // important to re-visit all failed members since "failed" set may
          // contain datastores which at the moment are imbalanced, but yet could
          // be candidates. If the failed members list is empty, its expected
          // that the next iteration clears the (already empty) list.
          boolean exhaustedPotentialCandidates =
              failedMembers.wasCleared() && potentialCandidateCount <= 0;
          boolean redundancySatisfied =
              acceptedMembers.size() > partitionedRegion.getRedundantCopies();
          boolean bucketNotCreated = acceptedMembers.isEmpty();

          if (isDebugEnabled) {
            logger.debug(
                "potentialCandidateCount={}, exhaustedPotentialCandidates={}, redundancySatisfied={}, bucketNotCreated={}",
                potentialCandidateCount, exhaustedPotentialCandidates, redundancySatisfied,
                bucketNotCreated);
          }

          if (bucketNotCreated) {
            // if we haven't managed to create the bucket on any nodes, retry.
            continue;
          }

          if (exhaustedPotentialCandidates && !redundancySatisfied) {
            insufficientStores(allStores, acceptedMembers, true);
          }

          // Allow the thread to potentially finish bucket creation even if redundancy was not met.
          if (redundancySatisfied || exhaustedPotentialCandidates) {
            // Tell one of the members to become primary.
            // The rest of the members will be allowed to volunteer for primary.
            endBucketCreation(bucketId, acceptedMembers, bucketPrimary, partitionName);

            int expectedRemoteHosts = acceptedMembers.size()
                - (acceptedMembers.contains(partitionedRegion.getMyId()) ? 1 : 0);

            boolean interrupted = Thread.interrupted();
            try {
              BucketMembershipObserverResults results = observer
                  .waitForOwnersGetPrimary(expectedRemoteHosts, acceptedMembers, partitionName);
              if (results.problematicDeparture) {
                // BZZZT! Member left. Start over.
                continue;
              }
              bucketPrimary = results.primary;
            } catch (InterruptedException e) {
              interrupted = true;
              partitionedRegion.getCancelCriterion().checkCancelInProgress(e);
            } finally {
              if (interrupted) {
                Thread.currentThread().interrupt();
              }
            }

            needToElectPrimary = false;

            return bucketPrimary;
          }
        }
      } catch (DiskAccessException dae) {
        CancelCriterion cancelCriterion = partitionedRegion.getCancelCriterion();
        if (cancelCriterion.isCancelInProgress()) {
          needToElectPrimary = false;
          cancelCriterion.checkCancelInProgress(dae);
        }

        throw dae;
      } catch (CancelException | RegionDestroyedException e) {
        // We don't need to elect a primary if the cache was closed. The other members will
        // take care of it. This ensures we don't compromise redundancy.
        needToElectPrimary = false;
        throw e;
      } catch (PartitionOfflineException e) {
        throw e;
      } catch (RuntimeException e) {
        if (isDebugEnabled) {
          logger.debug("Unable to create new bucket {}: {}", bucketId, e.getMessage(), e);
        }

        if (!finishIncompleteCreation) {
          cleanUpBucket(bucketId);
        }
        throw e;
      } finally {
        if (observer != null) {
          observer.stopMonitoring();
        }

        // Try to make sure everyone that created the bucket can volunteer for primary
        if (needToElectPrimary) {
          try {
            endBucketCreation(bucketId,
                partitionedRegion.getRegionAdvisor().getBucketOwners(bucketId),
                bucketPrimary, partitionName);
          } catch (Exception e) {
            // if region is going down, then no warning level logs
            if (e instanceof CancelException
                || partitionedRegion.getCancelCriterion().isCancelInProgress()) {
              logger.debug("Exception trying choose a primary after bucket creation failure", e);
            } else {
              logger.warn("Exception trying choose a primary after bucket creation failure", e);
            }
          }
        }
      }
    }
  }
Creates bucket atomically by creating all the copies to satisfy redundancy. In case all copies
can not be created, a PartitionedRegionStorageException is thrown to the user and
BucketBackupMessage is sent to the nodes to make copies of a bucket that was only partially
created. Other VMs are informed of bucket creation through updates through their
{@link BucketAdvisor.BucketProfile}s.

<p>
This method is synchronized to enforce a single threaded ordering, allowing for a more accurate
picture of bucket distribution in the face of concurrency.

<p>
This method is now slightly misnamed. Another member could be in the process of creating this
same bucket at the same time.

@param bucketId Id of the bucket to be created.
@param newBucketSize size of the first entry.
@return the primary member for the newly created bucket
@throws PartitionedRegionStorageException if required # of buckets can not be created to
        satisfy redundancy.
@throws PartitionedRegionException if d-lock can not be acquired to create bucket.
@throws PartitionOfflineException if persistent data recovery is not complete for a partitioned
        region referred to in the query.
geode-core/src/main/java/org/apache/geode/internal/cache/PRHARedundancyProvider.java