/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */
package org.apache.geode.internal.cache;

import static java.lang.String.format;
import static java.lang.System.lineSeparator;
import static java.util.Collections.emptyList;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.geode.internal.cache.ColocationHelper.checkMembersColocation;
import static org.apache.geode.internal.cache.PartitionedRegionHelper.printCollection;
import static org.apache.geode.util.internal.GeodeGlossary.GEMFIRE_PREFIX;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.TreeSet;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.BiFunction;

import org.apache.logging.log4j.Logger;

import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.cache.DiskAccessException;
import org.apache.geode.cache.PartitionedRegionStorageException;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.RegionDestroyedException;
import org.apache.geode.cache.persistence.PartitionOfflineException;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.internal.DistributionManager;
import org.apache.geode.distributed.internal.LonerDistributionManager;
import org.apache.geode.distributed.internal.MembershipListener;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.NanoTimer;
import org.apache.geode.internal.OneTaskOnlyExecutor;
import org.apache.geode.internal.cache.PartitionedRegion.RetryTimeKeeper;
import org.apache.geode.internal.cache.PartitionedRegionDataStore.CreateBucketResult;
import org.apache.geode.internal.cache.control.InternalResourceManager;
import org.apache.geode.internal.cache.partitioned.Bucket;
import org.apache.geode.internal.cache.partitioned.BucketBackupMessage;
import org.apache.geode.internal.cache.partitioned.CreateBucketMessage;
import org.apache.geode.internal.cache.partitioned.CreateMissingBucketsTask;
import org.apache.geode.internal.cache.partitioned.DataStoreBuckets;
import org.apache.geode.internal.cache.partitioned.EndBucketCreationMessage;
import org.apache.geode.internal.cache.partitioned.FetchPartitionDetailsMessage;
import org.apache.geode.internal.cache.partitioned.FetchPartitionDetailsMessage.FetchPartitionDetailsResponse;
import org.apache.geode.internal.cache.partitioned.InternalPRInfo;
import org.apache.geode.internal.cache.partitioned.InternalPartitionDetails;
import org.apache.geode.internal.cache.partitioned.LoadProbe;
import org.apache.geode.internal.cache.partitioned.ManageBackupBucketMessage;
import org.apache.geode.internal.cache.partitioned.ManageBucketMessage;
import org.apache.geode.internal.cache.partitioned.ManageBucketMessage.NodeResponse;
import org.apache.geode.internal.cache.partitioned.OfflineMemberDetails;
import org.apache.geode.internal.cache.partitioned.OfflineMemberDetailsImpl;
import org.apache.geode.internal.cache.partitioned.PRLoad;
import org.apache.geode.internal.cache.partitioned.PartitionMemberInfoImpl;
import org.apache.geode.internal.cache.partitioned.PartitionRegionInfoImpl;
import org.apache.geode.internal.cache.partitioned.PartitionedRegionRebalanceOp;
import org.apache.geode.internal.cache.partitioned.PartitionedRegionRebalanceOpFactory;
import org.apache.geode.internal.cache.partitioned.PersistentBucketRecoverer;
import org.apache.geode.internal.cache.partitioned.RecoveryRunnable;
import org.apache.geode.internal.cache.partitioned.RegionAdvisor;
import org.apache.geode.internal.cache.partitioned.RegionAdvisor.PartitionProfile;
import org.apache.geode.internal.cache.partitioned.rebalance.CompositeDirector;
import org.apache.geode.internal.cache.partitioned.rebalance.FPRDirector;
import org.apache.geode.internal.cache.partitioned.rebalance.RebalanceDirector;
import org.apache.geode.internal.cache.persistence.MembershipFlushRequest;
import org.apache.geode.internal.cache.persistence.PersistentMemberID;
import org.apache.geode.internal.monitoring.ThreadsMonitoring;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * This class provides the redundancy management for partitioned region. It will provide the
 * following to the PartitionedRegion:
 *
 * <ol>
 * <li>Redundancy management at the time of bucket creation.
 * <li>Redundancy management at the new node arrival.
 * <li>Redundancy management when the node leaves the partitioned region distributed system
 * gracefully. i.e. Cache.close()
 * <li>Redundancy management at random node failure.
 * </ol>
 */
public class PRHARedundancyProvider {
  private static final Logger logger = LogService.getLogger();

  public static final String TIMEOUT_MSG =
      "If your system has sufficient space, perhaps it is under membership or region creation stress?";

  /**
   * Signature string indicating that not enough stores are available.
   */
  public static final String INSUFFICIENT_STORES_MSG =
      "Advise you to start enough data store nodes";

