CreateBucketResult grabFreeBucket(final int possiblyFreeBucketId, final DistributedMember sender,
      final InternalDistributedMember moveSource, final boolean forceCreation,
      final boolean isRebalance, final boolean lockRedundancyLock, boolean replaceOffineData,
      InternalDistributedMember creationRequestor) {

    final boolean isDebugEnabled = logger.isDebugEnabled();

    long startTime = partitionedRegion.getPrStats().startBucketCreate(isRebalance);
    boolean createdBucket = false;
    PartitionedRegionObserver observer = PartitionedRegionObserverHolder.getInstance();
    observer.beforeBucketCreation(partitionedRegion, possiblyFreeBucketId);
    try {

      CreateBucketResult result = CreateBucketResult.FAILED;
      if (isManagingBucket(possiblyFreeBucketId)) {
        if (isDebugEnabled) {
          logger.debug("grabFreeBucket: VM {} already contains the bucket with bucketId={}{}{}",
              partitionedRegion.getMyId(), partitionedRegion.getPRId(),
              PartitionedRegion.BUCKET_ID_SEPARATOR, possiblyFreeBucketId);
        }
        partitionedRegion.checkReadiness();
        return CreateBucketResult.ALREADY_EXISTS;
      }
      StoppableReadLock parentBucketCreationLock = getParentBucketCreationLock();
      if (parentBucketCreationLock != null) {
        parentBucketCreationLock.lock();
      }

      try {
        if (!okToCreateChildBucket(possiblyFreeBucketId)) {
          return CreateBucketResult.FAILED;
        }
        StoppableReadLock lock = bucketCreationLock.readLock();
        lock.lock(); // prevent destruction while creating buckets
        try {
          // This counter is used by canAccomodateAnotherBucket to estimate if this member should
          // accept another bucket
          bucketCreatesInProgress.incrementAndGet();
          if (partitionedRegion.isDestroyed()) {
            return CreateBucketResult.FAILED;
          }

          // final boolean needsAllocation;
          if (isDebugEnabled) {
            logger.debug("grabFreeBucket: node list {} for bucketId={}{}{}",
                PartitionedRegionHelper.printCollection(partitionedRegion.getRegionAdvisor()
                    .getBucketOwners(possiblyFreeBucketId)),
                partitionedRegion.getPRId(), PartitionedRegion.BUCKET_ID_SEPARATOR,
                possiblyFreeBucketId);
          }


          // Final accommodation check under synchronization for
          // a stable view of bucket creation
          if (!forceCreation) {
            // Currently the balancing check should only run when creating buckets,
            // as opposed to during bucket recovery... it is assumed that
            // ifRedudnancyNotSatisfied is false during bucket recovery
            if (!canAccommodateAnotherBucket()) {
              result = CreateBucketResult.FAILED;
              return result;
            }
          }

          ProxyBucketRegion buk =
              partitionedRegion.getRegionAdvisor().getProxyBucketArray()[possiblyFreeBucketId];
          // Prevent other threads on the same VM from creating this bucket.
          // It doesn't look the the redundancy lock actually correctly
          // handles multiple threads, and the isManagingBucket call
          // also needs to be done under this lock
          synchronized (buk) {
            // DAN - this just needs to be done holding a lock for this particular bucket
            if (!verifyBucketBeforeGrabbing(possiblyFreeBucketId)) {
              result = CreateBucketResult.FAILED;
              return result;
            }

            if (!isManagingBucket(possiblyFreeBucketId)) {
              Integer possiblyFreeBucketIdInt = possiblyFreeBucketId;

              BucketRegion bukReg = null;
              Object redundancyLock =
                  lockRedundancyLock(moveSource, possiblyFreeBucketId, replaceOffineData);
              // DAN - I hope this is ok to do without that bucket admin lock
              try {
                buk.initializePrimaryElector(creationRequestor);
                if (getPartitionedRegion().getColocatedWith() == null) {
                  buk.getBucketAdvisor().markAllShadowBucketsAsNonDestroyed();
                }
                if (getPartitionedRegion().isShadowPR()) {
                  getPartitionedRegion().getColocatedWithRegion().getRegionAdvisor()
                      .getBucketAdvisor(possiblyFreeBucketId).markAllShadowBucketsAsNonDestroyed();
                }
                bukReg = createBucketRegion(possiblyFreeBucketId);
                // Mark the bucket as hosting and distribute to peers
                // before we release the dlock. This makes sure that our peers
                // won't think they need to satisfy redundancy
                if (bukReg != null) {
                  // Let the data store know about the real bucket at this point
                  // so that other VMs which discover the real bucket via a
                  // profile exchange can send messages to the data store and
                  // safely use the bucket.
                  observer.beforeAssignBucket(partitionedRegion, possiblyFreeBucketId);
                  assignBucketRegion(bukReg.getId(), bukReg);
                  buk.setHosting(true);
                  bukReg.invokePartitionListenerAfterBucketCreated();
                } else {
                  if (buk.getPartitionedRegion().getColocatedWith() == null) {
                    buk.getBucketAdvisor().markAllShadowBucketsAsDestroyed();
                    // clear tempQueue for all the shadowPR buckets
                    clearAllTempQueueForShadowPR(buk.getBucketId());
                  }
                }
              } finally {
                releaseRedundancyLock(redundancyLock);
                if (bukReg == null) {
                  buk.clearPrimaryElector();
                }
              }

              if (bukReg != null) {
                if (isDebugEnabled) {
                  logger.debug("grabFreeBucket: mapped bucketId={}{}{} on node = {}",
                      partitionedRegion.getPRId(), PartitionedRegion.BUCKET_ID_SEPARATOR,
                      possiblyFreeBucketId, partitionedRegion.getMyId());
                }

                createdBucket = true;
                result = CreateBucketResult.CREATED;
              } else {
                Assert.assertTrue(localBucket2RegionMap.get(possiblyFreeBucketIdInt) == null);
                result = CreateBucketResult.FAILED;
              }
            } else {
              // Handle the case where another concurrent thread decided to manage
              // the bucket and the creator may have died
              if (isDebugEnabled) {
                logger.debug("grabFreeBucket: bucketId={}{}{} already mapped on VM = {}",
                    partitionedRegion.getPRId(), PartitionedRegion.BUCKET_ID_SEPARATOR,
                    possiblyFreeBucketId, partitionedRegion.getMyId());
              }
              result = CreateBucketResult.ALREADY_EXISTS;
            }
            if (isDebugEnabled) {
              logger.debug("grabFreeBucket: Mapped bucketId={}{}{}",
                  partitionedRegion.getPRId(), PartitionedRegion.BUCKET_ID_SEPARATOR,
                  possiblyFreeBucketId);
            }
          }
        } catch (RegionDestroyedException rde) {
          RegionDestroyedException rde2 =
              new RegionDestroyedException(toString(), partitionedRegion.getFullPath(), rde);
          throw rde2;
        } catch (RedundancyAlreadyMetException e) {
          if (isDebugEnabled) {
            logger.debug("Redundancy already met {}{}{} assignment {}",
                partitionedRegion.getPRId(), PartitionedRegion.BUCKET_ID_SEPARATOR,
                possiblyFreeBucketId,
                localBucket2RegionMap.get(possiblyFreeBucketId));
          }
          result = CreateBucketResult.REDUNDANCY_ALREADY_SATISFIED;
        } finally {
          bucketCreatesInProgress.decrementAndGet();
          lock.unlock(); // prevent destruction while creating buckets
        }
      } finally {
        if (parentBucketCreationLock != null) {
          parentBucketCreationLock.unlock();
        }
      }
      partitionedRegion.checkReadiness();
      partitionedRegion.checkClosed();

      return result;

    } catch (PartitionOfflineException validationException) {
      // GEODE-3055
      PartitionedRegion leader = ColocationHelper.getLeaderRegion(partitionedRegion);
      boolean isLeader = leader.equals(partitionedRegion);
      if (!isLeader) {
        leader.getDataStore().removeBucket(possiblyFreeBucketId, true);
        if (isDebugEnabled) {
          logger.debug("For bucket " + possiblyFreeBucketId
              + ", failed to create cololcated child bucket for "
              + partitionedRegion.getFullPath() + ", removed leader region "
              + leader.getFullPath() + " bucket.");
        }
      }
      throw validationException;
    } finally {
      partitionedRegion.getPrStats().endBucketCreate(startTime, createdBucket, isRebalance);
    }
  }
Attempts to map a bucket id to this node. Creates real storage for the bucket by adding a new
Region to bucket2Map. Bucket creation is done under the d-lock on b2n region.

@param possiblyFreeBucketId the identity of the bucket + @param mustBeNew boolean enforcing
       that the bucket must not already exist
@param sender the member requesting the bucket
@param moveSource Where we are moving the bucket from, if this is a move.
@param forceCreation avoid any checks (with in reason) which might prevent bucket creation
@param isRebalance true if bucket creation is directed by rebalancing
@return true if successful
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionDataStore.java