HDFS-13703. Avoid allocation of CorruptedBlocks hashmap when no corrupted blocks are hit. Contributed by Todd Lipcon.

(cherry picked from commit 6ba99741086170b83c38d3e7e715d9e8046a1e00)

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSInputStream.java

@@ -1419,7 +1419,7 @@ public class DFSInputStream extends FSInputStream
 
     Map<ExtendedBlock, Set<DatanodeInfo>> corruptedBlockMap =
         corruptedBlocks.getCorruptionMap();
-    if (corruptedBlockMap.isEmpty()) {
+    if (corruptedBlockMap == null) {
       return;
     }
     List<LocatedBlock> reportList = new ArrayList<>(corruptedBlockMap.size());

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSUtilClient.java

@@ -751,14 +751,14 @@ public class DFSUtilClient {
   public static class CorruptedBlocks {
     private Map<ExtendedBlock, Set<DatanodeInfo>> corruptionMap;
 
-    public CorruptedBlocks() {
-      this.corruptionMap = new HashMap<>();
-    }
-
     /**
      * Indicate a block replica on the specified datanode is corrupted
      */
     public void addCorruptedBlock(ExtendedBlock blk, DatanodeInfo node) {
+      if (corruptionMap == null) {
+        corruptionMap = new HashMap<>();
+      }
+
       Set<DatanodeInfo> dnSet = corruptionMap.get(blk);
       if (dnSet == null) {
         dnSet = new HashSet<>();
@@ -770,7 +770,8 @@ public class DFSUtilClient {
     }
 
     /**
-     * @return the map that contains all the corruption entries.
+     * @return the map that contains all the corruption entries, or null if
+     * there were no corrupted entries
      */
     public Map<ExtendedBlock, Set<DatanodeInfo>> getCorruptionMap() {
       return corruptionMap;

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

@@ -1278,7 +1278,7 @@ public class DataNode extends ReconfigurableBase
       DFSUtilClient.CorruptedBlocks corruptedBlocks) throws IOException {
     Map<ExtendedBlock, Set<DatanodeInfo>> corruptionMap =
         corruptedBlocks.getCorruptionMap();
-    if (!corruptionMap.isEmpty()) {
+    if (corruptionMap != null) {
       for (Map.Entry<ExtendedBlock, Set<DatanodeInfo>> entry :
           corruptionMap.entrySet()) {
         for (DatanodeInfo dnInfo : entry.getValue()) {