  private static final boolean DISABLE_CREATE_BUCKET_RANDOMNESS =
      Boolean.getBoolean(GEMFIRE_PREFIX + "DISABLE_CREATE_BUCKET_RANDOMNESS");

  private static final String DATASTORE_DISCOVERY_TIMEOUT_PROPERTY_NAME =
      GEMFIRE_PREFIX + "partitionedRegionDatastoreDiscoveryTimeout";

  /**
   * Signature string indicating that there are enough stores available.
   */
  private static final String SUFFICIENT_STORES_MSG = "Found a member to host a bucket.";

  /**
   * string indicating the attempt to allocate a bucket
   */
  private static final String ALLOCATE_ENOUGH_MEMBERS_TO_HOST_BUCKET =
      "allocate enough members to host a new bucket";

  private static final long INSUFFICIENT_LOGGING_THROTTLE_TIME = TimeUnit.SECONDS.toNanos(
      Integer.getInteger(GEMFIRE_PREFIX + "InsufficientLoggingThrottleTime", 2));

  private static final ThreadLocal<Boolean> forceLocalPrimaries = new ThreadLocal<>();

  @MakeNotStatic
  private static final Long DATASTORE_DISCOVERY_TIMEOUT_MILLISECONDS =
      Long.getLong(DATASTORE_DISCOVERY_TIMEOUT_PROPERTY_NAME);

  @MakeNotStatic
  private static final AtomicLong insufficientLogTimeStamp = new AtomicLong(0);

  @MakeNotStatic
  private final AtomicBoolean firstInsufficientStoresLogged = new AtomicBoolean(false);

  private final PartitionedRegion partitionedRegion;
  private final InternalResourceManager resourceManager;
  private final PartitionedRegionRebalanceOpFactory rebalanceOpFactory;
  private final CompletableFuture<Void> providerStartupTask;

  /**
   * An executor to submit tasks for redundancy recovery too. It makes sure that there will only be
   * one redundancy recovery task in the queue at a time.
   */
  private final OneTaskOnlyExecutor recoveryExecutor;

  private final Object shutdownLock = new Object();

  private final BiFunction<PRHARedundancyProvider, Integer, PersistentBucketRecoverer> persistentBucketRecovererFunction;

  private volatile ScheduledFuture<?> recoveryFuture;

  /**
   * Used to consolidate logging for bucket regions waiting on other members to come online.
   */
  private volatile PersistentBucketRecoverer persistentBucketRecoverer;

  private boolean shutdown;

  /**
   * Constructor for PRHARedundancyProvider.
   *
   * @param partitionedRegion The PartitionedRegion for which the HA redundancy is required to be
   *        managed.
   */
  public PRHARedundancyProvider(PartitionedRegion partitionedRegion,
      InternalResourceManager resourceManager) {
    this(partitionedRegion, resourceManager, PersistentBucketRecoverer::new,
        PartitionedRegionRebalanceOp::new,
        new CompletableFuture<>());
  }

  @VisibleForTesting
  PRHARedundancyProvider(PartitionedRegion partitionedRegion,
      InternalResourceManager resourceManager,
      BiFunction<PRHARedundancyProvider, Integer, PersistentBucketRecoverer> persistentBucketRecovererFunction) {
    this(partitionedRegion, resourceManager, persistentBucketRecovererFunction,
        PartitionedRegionRebalanceOp::new, new CompletableFuture<>());
  }

  @VisibleForTesting
  PRHARedundancyProvider(PartitionedRegion partitionedRegion,
      InternalResourceManager resourceManager,
      BiFunction<PRHARedundancyProvider, Integer, PersistentBucketRecoverer> persistentBucketRecovererFunction,
      PartitionedRegionRebalanceOpFactory rebalanceOpFactory,
      CompletableFuture<Void> providerStartupTask) {
    this.partitionedRegion = partitionedRegion;
    this.resourceManager = resourceManager;
    this.rebalanceOpFactory = rebalanceOpFactory;
    this.providerStartupTask = providerStartupTask;
    recoveryExecutor = new OneTaskOnlyExecutor(resourceManager.getExecutor(),
        () -> InternalResourceManager.getResourceObserver().recoveryConflated(partitionedRegion),
        getThreadMonitorObj());
    this.persistentBucketRecovererFunction = persistentBucketRecovererFunction;
  }

