HDFS-5417. Fix storage IDs in PBHelper and UpgradeUtilities.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/HDFS-2832@1535791 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES_HDFS-2832.txt

@@ -47,3 +47,5 @@ IMPROVEMENTS:
     (Arpit Agarwal)
 
     HDFS-5401. Fix NPE in Directory Scanner. (Arpit Agarwal)
+
+    HDFS-5417. Fix storage IDs in PBHelper and UpgradeUtilities.  (szetszwo)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocolPB/PBHelper.java

@@ -581,7 +581,10 @@ public class PBHelper {
         builder.addStorageTypes(PBHelper.convertStorageType(storageTypes[i]));
       }
     }
-    builder.addAllStorageIDs(Arrays.asList(b.getStorageIDs()));
+    final String[] storageIDs = b.getStorageIDs();
+    if (storageIDs != null) {
+      builder.addAllStorageIDs(Arrays.asList(b.getStorageIDs()));
+    }
 
     return builder.setB(PBHelper.convert(b.getBlock()))
         .setBlockToken(PBHelper.convert(b.getBlockToken()))
@@ -609,9 +612,11 @@ public class PBHelper {
       }
     }
 
+    final int storageIDsCount = proto.getStorageIDsCount();
+    final String[] storageIDs = storageIDsCount == 0? null
+        : proto.getStorageIDsList().toArray(new String[storageIDsCount]);
     LocatedBlock lb = new LocatedBlock(PBHelper.convert(proto.getB()), targets,
-        proto.getStorageIDsList().toArray(new String[proto.getStorageIDsCount()]),
-        storageTypes, proto.getOffset(), proto.getCorrupt());
+        storageIDs, storageTypes, proto.getOffset(), proto.getCorrupt());
     lb.setBlockToken(PBHelper.convert(proto.getBlockToken()));
     return lb;
   }

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/UpgradeUtilities.java

@@ -459,6 +459,7 @@ public class UpgradeUtilities {
     for (int i = 0; i < parent.length; i++) {
       File versionFile = new File(parent[i], "VERSION");
       StorageDirectory sd = new StorageDirectory(parent[i].getParentFile());
+      storage.createStorageID(sd);
       storage.writeProperties(versionFile, sd);
       versionFiles[i] = versionFile;
       File bpDir = BlockPoolSliceStorage.getBpRoot(bpid, parent[i]);