HADOOP-3776. Fix NPE at NameNode when datanode reports a block after it is

deleted at NameNode. (rangadi)


git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.18@679461 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -779,6 +779,9 @@ Release 0.18.0 - Unreleased
 
     HADOOP-3806. Remove debug statement to stdout from QuickSort. (cdouglas)
 
+    HADOOP-3776. Fix NPE at NameNode when datanode reports a block after it is
+    deleted at NameNode. (rangadi)
+
 Release 0.17.2 - Unreleased
 
   BUG FIXES

src/hdfs/org/apache/hadoop/dfs/FSNamesystem.java

@@ -2707,15 +2707,21 @@ class FSNamesystem implements FSConstants, FSNamesystemMBean {
                                     DatanodeDescriptor node,
                                     DatanodeDescriptor delNodeHint) {
     BlockInfo storedBlock = blocksMap.getStoredBlock(block);
-    boolean added = false;
-    if(storedBlock == null) { // block is not in the blocksMaps
-      // add block to the blocksMap and to the data-node
-      added = blocksMap.addNode(block, node, defaultReplication);
-      storedBlock = blocksMap.getStoredBlock(block);
-    } else {
-      // add block to the data-node
-      added = node.addBlock(storedBlock);
-    }
+    if(storedBlock == null || storedBlock.getINode() == null) {
+      // If this block does not belong to anyfile, then we are done.
+      NameNode.stateChangeLog.info("BLOCK* NameSystem.addStoredBlock: "
+                                   + "addStoredBlock request received for " 
+                                   + block + " on " + node.getName()
+                                   + " size " + block.getNumBytes()
+                                   + " But it does not belong to any file.");
+      // we could add this block to invalidate set of this datanode. 
+      // it will happen in next block report otherwise.
+      return block;      
+    }
+     
+    // add block to the data-node
+    boolean added = node.addBlock(storedBlock);
+    
     assert storedBlock != null : "Block must be stored by now";
 
     if (block != storedBlock) {
@@ -2793,8 +2799,6 @@ class FSNamesystem implements FSConstants, FSNamesystemMBean {
                                    + " size " + block.getNumBytes());
     }
 
-    assert isValidBlock(storedBlock) : "Trying to add an invalid block";
-
     // filter out containingNodes that are marked for decommission.
     NumberReplicas num = countNodes(storedBlock);
     int numLiveReplicas = num.liveReplicas();