private ServerErrorCode handleStopStoreRequest(PartitionId partitionId, short numReplicasCaughtUpPerPartition) {
    ServerErrorCode error = validateRequest(partitionId, RequestOrResponseType.AdminRequest, false);
    if (!error.equals(ServerErrorCode.NoError)) {
      logger.debug("Validate request fails for {} with error code {} when trying to stop store", partitionId, error);
      return error;
    }
    if (numReplicasCaughtUpPerPartition < 0) {
      logger.debug("The number of replicas to catch up should not be less than zero {}",
          numReplicasCaughtUpPerPartition);
      return ServerErrorCode.BadRequest;
    }
    if (!storeManager.controlCompactionForBlobStore(partitionId, false)) {
      logger.error("Disable compaction fails on given BlobStore {}", partitionId);
      return ServerErrorCode.UnknownError;
    }
    Collection<PartitionId> partitionIds = Collections.singletonList(partitionId);
    controlRequestForPartitions(EnumSet.of(RequestOrResponseType.PutRequest, RequestOrResponseType.DeleteRequest,
        RequestOrResponseType.TtlUpdateRequest), partitionIds, false);
    if (!replicationEngine.controlReplicationForPartitions(partitionIds, Collections.<String>emptyList(), false)) {
      logger.error("Could not disable replication on {}", partitionIds);
      return ServerErrorCode.UnknownError;
    }
    if (!isRemoteLagLesserOrEqual(partitionIds, 0, numReplicasCaughtUpPerPartition)) {
      logger.debug("Catchup not done on {}", partitionIds);
      return RetryAfterBackoff;
    }
    controlRequestForPartitions(
        EnumSet.of(RequestOrResponseType.ReplicaMetadataRequest, RequestOrResponseType.GetRequest), partitionIds,
        false);
    // Shutdown the BlobStore completely
    if (storeManager.shutdownBlobStore(partitionId)) {
      logger.info("Store is successfully shutdown for partition: {}", partitionId);
      List<PartitionId> failToUpdateList =
          storeManager.setBlobStoreStoppedState(Collections.singletonList(partitionId), true);
      if (!failToUpdateList.isEmpty() && clusterParticipant != null) {
        logger.error("Fail to add BlobStore(s) {} to stopped list after stop operation completed",
            failToUpdateList.toArray());
        error = ServerErrorCode.UnknownError;
      }
      // After store is shut down and stopped state is updated, we also need to temporarily disable this replica if Helix
      // is adopted. The intention here is to force Helix to re-elect a new leader replica if necessary.
      if (storeManager instanceof StorageManager && clusterParticipant != null) {
        // Previous operation has guaranteed the store is not null
        Store store = ((StorageManager) storeManager).getStore(partitionId, true);
        // Setting disable state will trigger transition error at the very beginning of STANDBY -> INACTIVE transition.
        // Thus it doesn't go through decommission process.
        ((BlobStore) store).setDisableState(true);
        clusterParticipant.setReplicaDisabledState(storeManager.getReplica(partitionId.toPathString()), true);
      }
    } else {
      error = ServerErrorCode.UnknownError;
      logger.error("Shutting down BlobStore fails on {}", partitionId);
    }
    return error;
  }
Handles admin request that stops BlobStore
@param partitionId the {@link PartitionId} associated with BlobStore
@param numReplicasCaughtUpPerPartition the minimum number of peer replicas per partition that should catch up with
                                       local store before stopping it.
@return {@link ServerErrorCode} represents result of handling admin request.
ambry-server/src/main/java/com/github/ambry/server/AmbryServerRequests.java