HDFS-5077. NPE in FSNamesystem.commitBlockSynchronization(). Contributed by Plamen Jeliazkov.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1518852 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -175,6 +175,9 @@ Release 2.1.1-beta - UNRELEASED
     HDFS-5132. Deadlock in NameNode between SafeModeMonitor#run and 
     DatanodeManager#handleHeartbeat. (kihwal)
 
+    HDFS-5077. NPE in FSNamesystem.commitBlockSynchronization().
+    (Plamen Jeliazkov via shv)
+
 Release 2.1.0-beta - 2013-08-22
 
   INCOMPATIBLE CHANGES

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -174,7 +174,6 @@ import org.apache.hadoop.hdfs.server.common.Util;
 import org.apache.hadoop.hdfs.server.namenode.INode.BlocksMapUpdateInfo;
 import org.apache.hadoop.hdfs.server.namenode.JournalSet.JournalAndStream;
 import org.apache.hadoop.hdfs.server.namenode.LeaseManager.Lease;
-import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.hdfs.server.namenode.NameNode.OperationCategory;
 import org.apache.hadoop.hdfs.server.namenode.startupprogress.Phase;
 import org.apache.hadoop.hdfs.server.namenode.startupprogress.StartupProgress;
@@ -3755,24 +3754,32 @@ public class FSNamesystem implements Namesystem, FSClusterStats,
         // find the DatanodeDescriptor objects
         // There should be no locations in the blockManager till now because the
         // file is underConstruction
-        DatanodeDescriptor[] descriptors = null;
+        List<DatanodeDescriptor> targetList =
+            new ArrayList<DatanodeDescriptor>(newtargets.length);
         if (newtargets.length > 0) {
-          descriptors = new DatanodeDescriptor[newtargets.length];
-          for(int i = 0; i < newtargets.length; i++) {
-            descriptors[i] = blockManager.getDatanodeManager().getDatanode(
-                newtargets[i]);
+          for (DatanodeID newtarget : newtargets) {
+            // try to get targetNode
+            DatanodeDescriptor targetNode =
+                blockManager.getDatanodeManager().getDatanode(newtarget);
+            if (targetNode != null)
+              targetList.add(targetNode);
+            else if (LOG.isDebugEnabled()) {
+              LOG.debug("DatanodeDescriptor (=" + newtarget + ") not found");
+            }
           }
         }
-        if ((closeFile) && (descriptors != null)) {
+        if ((closeFile) && !targetList.isEmpty()) {
           // the file is getting closed. Insert block locations into blockManager.
           // Otherwise fsck will report these blocks as MISSING, especially if the
           // blocksReceived from Datanodes take a long time to arrive.
-          for (int i = 0; i < descriptors.length; i++) {
-            descriptors[i].addBlock(storedBlock);
+          for (DatanodeDescriptor targetNode : targetList) {
+            targetNode.addBlock(storedBlock);
           }
         }
         // add pipeline locations into the INodeUnderConstruction
-        pendingFile.setLastBlock(storedBlock, descriptors);
+        DatanodeDescriptor[] targetArray =
+            new DatanodeDescriptor[targetList.size()];
+        pendingFile.setLastBlock(storedBlock, targetList.toArray(targetArray));
       }
 
       if (closeFile) {

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestCommitBlockSynchronization.java

@@ -169,4 +169,23 @@ public class TestCommitBlockSynchronization {
     namesystemSpy.commitBlockSynchronization(
         lastBlock, genStamp, length, true, false, newTargets, null);
   }
+
+  @Test
+  public void testCommitBlockSynchronizationWithCloseAndNonExistantTarget()
+      throws IOException {
+    INodeFileUnderConstruction file = mock(INodeFileUnderConstruction.class);
+    Block block = new Block(blockId, length, genStamp);
+    FSNamesystem namesystemSpy = makeNameSystemSpy(block, file);
+    DatanodeID[] newTargets = new DatanodeID[]{
+        new DatanodeID("0.0.0.0", "nonexistantHost", "1", 0, 0, 0)};
+
+    ExtendedBlock lastBlock = new ExtendedBlock();
+    namesystemSpy.commitBlockSynchronization(
+        lastBlock, genStamp, length, true,
+        false, newTargets, null);
+
+    // Repeat the call to make sure it returns true
+    namesystemSpy.commitBlockSynchronization(
+        lastBlock, genStamp, length, true, false, newTargets, null);
+  }
 }