HDFS-15416. DataStorage#addStorageLocations() should add more reasonable information verification.

Author: zhujianghua

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

@@ -388,6 +388,11 @@ synchronized List<StorageDirectory> addStorageLocations(DataNode datanode,
     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 @@
 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,35 @@ public void testAddStorageDirectories() throws IOException,
     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 {