@Override
    public void onPartitionBecomeDroppedFromOffline(String partitionName) {
      ReplicaId replica = partitionNameToReplicaId.get(partitionName);
      if (replica == null) {
        // During decommission, imagine an edge case where the node crashed immediately after removing replica from
        // InstanceConfig. So next time when this node is restarted, old replica is no longer present in Helix cluster
        // map. We still need to clean up file/directory on disk that is associated with removed replica.

        // There is another case where the replica would be null, which is when the disk failed, we removed all the
        // replicas on the disk. We can try to remove the diretory, but it won't find any.
        try {
          maybeDeleteResidualDirectory(partitionName);
        } catch (IOException e) {
          logger.error("Exception occurs when deleting residual dir of replica {} with error msg: {}", partitionName,
              e.getMessage());
          metrics.resumeDecommissionErrorCount.inc();
          throw new StateTransitionException("Failed to delete residual dir of store " + partitionName,
              ReplicaOperationFailure);
        }
        return;
      }
      Map<StateModelListenerType, PartitionStateChangeListener> partitionStateChangeListeners =
          primaryClusterParticipant == null ? new HashMap<>()
              : primaryClusterParticipant.getPartitionStateChangeListeners();
      replicationManagerListener = partitionStateChangeListeners.get(StateModelListenerType.ReplicationManagerListener);
      statsManagerListener = partitionStateChangeListeners.get(StateModelListenerType.StatsManagerListener);
      // get the store (skip the state check here, because probably the store is stopped in previous transition. Also,
      // the store could be started if it failed on decommission last time. Helix may directly reset it to OFFLINE
      // without stopping it)
      BlobStore store = (BlobStore) getStore(replica.getPartitionId(), true);
      if (store == null && isReplicaOnFailedDisk(replica)) {
        logger.info("Replica is in a failed disk, blob store not started. skip");
      } else {
        //1. Check if the store should resume decommission
        if (shouldResumeDecommission(store, replica)) {
          try {
            resumeDecommission(partitionName);
          } catch (Exception e) {
            logger.error("Exception occurs when resuming decommission on replica {} with error msg: {}", replica,
                e.getMessage());
            metrics.resumeDecommissionErrorCount.inc();
            throw new StateTransitionException(
                "Exception occurred when resuming decommission on replica " + partitionName, ReplicaOperationFailure);
          }
        }
        // 2. Shut down the store
        if (!shutdownBlobStore(replica.getPartitionId())) {
          throw new StateTransitionException("Failed to shutdown store " + partitionName, ReplicaOperationFailure);
        }
        logger.info("Store {} is successfully shut down during Offline-To-Dropped transition", partitionName);
      }
      // 3. Remove replica from data node configs
      if (primaryClusterParticipant != null) {
        try {
          if (!primaryClusterParticipant.updateDataNodeInfoInCluster(replica, false)) {
            logger.error("Failed to remove partition {} from DataNodeConfig of current node", partitionName);
            throw new StateTransitionException("Failed to remove partition " + partitionName + " from DataNodeConfig",
                StateTransitionException.TransitionErrorCode.HelixUpdateFailure);
          }
          logger.info("Partition {} is successfully removed from DataNodeConfig of current node", partitionName);
        } catch (IllegalStateException e) {
          throw new StateTransitionException(e.getMessage(),
              StateTransitionException.TransitionErrorCode.HelixUpdateFailure);
        }
      }

      // 4. invoke PartitionStateChangeListener in Replication Manager and Stats Manager to remove replica
      logger.info("Invoking state listeners to remove replica {} from stats and replication manager", partitionName);
      if (statsManagerListener != null) {
        statsManagerListener.onPartitionBecomeDroppedFromOffline(partitionName);
      }
      if (replicationManagerListener != null) {
        replicationManagerListener.onPartitionBecomeDroppedFromOffline(partitionName);
      }

      // 5. remove store and delete all files associated with given replica in Storage Manager
      try {
        if (!removeBlobStore(replica.getPartitionId())) {
          throw new StateTransitionException("Failed to remove store " + partitionName + " from storage manager",
              ReplicaOperationFailure);
        }
      } catch (StateTransitionException | IOException | StoreException e) {
        if (isReplicaOnFailedDisk(replica)) {
          logger.error("Failed to remove blob store for {}, but this is a a failed disk {} so ignore", partitionName,
              replica.getDiskId().getMountPath());
        } else {
          throw new StateTransitionException("Failed to delete directory for store " + partitionName,
              ReplicaOperationFailure);
        }
      }
      partitionNameToReplicaId.remove(partitionName);
      logger.info("Partition {} is successfully dropped on current node", partitionName);
    }
Constructor
@param isPrimaryClusterManagerListener Indicates whether it's a listener object for primary helix cluster-manager
ambry-store/src/main/java/com/github/ambry/store/StorageManager.java