HADOOP-3724. Fixes two problems related to storing and recovering lease

in the fsimage. The unit tests are not merged into the 0.18 branch. (dhruba)
svn merge -c 679866 from trunk.




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

CHANGES.txt

@@ -768,6 +768,9 @@ Release 0.18.0 - Unreleased
     HADOOP-3521. Reverted the missing cast to float for sending Counters' values
     to Hadoop metrics which was removed by HADOOP-544. (acmurthy)   
 
+    HADOOP-3724. Fixes two problems related to storing and recovering lease
+    in the fsimage. (dhruba)
+
 Release 0.17.2 - Unreleased
 
   BUG FIXES

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

@@ -378,7 +378,14 @@ public class DatanodeDescriptor extends DatanodeInfo {
         continue;
       }
       if(storedBlock.findDatanode(this) < 0) {// Known block, but not on the DN
-        toAdd.add(storedBlock);
+        // if the size differs from what is in the blockmap, then return
+        // the new block. addStoredBlock will then pick up the right size of this
+        // block and will update the block object in the BlocksMap
+        if (storedBlock.getNumBytes() != iblk.getNumBytes()) {
+          toAdd.add(new Block(iblk));
+        } else {
+          toAdd.add(storedBlock);
+        }
         continue;
       }
       // move block to the head of the list

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

@@ -631,8 +631,11 @@ class FSDirectory implements FSConstants, Closeable {
                               "failed to remove " + path);
       } 
       rootDir.addNode(path, newnode); 
+      int index = 0;
       for (Block b : newnode.getBlocks()) {
-        namesystem.blocksMap.addINode(b, newnode);
+        BlockInfo info = namesystem.blocksMap.addINode(b, newnode);
+        newnode.setBlock(index, info); // inode refers to the block in BlocksMap
+        index++;
       }
     }
   }

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

@@ -34,6 +34,7 @@ import java.nio.channels.FileChannel;
 
 import org.apache.hadoop.io.*;
 import org.apache.hadoop.fs.permission.*;
+import org.apache.hadoop.dfs.DFSFileInfo;
 
 /**
  * FSEditLog maintains a log of the namespace modifications.
@@ -557,8 +558,9 @@ class FSEditLog {
             String s = FSImage.readString(in);
             String d = FSImage.readString(in);
             timestamp = readLong(in);
+            DFSFileInfo dinfo = fsDir.getFileInfo(d);
             fsDir.unprotectedRenameTo(s, d, timestamp);
-            fsNamesys.changeLease(s, d);
+            fsNamesys.changeLease(s, d, dinfo);
             break;
           }
           case OP_DELETE: {

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

@@ -1448,8 +1448,9 @@ class FSNamesystem implements FSConstants, FSNamesystemMBean {
       checkAncestorAccess(actualdst, FsAction.WRITE);
     }
 
+    DFSFileInfo dinfo = dir.getFileInfo(dst);
     if (dir.renameTo(src, dst)) {
-      changeLease(src, dst);     // update lease with new filename
+      changeLease(src, dst, dinfo);     // update lease with new filename
       return true;
     }
     return false;
@@ -4372,12 +4373,17 @@ class FSNamesystem implements FSConstants, FSNamesystemMBean {
   // rename was successful. If any part of the renamed subtree had
   // files that were being written to, update with new filename.
   //
-  void changeLease(String src, String dst) throws IOException {
+  void changeLease(String src, String dst, DFSFileInfo dinfo) 
+                   throws IOException {
     String overwrite;
     String replaceBy;
 
-    DFSFileInfo dinfo = dir.getFileInfo(dst);
-    if (dinfo.isDir()) {
+    boolean destinationExisted = true;
+    if (dinfo == null) {
+      destinationExisted = false;
+    }
+
+    if (destinationExisted && dinfo.isDir()) {
       Path spath = new Path(src);
       overwrite = spath.getParent().toString() + Path.SEPARATOR;
       replaceBy = dst + Path.SEPARATOR;