void start(boolean shouldRemoveUnexpectedDirs) throws InterruptedException {
    long startTimeMs = time.milliseconds();
    final AtomicInteger numStoreFailures = new AtomicInteger(0);
    rwLock.readLock().lock();
    try {
      checkMountPathAccessible();
      // Report unrecognized directories before trying to satisfy disk space requirements.
      reportUnrecognizedDirs();
      if (shouldRemoveUnexpectedDirs) {
        // Try to remove all the unexpected directories before starting any blob store.
        // If the disk is near full because of the unexpected directoires, then blob store
        // might not be able to start since each blob store would request at least one log
        // segment.
        tryRemoveAllUnexpectedDirs();
      }

      ConcurrentHashMap<PartitionId, Exception> startExceptions = new ConcurrentHashMap<>();
      List<Thread> startupThreads = new ArrayList<>();
      for (final Map.Entry<PartitionId, BlobStore> partitionAndStore : stores.entrySet()) {
        if (stoppedReplicas.contains(partitionAndStore.getKey().toPathString())) {
          logger.info("Skip the store {} because it is on the stopped list", partitionAndStore.getKey());
          continue;
        }
        Thread thread = Utils.newThread("store-startup-" + partitionAndStore.getKey(), () -> {
          try {
            partitionAndStore.getValue().start();
          } catch (Exception e) {
            numStoreFailures.incrementAndGet();
            logger.error("Exception while starting store for the {}", partitionAndStore.getKey(), e);
            startExceptions.put(partitionAndStore.getKey(), e);
          }
        }, false);
        thread.start();
        startupThreads.add(thread);
      }
      for (Thread startupThread : startupThreads) {
        startupThread.join();
      }
      if (numStoreFailures.get() > 0) {
        logger.error("Could not start {} out of {} stores on the disk {}", numStoreFailures.get(), stores.size(), disk);
        maybeRecoverBlobStores(numStoreFailures.get(), startExceptions);
      }

      // DiskSpaceAllocator startup. This happens after BlobStore startup because it needs disk space requirements
      // from each store.
      List<DiskSpaceRequirements> requirementsList = new ArrayList<>();
      for (BlobStore blobStore : stores.values()) {
        if (blobStore.isStarted()) {
          DiskSpaceRequirements requirements = blobStore.getDiskSpaceRequirements();
          if (requirements != null) {
            requirementsList.add(requirements);
          }
        }
      }
      diskSpaceAllocator.initializePool(requirementsList);
      compactionManager.enable();
      bootstrapSessionManager.enable();
      running = true;
      if (diskHealthCheck.isEnabled()) {
        logger.info("Starting Disk Healthchecker");
        scheduler.scheduleAtFixedRate(() -> diskHealthCheck.diskHealthTest(), 0,
            diskManagerConfig.diskManagerDiskHealthCheckIntervalSeconds, TimeUnit.SECONDS);
      }
    } catch (StoreException e) {
      logger.error("Error while starting the DiskManager for {} ; no stores will be accessible on this disk.",
          disk.getMountPath(), e);
    } finally {
      if (!running) {
        metrics.totalStoreStartFailures.inc(stores.size());
        metrics.diskDownCount.inc();
      } else {
        metrics.totalStoreStartFailures.inc(numStoreFailures.get());
      }
      metrics.diskStartTimeMs.update(time.milliseconds() - startTimeMs);
      rwLock.readLock().unlock();
    }
  }
Starts all the stores on this disk.
@throws InterruptedException
ambry-store/src/main/java/com/github/ambry/store/DiskManager.java