List<PartitionId> setBlobStoreStoppedState(List<PartitionId> partitionIds, boolean markStop) {
    Set<PartitionId> failToUpdateStores = new HashSet<>();
    List<ReplicaId> replicasToUpdate = new ArrayList<>();
    rwLock.readLock().lock();
    try {
      for (PartitionId id : partitionIds) {
        BlobStore store = stores.get(id);
        if (store == null) {
          // no need to check if the store is started because this method could be called after store is successfully shutdown.
          logger.error("store is not found on this disk when trying to update stoppedReplicas list");
          failToUpdateStores.add(id);
        } else {
          replicasToUpdate.add(partitionToReplicaMap.get(id));
        }
      }
    } finally {
      rwLock.readLock().unlock();
    }
    boolean updated = true;
    if (replicaStatusDelegates != null && !replicaStatusDelegates.isEmpty()) {
      logger.trace("Setting replica stopped state via ReplicaStatusDelegate on replica {}",
          Arrays.toString(replicasToUpdate.toArray()));
      for (ReplicaStatusDelegate replicaStatusDelegate : replicaStatusDelegates) {
        updated &= markStop ? replicaStatusDelegate.markStopped(replicasToUpdate)
            : replicaStatusDelegate.unmarkStopped(replicasToUpdate);
      }
    } else {
      logger.warn("The ReplicaStatusDelegate is not instantiated");
      updated = false;
    }
    if (!updated) {
      // either mark/unmark operation fails or ReplicaStatusDelegate is not instantiated.
      failToUpdateStores.addAll(partitionIds);
    }
    return new ArrayList<>(failToUpdateStores);
  }
Set the BlobStore stopped state with given {@link PartitionId} {@code id}.
@param partitionIds a list of {@link PartitionId} of the {@link BlobStore} whose stopped state should be set.
@param markStop whether to mark BlobStore as stopped ({@code true}) or started.
@return a list of {@link PartitionId} whose stopped state fails to be updated.
ambry-store/src/main/java/com/github/ambry/store/DiskManager.java