  /**
   * Display bucket allocation status
   *
   * @param partitionedRegion the given region
   * @param allStores the list of available stores. If null, unknown.
   * @param alreadyUsed stores allocated; only used if allStores != null
   * @param forLog true if the generated string is for a log message
   * @return the description string
   */
  private static String regionStatus(PartitionedRegion partitionedRegion,
      Collection<InternalDistributedMember> allStores,
      Collection<InternalDistributedMember> alreadyUsed, boolean forLog) {
    String newLine = forLog ? " " : lineSeparator();
    String spaces = forLog ? "" : "   ";

    StringBuilder sb = new StringBuilder();
    sb.append("Partitioned Region name = ");
    sb.append(partitionedRegion.getFullPath());

    if (allStores != null) {
      sb.append(newLine).append(spaces);
      sb.append("Redundancy level set to ");
      sb.append(partitionedRegion.getRedundantCopies());
      sb.append(newLine);
      sb.append(". Number of available data stores: ");
      sb.append(allStores.size());
      sb.append(newLine).append(spaces);
      sb.append(". Number successfully allocated = ");
      sb.append(alreadyUsed.size());
      sb.append(newLine);
      sb.append(". Data stores: ");
      sb.append(printCollection(allStores));
      sb.append(newLine);
      sb.append(". Data stores successfully allocated: ");
      sb.append(printCollection(alreadyUsed));
      sb.append(newLine);
      sb.append(". Equivalent members: ");
      sb.append(printCollection(partitionedRegion.getDistributionManager().getMembersInThisZone()));
    }

    return sb.toString();
  }

  /**
   * Indicate a timeout due to excessive retries among available peers
   *
   * @param allStores all feasible stores. If null, we don't know.
   * @param alreadyUsed those that have already accepted, only used if allStores != null
   * @param opString description of the operation which timed out
   */
  public static void timedOut(PartitionedRegion partitionedRegion,
      Set<InternalDistributedMember> allStores,
      Collection<InternalDistributedMember> alreadyUsed, String opString, long timeOut) {
    throw new PartitionedRegionStorageException(
        format("Timed out attempting to %s in the partitioned region.%sWaited for: %s ms.",
            opString, regionStatus(partitionedRegion, allStores, alreadyUsed, true), timeOut)
            + TIMEOUT_MSG);
  }

  public PartitionedRegion getPartitionedRegion() {
    return partitionedRegion;
  }

  private Set<InternalDistributedMember> getAllStores(String partitionName) {
    if (partitionName != null) {
      return getFixedPartitionStores(partitionName);
    }
    final Set<InternalDistributedMember> allStores =
        partitionedRegion.getRegionAdvisor().adviseDataStore(true);
    PartitionedRegionDataStore localDataStore = partitionedRegion.getDataStore();
    if (localDataStore != null) {
      allStores.add(partitionedRegion.getDistributionManager().getId());
    }
    return allStores;
  }

  /**
   * This is for FPR, for given partition, we have to return the set of datastores on which the
   * given partition is defined
   *
   * @param partitionName name of the partition for which datastores need to be found out
   */
  private Set<InternalDistributedMember> getFixedPartitionStores(String partitionName) {
    Set<InternalDistributedMember> members =
        partitionedRegion.getRegionAdvisor().adviseFixedPartitionDataStores(partitionName);

    List<FixedPartitionAttributesImpl> allFixedPartitionAttributes =
        partitionedRegion.getFixedPartitionAttributesImpl();

    if (allFixedPartitionAttributes != null) {
      for (FixedPartitionAttributesImpl fixedPartitionAttributes : allFixedPartitionAttributes) {
        if (fixedPartitionAttributes.getPartitionName().equals(partitionName)) {
          members.add(partitionedRegion.getMyId());
        }
      }
    }
    return members;
  }

  /**
   * Indicate that we are unable to allocate sufficient stores and the timeout period has passed
   *
   * @param allStores stores we know about
   * @param alreadyUsed ones already committed
   * @param onlyLog true if only a warning log messages should be generated.
   */
  private void insufficientStores(Set<InternalDistributedMember> allStores,
      Collection<InternalDistributedMember> alreadyUsed, boolean onlyLog) {
    String regionStat = regionStatus(partitionedRegion, allStores, alreadyUsed, onlyLog);
    String newLine = onlyLog ? " " : lineSeparator();
    String notEnoughValidNodes = alreadyUsed.isEmpty()
        ? "Unable to find any members to host a bucket in the partitioned region. %s.%s"
        : "Configured redundancy level could not be satisfied. %s to satisfy redundancy for the region.%s";
    if (onlyLog) {
      logger.warn(format(notEnoughValidNodes, INSUFFICIENT_STORES_MSG,
          newLine + regionStat + newLine));
    } else {
      throw new PartitionedRegionStorageException(
          format(notEnoughValidNodes, INSUFFICIENT_STORES_MSG, newLine + regionStat + newLine));
    }
  }

  /**
   * 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
   */
  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;
  }

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

  /**
   * 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.
   */
  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;
