@Override
    public void onPartitionBecomePreBootstrapFromOffline(String partitionName) {
      // check if partition exists on current node
      ReplicaId replica = partitionNameToReplicaId.get(partitionName);
      Store store;
      if (replica == null) {

        ReplicaId replicaToAdd;
        boolean replicaAdded = false;
        do {
          // there can be two scenarios:
          // 1. this is the first time to add new replica onto current node;
          // 2. last replica addition failed at some point before updating InstanceConfig in Helix
          // In either case, we should add replica to current node by calling "addBlobStore(ReplicaId replica)"
          replicaToAdd = clusterMap.getBootstrapReplica(partitionName, currentNode);
          if (replicaToAdd == null) {
            logger.error("No new replica found for partition {} in cluster map", partitionName);
            throw new StateTransitionException(
                "New replica " + partitionName + " is not found in clustermap for " + currentNode, ReplicaNotFound);
          }
          // Attempt to add store into storage manager. If store already exists on disk (but not in clustermap), make
          // sure old store of this replica is deleted (this store may be created in previous replica addition but failed
          // at some point). Then a brand new store associated with this replica should be created and started.
          if (!initializeBlobStore(replicaToAdd)) {
            // We have decreased the available disk space in HelixClusterManager#getDiskForBootstrapReplica. Increase it
            // back since addition of store failed.
            replicaToAdd.getDiskId().increaseAvailableSpaceInBytes(replicaToAdd.getCapacityInBytes());
            if (!clusterMap.isDataNodeInFullAutoMode(currentNode)) {
              logger.error("Failed to add store {} into storage manager", partitionName);
              throw new StateTransitionException("Failed to add store " + partitionName + " into storage manager",
                  ReplicaOperationFailure);
            } else {
              logger.info("Failed to add store {} at location {}. Retrying bootstrapping replica at different location",
                  partitionName, replicaToAdd.getReplicaPath());
              tryRemoveFailedBootstrapBlobStore(replicaToAdd);
            }
          } else {
            replicaAdded = true;
          }
        } while (!replicaAdded);

        // note that partitionNameToReplicaId should be updated if addBlobStore succeeds, so replicationManager should be
        // able to get new replica from storageManager without querying Helix

        if (primaryClusterParticipant != null) {
          // update InstanceConfig in Helix
          try {
            if (!primaryClusterParticipant.updateDataNodeInfoInCluster(replicaToAdd, true)) {
              logger.error("Failed to add partition {} into InstanceConfig of current node", partitionName);
              throw new StateTransitionException("Failed to add partition " + partitionName + " into InstanceConfig",
                  StateTransitionException.TransitionErrorCode.HelixUpdateFailure);
            }
            logger.info("Partition {} is successfully added into InstanceConfig of current node", partitionName);
          } catch (IllegalStateException e) {
            throw new StateTransitionException(e.getMessage(),
                StateTransitionException.TransitionErrorCode.HelixUpdateFailure);
          }
        }
      } else {
        // if the replica is already on current node, there are 4 cases need to discuss:
        // 1. replica was initially present in clustermap and this is a regular reboot.
        // 2. replica was dynamically added to this node but may fail during BOOTSTRAP -> STANDBY transition.
        // 3. replica is on current node but its disk is offline. The replica is not able to start.
        // 4. replica was initially present in clustermap but it's current being recreated due to disk failure before.

        // For case 1 and 2, OFFLINE -> BOOTSTRAP is complete, we leave remaining actions (if there any) to other transition.
        // For case 3, we should throw exception to make replica stay in ERROR state (thus, frontends won't pick this replica)
        // For case 4, we check it's current used capacity and put it in BOOTSTRAP state if necessary. This is to ensure
        //             it catches up with peers before serving PUT traffic (or being selected as LEADER)
        store = getStore(replica.getPartitionId(), false);

        // store should be in started if this is not a first time added replica
        // as we will start all stores on the host during a restart
        if (store == null) {
          throw new StateTransitionException(
              "Store " + partitionName + " didn't start correctly, replica should be set to ERROR state",
              StoreNotStarted);
        }

        File decommissionFile = new File(replica.getReplicaPath(), BlobStore.DECOMMISSION_FILE_NAME);
        if (decommissionFile.exists()) {
          // Delete any decommission file if present.
          // During migration from semi-auto to full-auto, we observed that helix could issue state transitions from
          // Standby -> Inactive -> Offline -> Bootstrap -> Standby. We would have created decommission file during
          // Standby -> Inactive step of this process. Delete this file now since the same replica is being bootstrapped
          // again.
          decommissionFile.delete();
          logger.info("Old decommission file is deleted for replica {}", replica.getReplicaPath());
          ((BlobStore) store).setRecoverFromDecommission(false);
        }

        // if store's used capacity is less than or equal to header size, we create a bootstrap_in_progress file and force
        // it to stay in BOOTSTRAP state when catching up with peers.
        long storeUsedCapacity = store.getSizeInBytes();
        if (storeUsedCapacity <= HEADER_SIZE) {
          logger.info(
              "Store {} has used capacity {} less than or equal to {} bytes, consider it recently created and make it go through bootstrap process.",
              partitionName, storeUsedCapacity, HEADER_SIZE);
          try {
            createBootstrapFileIfAbsent(replica);
          } catch (IOException e) {
            logger.error("Failed to create bootstrap file for store {}", partitionName);
            throw new StateTransitionException("Failed to create bootstrap file for " + partitionName,
                ReplicaOperationFailure);
          }
        }
      }
    }
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