public boolean removeBucket(int bucketId, boolean forceRemovePrimary) {
    waitForInProgressBackup();

    // Don't allow the removal of a bucket if we haven't
    // finished recovering from disk
    if (!forceRemovePrimary
        && !partitionedRegion.getRedundancyProvider().isPersistentRecoveryComplete()) {
      if (logger.isDebugEnabled()) {
        logger.debug(
            "Returning false from removeBucket because we have not finished recovering all colocated regions from disk");
      }
      return false;
    }
    // Lock out bucket creation while doing this :-)
    StoppableWriteLock lock = bucketCreationLock.writeLock();
    lock.lock();
    try {
      BucketRegion bucketRegion = localBucket2RegionMap.get(bucketId);
      if (bucketRegion == null) {
        if (logger.isDebugEnabled()) {
          logger.debug(
              "Returning true from removeBucket because we don't have the bucket we've been told to remove");
        }
        return true;

      }

      PartitionedRegion leader = ColocationHelper.getLeaderRegion(partitionedRegion);
      boolean isLeader = leader.equals(partitionedRegion);
      BucketAdvisor bucketAdvisor = bucketRegion.getBucketAdvisor();
      InternalDistributedMember myId =
          partitionedRegion.getDistributionManager().getDistributionManagerId();
      Lock primaryMoveReadLock = bucketAdvisor.getPrimaryMoveReadLock();

      // Fix for 43613 - don't remove the bucket
      // if we are primary. We hold the lock here
      // to prevent this member from becoming primary until this
      // member is no longer hosting the bucket.
      primaryMoveReadLock.lock();
      try {
        // forceRemovePrimary==true will enable remove the bucket even when:
        // 1) it's primary
        // 2) no other primary ready yet
        // 3) colocated bucket and its child is not completely ready
        if (!forceRemovePrimary && bucketAdvisor.isPrimary()) {
          return false;
        }

        if (isLeader) {
          if (!forceRemovePrimary && !isRemotePrimaryReadyForColocatedChildren(bucketId)) {
            return false;
          }

          InternalDistributedMember primary = bucketAdvisor.getPrimary();
          if (!forceRemovePrimary && (primary == null || myId.equals(primary))) {
            if (logger.isDebugEnabled()) {
              logger.debug("Bucket region " + bucketRegion
                  + " does not have a remote primary yet. Not to remove.");
            }
            return false;
          }

          if (logger.isDebugEnabled()) {
            logger.debug("Bucket region " + bucketRegion + " has primary at " + primary);
          }
        }

        // recurse down to each tier of children to remove first
        removeBucketForColocatedChildren(bucketId, forceRemovePrimary);

        if (bucketRegion.getPartitionedRegion().isShadowPR()) {
          if (bucketRegion.getPartitionedRegion().getColocatedWithRegion() != null) {
            bucketRegion.getPartitionedRegion().getColocatedWithRegion().getRegionAdvisor()
                .getBucketAdvisor(bucketId).markAllShadowBucketsAsDestroyed();
          }
        }
        bucketAdvisor.getProxyBucketRegion().removeBucket();
      } finally {
        primaryMoveReadLock.unlock();
      }

      if (logger.isDebugEnabled()) {
        logger.debug("Removed bucket {} from advisor", bucketRegion);
      }

      // Flush the state of the primary. This make sure we have processed
      // all operations were sent out before we removed our profile from
      // our peers.
      //
      // Another option, instead of using the StateFlushOperation, could
      // be to send a message which waits until it acquires the
      // activePrimaryMoveLock on primary the bucket region. That would also
      // wait for in progress writes. I choose to use the StateFlushOperation
      // because it won't block write operations while we're trying to acquire
      // the activePrimaryMoveLock
      InternalDistributedMember primary = bucketAdvisor.getPrimary();
      if (!myId.equals(primary)) {
        StateFlushOperation flush = new StateFlushOperation(bucketRegion);
        int executor = OperationExecutors.WAITING_POOL_EXECUTOR;
        try {
          flush.flush(Collections.singleton(primary), myId, executor, false);
        } catch (InterruptedException e) {
          partitionedRegion.getCancelCriterion().checkCancelInProgress(e);
          Thread.currentThread().interrupt();
          throw new InternalGemFireException("Interrupted while flushing state");
        }
        if (logger.isDebugEnabled()) {
          logger.debug("Finished state flush for removal of {}", bucketRegion);
        }
      } else {
        if (logger.isDebugEnabled()) {
          logger.debug("We became primary while destroying the bucket. Too late to stop now.");
        }
      }
      bucketRegion.invokePartitionListenerAfterBucketRemoved();
      bucketRegion.preDestroyBucket(bucketId);
      bucketRegion.localDestroyRegion();
      bucketAdvisor.getProxyBucketRegion().finishRemoveBucket();
      if (logger.isDebugEnabled()) {
        logger.debug("Destroyed {}", bucketRegion);
      }
      localBucket2RegionMap.remove(bucketId);
      partitionedRegion.getPrStats().incBucketCount(-1);
      return true;
    } finally {
      lock.unlock();
    }
  }
Removes a redundant bucket hosted by this data store. The rebalancer invokes this method
directly or sends this member a message to invoke it.

From the spec:

How to Remove a Redundant Bucket

This operation is done by the rebalancer (REB) and can only be done on non-primary buckets. If
you want to remove a primary bucket first send one of its peers "become primary" and then send
it "unhost" (we could offer a "unhost" option on "become primary" or a "becomePrimary" option
on "create redundant"). The member that has the bucket being removed is called the bucket host
(BH).

1. REB sends an "unhostBucket" message to BH. This message will be rejected if the member finds
itself to be the primary or if it doesn't host the bucket by sending a failure reply to REB. 2.
BH marks itself as "not-hosting". This causes any read operations that come in to not start and
retry. BH also updates the advisor to know that it is no longer hosting the bucket. 3. BH then
waits for any in-progress reads (which read ops to wait for are TBD) to complete. 4. BH then
removes the bucket region from its cache. 5. BH then sends a success reply to REB.

This method is now also used by the PartitionManager. For the PartitionManager, it does remove
the primary bucket.

@param bucketId the id of the bucket to remove

@return true if the bucket was removed; false if unable to remove or if bucket is not hosted
geode-core/src/main/java/org/apache/geode/internal/cache/PartitionedRegionDataStore.java