@VisibleForTesting
  synchronized List<StorageDirectory> addStorageLocations(DataNode datanode,
      NamespaceInfo nsInfo, Collection<StorageLocation> dataDirs,
      StartupOption startOpt) throws IOException {
    final int numThreads = getParallelVolumeLoadThreadsNum(
        dataDirs.size(), datanode.getConf());
    final ExecutorService executor = Executors.newFixedThreadPool(numThreads);
    try {
      final List<StorageLocation> successLocations = loadDataStorage(
          datanode, nsInfo, dataDirs, startOpt, executor);

      if (successLocations.isEmpty()) {
        return Lists.newArrayList();
      }

      return loadBlockPoolSliceStorage(
          datanode, nsInfo, successLocations, startOpt, executor);
    } finally {
      executor.shutdown();
    }
  }
@Test
  public void testAddStorageDirectoriesFailure() throws IOException {
    final int numLocations = 1;
    List<StorageLocation> locations = createStorageLocations(numLocations);
    assertEquals(numLocations, locations.size());

    NamespaceInfo namespaceInfo = new NamespaceInfo(0, CLUSTER_ID,
        DEFAULT_BPID, CTIME, BUILD_VERSION, SOFTWARE_VERSION);
    List<StorageDirectory> successLocations = storage.addStorageLocations(
        mockDN, namespaceInfo, locations, START_OPT);
    assertEquals(1, successLocations.size());

    // After the DataNode restarts, the value of the clusterId is different
    // from the value before the restart.
    storage.unlockAll();
    DataNode newMockDN = Mockito.mock(DataNode.class);
    Mockito.when(newMockDN.getConf()).thenReturn(new HdfsConfiguration());
    DataStorage newStorage = new DataStorage();
    NamespaceInfo newNamespaceInfo = new NamespaceInfo(0, CLUSTER_ID2,
        DEFAULT_BPID, CTIME, BUILD_VERSION, SOFTWARE_VERSION);
    successLocations = newStorage.addStorageLocations(
            newMockDN, newNamespaceInfo, locations, START_OPT);
    assertEquals(0, successLocations.size());
    newStorage.unlockAll();
    newMockDN.shutdown();
  }