HDFS-11945. Internal lease recovery may not be retried for a long time. Contributed by Kihwal Lee

(cherry picked from commit facb6812d9)

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

@@ -38,6 +38,7 @@ import org.apache.hadoop.hdfs.protocol.HdfsConstants;
 import org.apache.hadoop.hdfs.server.blockmanagement.BlockInfo;
 import org.apache.hadoop.hdfs.server.common.HdfsServerConstants;
 import org.apache.hadoop.util.Daemon;
+import org.apache.hadoop.util.Time;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
@@ -72,6 +73,8 @@ public class LeaseManager {
 
   private long softLimit = HdfsConstants.LEASE_SOFTLIMIT_PERIOD;
   private long hardLimit = HdfsConstants.LEASE_HARDLIMIT_PERIOD;
+  private long lastHolderUpdateTime;
+  private String internalLeaseHolder;
 
   //
   // Used for handling lock-leases
@@ -92,7 +95,26 @@ public class LeaseManager {
   private Daemon lmthread;
   private volatile boolean shouldRunMonitor;
 
-  LeaseManager(FSNamesystem fsnamesystem) {this.fsnamesystem = fsnamesystem;}
+  LeaseManager(FSNamesystem fsnamesystem) {
+    this.fsnamesystem = fsnamesystem;
+    updateInternalLeaseHolder();
+  }
+
+  // Update the internal lease holder with the current time stamp.
+  private void updateInternalLeaseHolder() {
+    this.lastHolderUpdateTime = Time.monotonicNow();
+    this.internalLeaseHolder = HdfsServerConstants.NAMENODE_LEASE_HOLDER +
+        "-" + Time.formatTime(Time.now());
+  }
+
+  // Get the current internal lease holder name.
+  String getInternalLeaseHolder() {
+    long elapsed = Time.monotonicNow() - lastHolderUpdateTime;
+    if (elapsed > hardLimit) {
+      updateInternalLeaseHolder();
+    }
+    return internalLeaseHolder;
+  }
 
   Lease getLease(String holder) {
     return leases.get(holder);
@@ -372,6 +394,7 @@ public class LeaseManager {
       Long[] leaseINodeIds = files.toArray(new Long[files.size()]);
       FSDirectory fsd = fsnamesystem.getFSDirectory();
       String p = null;
+      String newHolder = getInternalLeaseHolder();
       for(Long id : leaseINodeIds) {
         try {
           INodesInPath iip = INodesInPath.fromINode(fsd.getInode(id));
@@ -383,8 +406,7 @@ public class LeaseManager {
           boolean completed = false;
           try {
             completed = fsnamesystem.internalReleaseLease(
-                leaseToCheck, p, iip,
-                HdfsServerConstants.NAMENODE_LEASE_HOLDER);
+                leaseToCheck, p, iip, newHolder);
           } catch (IOException e) {
             LOG.warn("Cannot release the path " + p + " in the lease "
                 + leaseToCheck + ". It will be retried.", e);

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestLeaseRecovery2.java

@@ -465,7 +465,8 @@ public class TestLeaseRecovery2 {
         cluster.getNameNode(), fileStr);
     
     assertFalse("original lease holder should not be the NN",
-        originalLeaseHolder.equals(HdfsServerConstants.NAMENODE_LEASE_HOLDER));
+        originalLeaseHolder.startsWith(
+        HdfsServerConstants.NAMENODE_LEASE_HOLDER));
 
     // hflush file
     AppendTestUtil.LOG.info("hflush");
@@ -501,8 +502,9 @@ public class TestLeaseRecovery2 {
     GenericTestUtils.waitFor(new Supplier<Boolean>() {
       @Override
       public Boolean get() {
-        return HdfsServerConstants.NAMENODE_LEASE_HOLDER.equals(
-            NameNodeAdapter.getLeaseHolderForPath(cluster.getNameNode(), path));
+        String holder =
+            NameNodeAdapter.getLeaseHolderForPath(cluster.getNameNode(), path);
+        return holder.startsWith(HdfsServerConstants.NAMENODE_LEASE_HOLDER);
       }
     }, (int)SHORT_LEASE_PERIOD, (int)SHORT_LEASE_PERIOD * 10);
 
@@ -563,8 +565,8 @@ public class TestLeaseRecovery2 {
     if (size == 0) {
       assertEquals("lease holder should null, file is closed", null, holder);
     } else {
-      assertEquals("lease holder should now be the NN",
-          HdfsServerConstants.NAMENODE_LEASE_HOLDER, holder);
+      assertTrue("lease holder should now be the NN",
+          holder.startsWith(HdfsServerConstants.NAMENODE_LEASE_HOLDER));
     }
     
   }

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

@@ -641,7 +641,7 @@ public class TestFileTruncate {
       String leaseHolder =
           NameNodeAdapter.getLeaseHolderForPath(cluster.getNameNode(),
           p.toUri().getPath());
-      if(leaseHolder.equals(HdfsServerConstants.NAMENODE_LEASE_HOLDER)) {
+      if(leaseHolder.startsWith(HdfsServerConstants.NAMENODE_LEASE_HOLDER)) {
         recoveryTriggered = true;
         break;
       }

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

@@ -20,6 +20,7 @@ package org.apache.hadoop.hdfs.server.namenode;
 import static org.hamcrest.CoreMatchers.is;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.assertNotEquals;
 import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 
@@ -88,6 +89,19 @@ public class TestLeaseManager {
     assertTrue(lm.countLease() < numLease);
   }
 
+  /**
+   * Test whether the internal lease holder name is updated properly.
+   */
+  @Test
+  public void testInternalLeaseHolder() throws Exception {
+    LeaseManager lm = new LeaseManager(makeMockFsNameSystem());
+    // Set the hard lease limit to 500ms.
+    lm.setLeasePeriod(100L, 500L);
+    String holder = lm.getInternalLeaseHolder();
+    Thread.sleep(1000);
+    assertNotEquals(holder, lm.getInternalLeaseHolder());
+  }
+
   @Test
   public void testCountPath() {
     LeaseManager lm = new LeaseManager(makeMockFsNameSystem());