public void start() throws InterruptedException, StoreException {
    long startTimeMs = time.milliseconds();
    try {
      logger.info("Starting storage manager");
      List<Thread> startupThreads = new ArrayList<>();
      for (final DiskManager diskManager : diskToDiskManager.values()) {
        Thread thread = Utils.newThread("disk-manager-startup-" + diskManager.getDisk(), () -> {
          try {
            diskManager.start(
                storeConfig.storeRemoveUnexpectedDirsInFullAuto && clusterMap.isDataNodeInFullAutoMode(currentNode));
          } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            logger.error("Disk manager startup thread interrupted for disk {}", diskManager.getDisk(), e);
          }
        }, false);
        thread.start();
        startupThreads.add(thread);
      }
      for (Thread startupThread : startupThreads) {
        startupThread.join();
      }
      // Verify disk health before moving on.
      verifyDiskHealth();
      metrics.initializeCompactionThreadsTracker(this, diskToDiskManager.size());
      metrics.initializeHostUtilizationTracker(this);
      if (clusterParticipants != null) {
        clusterParticipants.forEach(clusterParticipant -> {
          clusterParticipant.registerPartitionStateChangeListener(StateModelListenerType.StorageManagerListener,
              new PartitionStateChangeListenerImpl(clusterParticipant == primaryClusterParticipant));
          clusterParticipant.setInitialLocalPartitions(partitionNameToReplicaId.keySet());
        });
      }
      diskToDiskManager.values().forEach(diskManager -> unexpectedDirs.addAll(diskManager.getUnexpectedDirs()));

      // Add the background task to update the disk capacity
      if (storeConfig.storeDiskFailureHandlerEnabled) {
        diskFailureHandler = new DiskFailureHandler();
        scheduler.scheduleAtFixedRate(diskFailureHandler, storeConfig.storeDiskFailureHandlerTaskIntervalInSeconds,
            storeConfig.storeDiskFailureHandlerTaskIntervalInSeconds, TimeUnit.SECONDS);
        metrics.initializeFailedDiskCount(this);
      }
      logger.info("Starting storage manager complete");
    } finally {
      metrics.storageManagerStartTimeMs.update(time.milliseconds() - startTimeMs);
    }
  }
Start the {@link DiskManager}s for all disks on this node.
@throws InterruptedException
ambry-store/src/main/java/com/github/ambry/store/StorageManager.java