HDFS-17685: Option to explicitly choose DFS client lease renewal interval

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/client/HdfsClientConfigKeys.java

@@ -202,6 +202,9 @@ public interface HdfsClientConfigKeys {
   long DFS_CLIENT_DEAD_NODE_DETECTION_PROBE_SUSPECT_NODE_INTERVAL_MS_DEFAULT =
       300; // 300ms
 
+  String DFS_CLIENT_LEASE_RENEWAL_INTERVAL_KEY = "dfs.client.lease.renewal.interval.ms";
+  int DFS_CLIENT_LEASE_RENEWAL_INTERVAL_DEFAULT = 0;
+
   // refreshing LocatedBlocks period. A value of 0 disables the feature.
   String  DFS_CLIENT_REFRESH_READ_BLOCK_LOCATIONS_MS_KEY =
       "dfs.client.refresh.read-block-locations.ms";

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/client/impl/DfsClientConf.java

@@ -108,6 +108,7 @@ public class DfsClientConf {
                                                                 .class);
 
   private final int hdfsTimeout;    // timeout value for a DFS operation.
+  private final int leaseRenewalIntervalMs;
 
   private final int maxFailoverAttempts;
   private final int maxRetryAttempts;
@@ -170,7 +171,9 @@ public class DfsClientConf {
   public DfsClientConf(Configuration conf) {
     // The hdfsTimeout is currently the same as the ipc timeout
     hdfsTimeout = Client.getRpcTimeout(conf);
-
+    leaseRenewalIntervalMs = conf.getInt(
+        HdfsClientConfigKeys.DFS_CLIENT_LEASE_RENEWAL_INTERVAL_KEY,
+        HdfsClientConfigKeys.DFS_CLIENT_LEASE_RENEWAL_INTERVAL_DEFAULT);
     maxRetryAttempts = conf.getInt(
         Retry.MAX_ATTEMPTS_KEY,
         Retry.MAX_ATTEMPTS_DEFAULT);
@@ -183,7 +186,6 @@ public DfsClientConf(Configuration conf) {
     retryIntervalForGetLastBlockLength = conf.getInt(
         Retry.INTERVAL_GET_LAST_BLOCK_LENGTH_KEY,
         Retry.INTERVAL_GET_LAST_BLOCK_LENGTH_DEFAULT);
-
     maxFailoverAttempts = conf.getInt(
         Failover.MAX_ATTEMPTS_KEY,
         Failover.MAX_ATTEMPTS_DEFAULT);
@@ -431,6 +433,10 @@ public int getHdfsTimeout() {
     return hdfsTimeout;
   }
 
+  public int getLeaseRenewalIntervalMs() {
+    return leaseRenewalIntervalMs;
+  }
+
   /**
    * @return the maxFailoverAttempts
    */

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/client/impl/LeaseRenewer.java

@@ -20,6 +20,7 @@
 import java.io.IOException;
 import java.net.SocketTimeoutException;
 import java.util.ArrayList;
+import java.util.Collection;
 import java.util.Collections;
 import java.util.Comparator;
 import java.util.HashMap;
@@ -248,14 +249,7 @@ private synchronized void addClient(final DFSClient dfsc) {
     //client not found, add it
     dfsclients.add(dfsc);
 
-    //update renewal time
-    final int hdfsTimeout = dfsc.getConf().getHdfsTimeout();
-    if (hdfsTimeout > 0) {
-      final long half = hdfsTimeout/2;
-      if (half < renewal) {
-        this.renewal = half;
-      }
-    }
+    renewal = getNewRenewalIntervalMs(dfsclients);
   }
 
   private synchronized void clearClients() {
@@ -378,17 +372,42 @@ public synchronized void closeClient(final DFSClient dfsc) {
       }
     }
 
-    //update renewal time
-    if (renewal == dfsc.getConf().getHdfsTimeout()/2) {
-      long min = HdfsConstants.LEASE_SOFTLIMIT_PERIOD;
-      for(DFSClient c : dfsclients) {
-        final int timeout = c.getConf().getHdfsTimeout();
-        if (timeout > 0 && timeout < min) {
-          min = timeout;
+    renewal = getNewRenewalIntervalMs(dfsclients);
+  }
+
+  @VisibleForTesting
+  static long getNewRenewalIntervalMs(Collection<DFSClient> dfsClients) {
+    // Update renewal time to the first applicable of:
+    //   1. Requested renewal time amongst all DFSClients, if requested and smaller than the default
+    //      renewal interval
+    //   2. Half the HDFS timeout amongst all DFSClients, if requested and smaller than the default
+    //      renewal interval
+    //   3. Default renewal time of HdfsConstants.LEASE_SOFTLIMIT_PERIOD / 2
+    // #2 exists because users with small timeouts want to find out quickly when a NameNode dies,
+    // and a small lease renewal interval will help to inform them quickly. See HDFS-278.
+    long min = HdfsConstants.LEASE_SOFTLIMIT_PERIOD / 2;
+    boolean leaseOverrideSet = false;
+    for (DFSClient c : dfsClients) {
+      final int newLeaseRenewalInterval = c.getConf().getLeaseRenewalIntervalMs();
+      if (newLeaseRenewalInterval > 0 && newLeaseRenewalInterval <= min) {
+        min = newLeaseRenewalInterval;
+        leaseOverrideSet = true;
+      }
+    }
+    if (leaseOverrideSet) {
+      return min;
+    }
+
+    for (DFSClient c : dfsClients) {
+      final int hdfsTimeout = c.getConf().getHdfsTimeout();
+      if (hdfsTimeout > 0) {
+        final long half = hdfsTimeout/2;
+        if (half < min) {
+          min = half;
         }
       }
-      renewal = min/2;
     }
+    return min;
   }
 
   public void interruptAndJoin() throws InterruptedException {

hadoop-hdfs-project/hadoop-hdfs-client/src/test/java/org/apache/hadoop/hdfs/client/impl/TestLeaseRenewer.java

@@ -20,6 +20,7 @@
 import java.util.function.Supplier;
 import org.apache.hadoop.hdfs.DFSClient;
 import org.apache.hadoop.hdfs.DFSOutputStream;
+import org.apache.hadoop.hdfs.protocol.HdfsConstants;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.hadoop.util.Time;
@@ -37,7 +38,10 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.regex.Pattern;
 
+import org.apache.hadoop.thirdparty.com.google.common.collect.ImmutableList;
+
 import static org.junit.Assert.assertSame;
+import static org.mockito.Mockito.doReturn;
 
 public class TestLeaseRenewer {
   private final String FAKE_AUTHORITY="hdfs://nn1/";
@@ -284,4 +288,67 @@ private static int countThreadMatching(Pattern pattern) {
     }
     return count;
   }
+
+  @Test
+  public void testDefaultRenewalInterval() {
+    final DfsClientConf mockConf = Mockito.mock(DfsClientConf.class);
+    doReturn(0).when(mockConf).getLeaseRenewalIntervalMs();
+    doReturn(0).when(mockConf).getHdfsTimeout();
+    DFSClient dfsClient = Mockito.mock(DFSClient.class);
+    doReturn(mockConf).when(dfsClient).getConf();
+    // Use default renewal interval if both explicit renewal interval and HDFS timeout are not set
+    Assert.assertEquals(HdfsConstants.LEASE_SOFTLIMIT_PERIOD / 2,
+        LeaseRenewer.getNewRenewalIntervalMs(ImmutableList.of(dfsClient)));
+  }
+
+  @Test
+  public void testShortExplicitRenewalInterval() {
+    final DfsClientConf mockConf = Mockito.mock(DfsClientConf.class);
+    doReturn(37).when(mockConf).getLeaseRenewalIntervalMs();
+    doReturn(100).when(mockConf).getHdfsTimeout();
+    DFSClient dfsClient = Mockito.mock(DFSClient.class);
+    doReturn(mockConf).when(dfsClient).getConf();
+    // Explicit renewal interval is shorter than half the HDFS timeout, renewer should use it
+    Assert.assertEquals(37,
+        LeaseRenewer.getNewRenewalIntervalMs(ImmutableList.of(dfsClient)));
+  }
+
+  @Test
+  public void testMediumExplicitRenewalInterval() {
+    final DfsClientConf mockConf = Mockito.mock(DfsClientConf.class);
+    doReturn(370).when(mockConf).getLeaseRenewalIntervalMs();
+    doReturn(100).when(mockConf).getHdfsTimeout();
+    DFSClient dfsClient = Mockito.mock(DFSClient.class);
+    doReturn(mockConf).when(dfsClient).getConf();
+    // Explicit renewal interval is longer than half the HDFS timeout,
+    // renewer should use it regardless
+    Assert.assertEquals(370,
+        LeaseRenewer.getNewRenewalIntervalMs(ImmutableList.of(dfsClient)));
+  }
+
+  @Test
+  public void testLongExplicitRenewalIntervalWithShortHdfsTimeout() {
+    final DfsClientConf mockConf = Mockito.mock(DfsClientConf.class);
+    doReturn(37000).when(mockConf).getLeaseRenewalIntervalMs();
+    doReturn(100).when(mockConf).getHdfsTimeout();
+    DFSClient dfsClient = Mockito.mock(DFSClient.class);
+    doReturn(mockConf).when(dfsClient).getConf();
+    // Explicit renewal interval is longer than HdfsConstants.LEASE_SOFTLIMIT_PERIOD / 2,
+    // but half the HDFS timeout is shorter, so renewer should use the HDFS timeout
+    Assert.assertEquals(50,
+        LeaseRenewer.getNewRenewalIntervalMs(ImmutableList.of(dfsClient)));
+  }
+
+  @Test
+  public void testLongExplicitRenewalIntervalWithLongHdfsTimeout() {
+    final DfsClientConf mockConf = Mockito.mock(DfsClientConf.class);
+    doReturn(37000).when(mockConf).getLeaseRenewalIntervalMs();
+    doReturn(120000).when(mockConf).getHdfsTimeout();
+    DFSClient dfsClient = Mockito.mock(DFSClient.class);
+    doReturn(mockConf).when(dfsClient).getConf();
+    // Explicit renewal interval and HDFS timeout is longer than
+    // HdfsConstants.LEASE_SOFTLIMIT_PERIOD / 2, so renewer should use default renewal interval
+    Assert.assertEquals(HdfsConstants.LEASE_SOFTLIMIT_PERIOD / 2,
+        LeaseRenewer.getNewRenewalIntervalMs(ImmutableList.of(dfsClient)));
+  }
 }

hadoop-hdfs-project/hadoop-hdfs/src/main/resources/hdfs-default.xml

@@ -3444,6 +3444,16 @@
     </description>
   </property>
 
+  <property>
+    <name>dfs.client.lease.renewal.interval.ms</name>
+    <value>0</value>
+    <description>
+      If set between 0 and 30000 inclusive, HDFS clients will renew leases for files they are writing at this interval.
+      If dfs.client.lease.renewal.interval.ms is not set and ipc.client.rpc-timeout.ms is set between 0 and 60000,
+      then the value of ipc.client.rpc-timeout.ms / 2 will be used as the lease renewal interval.
+    </description>
+  </property>
+
   <property>
     <name>dfs.client.refresh.read-block-locations.ms</name>
     <value>0</value>