HDFS-15416. Improve DataStorage#addStorageLocations() for empty locations. Contibuted by jianghua zhu.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/datanode/DataStorage.java

@@ -388,6 +388,11 @@ public class DataStorage extends Storage {
     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 {

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/datanode/TestDataStorage.java

@@ -44,6 +44,7 @@ import static org.junit.Assert.fail;
 public class TestDataStorage {
   private final static String DEFAULT_BPID = "bp-0";
   private final static String CLUSTER_ID = "cluster0";
+  private final static String CLUSTER_ID2 = "cluster1";
   private final static String BUILD_VERSION = "2.0";
   private final static String SOFTWARE_VERSION = "2.0";
   private final static long CTIME = 1;
@@ -165,6 +166,33 @@ public class TestDataStorage {
     assertEquals(6, storage.getNumStorageDirs());
   }
 
+  @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();
+  }
+
   @Test
   public void testMissingVersion() throws IOException,
       URISyntaxException {