HDFS-10763. Open files can leak permanently due to inconsistent lease update. Contributed by Kihwal Lee.

(cherry picked from commit 864f878d5912c82f3204f1582cfb7eb7c9f1a1da)
(cherry picked from commit e78db7d2a430983807750666fb72ebd5c97ce867)

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

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

@@ -278,18 +278,14 @@ public final class FSImageFormatPBINode {
      * Load the under-construction files section, and update the lease map
      */
     void loadFilesUnderConstructionSection(InputStream in) throws IOException {
+      // Leases are added when the inode section is loaded. This section is
+      // still read in for compatibility reasons.
       while (true) {
         FileUnderConstructionEntry entry = FileUnderConstructionEntry
             .parseDelimitedFrom(in);
         if (entry == null) {
           break;
         }
-        // update the lease manager
-        INodeFile file = dir.getInode(entry.getInodeId()).asFile();
-        FileUnderConstructionFeature uc = file.getFileUnderConstructionFeature();
-        Preconditions.checkState(uc != null); // file must be under-construction
-        fsn.leaseManager.addLease(uc.getClientName(),
-                entry.getInodeId());
       }
     }
 
@@ -360,6 +356,8 @@ public final class FSImageFormatPBINode {
       if (f.hasFileUC()) {
         INodeSection.FileUnderConstructionFeature uc = f.getFileUC();
         file.toUnderConstruction(uc.getClientName(), uc.getClientMachine());
+        // update the lease manager
+        fsn.leaseManager.addLease(uc.getClientName(), file.getId());
         if (blocks.length > 0) {
           BlockInfo lastBlk = file.getLastBlock();
           lastBlk.convertToBlockUnderConstruction(

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

@@ -3339,7 +3339,6 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
       throw new IOException("Cannot finalize file " + src
           + " because it is not under construction");
     }
-    leaseManager.removeLease(uc.getClientName(), pendingFile);
 
     pendingFile.recordModification(latestSnapshot);
 
@@ -3350,6 +3349,8 @@ public class FSNamesystem implements Namesystem, FSNamesystemMBean,
         allowCommittedBlock? numCommittedAllowed: 0,
         blockManager.getMinReplication());
 
+    leaseManager.removeLease(uc.getClientName(), pendingFile);
+
     waitForLoadingFSImage();
     // close file and persist block allocations for this file
     closeFile(src, pendingFile);

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

@@ -24,8 +24,14 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 
 import com.google.common.collect.Lists;
+import org.apache.hadoop.fs.FSDataOutputStream;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.fs.permission.PermissionStatus;
+import org.apache.hadoop.hdfs.DistributedFileSystem;
+import org.apache.hadoop.hdfs.HdfsConfiguration;
+import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.protocol.HdfsConstants.SafeModeAction;
 import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfo;
 import org.junit.Rule;
 import org.junit.Test;
@@ -109,6 +115,47 @@ public class TestLeaseManager {
     assertThat(lm.countPath(), is(1L));
   }
 
+  /**
+   * Make sure the lease is restored even if only the inode has the record.
+   */
+  @Test
+  public void testLeaseRestorationOnRestart() throws Exception {
+    MiniDFSCluster cluster = null;
+    try {
+      cluster = new MiniDFSCluster.Builder(new HdfsConfiguration())
+          .numDataNodes(1).build();
+      DistributedFileSystem dfs = cluster.getFileSystem();
+
+      // Create an empty file
+      String path = "/testLeaseRestorationOnRestart";
+      FSDataOutputStream out = dfs.create(new Path(path));
+
+      // Remove the lease from the lease manager, but leave it in the inode.
+      FSDirectory dir = cluster.getNamesystem().getFSDirectory();
+      INodeFile file = dir.getINode(path).asFile();
+      cluster.getNamesystem().leaseManager.removeLease(
+          file.getFileUnderConstructionFeature().getClientName(), file);
+
+      // Save a fsimage.
+      dfs.setSafeMode(SafeModeAction.SAFEMODE_ENTER);
+      cluster.getNameNodeRpc().saveNamespace();
+      dfs.setSafeMode(SafeModeAction.SAFEMODE_LEAVE);
+
+      // Restart the namenode.
+      cluster.restartNameNode(true);
+
+      // Check whether the lease manager has the lease
+      dir = cluster.getNamesystem().getFSDirectory();
+      file = dir.getINode(path).asFile();
+      assertTrue("Lease should exist.",
+          cluster.getNamesystem().leaseManager.getLease(file) != null);
+    } finally {
+      if (cluster != null) {
+        cluster.shutdown();
+      }
+    }
+  }
+
   private static FSNamesystem makeMockFsNameSystem() {
     FSDirectory dir = mock(FSDirectory.class);
     FSNamesystem fsn = mock(FSNamesystem.class);