YARN-12048. Fix keep-alive time units in Federation Router thread pools

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-api/src/main/java/org/apache/hadoop/yarn/conf/YarnConfiguration.java

@@ -4457,13 +4457,10 @@ public static boolean isAclEnabled(Configuration conf) {
       ROUTER_PREFIX + "interceptor.user-thread-pool.keep-alive-time";
 
   /**
-   * This configurable is used to set the default time of keepAliveTime
-   * of the thread pool of the interceptor.
-   *
-   * the default value is 0s.
+   * Default interceptor thread-pool keep-alive time, in seconds.
    */
   public static final long DEFAULT_ROUTER_USER_CLIENT_THREAD_POOL_KEEP_ALIVE_TIME =
-      TimeUnit.SECONDS.toMillis(0); // 0s
+      30L;
 
   /**
    * This method configures the policy for core threads regarding termination

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-common/src/main/resources/yarn-default.xml

@@ -5281,6 +5281,8 @@
     <value>30s</value>
     <description>
       This configurable is used to set the keepAliveTime of the thread pool of the interceptor.
+      Specify a duration with a time unit suffix, such as 30s or 500ms.
+      Values without a suffix are interpreted as seconds.
       Default is 30s.
     </description>
   </property>

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/FederationClientInterceptor.java

@@ -223,7 +223,8 @@ public void init(String userName) {
 
     long keepAliveTime = getConf().getTimeDuration(
         YarnConfiguration.ROUTER_USER_CLIENT_THREAD_POOL_KEEP_ALIVE_TIME,
-        YarnConfiguration.DEFAULT_ROUTER_USER_CLIENT_THREAD_POOL_KEEP_ALIVE_TIME, TimeUnit.SECONDS);
+        YarnConfiguration.DEFAULT_ROUTER_USER_CLIENT_THREAD_POOL_KEEP_ALIVE_TIME,
+        TimeUnit.SECONDS, TimeUnit.MILLISECONDS);
 
     ThreadFactory threadFactory = new ThreadFactoryBuilder()
         .setNameFormat("RPC Router Client-" + userName + "-%d ").build();
@@ -2108,6 +2109,11 @@ public void shutdown() {
     super.shutdown();
   }
 
+  @VisibleForTesting
+  ThreadPoolExecutor getExecutorService() {
+    return executorService;
+  }
+
   @Override
   public GetAttributesToNodesResponse getAttributesToNodes(
       GetAttributesToNodesRequest request) throws YarnException, IOException {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/rmadmin/FederationRMAdminInterceptor.java

@@ -150,7 +150,8 @@ public void init(String userName) {
 
     long keepAliveTime = getConf().getTimeDuration(
         YarnConfiguration.ROUTER_USER_CLIENT_THREAD_POOL_KEEP_ALIVE_TIME,
-        YarnConfiguration.DEFAULT_ROUTER_USER_CLIENT_THREAD_POOL_KEEP_ALIVE_TIME, TimeUnit.SECONDS);
+        YarnConfiguration.DEFAULT_ROUTER_USER_CLIENT_THREAD_POOL_KEEP_ALIVE_TIME,
+        TimeUnit.SECONDS, TimeUnit.MILLISECONDS);
 
     BlockingQueue<Runnable> workQueue = new LinkedBlockingQueue<>();
     this.executorService = new ThreadPoolExecutor(numThreads, numThreads,

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/clientrm/TestFederationInterceptorThreadPool.java

@@ -0,0 +1,81 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.yarn.server.router.clientrm;
+
+import java.util.concurrent.TimeUnit;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.server.federation.store.impl.MemoryFederationStateStore;
+import org.apache.hadoop.yarn.server.router.rmadmin.FederationRMAdminInterceptor;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.CsvSource;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class TestFederationInterceptorThreadPool {
+
+  @ParameterizedTest
+  @CsvSource({", 30000", "0, 0", "0s, 0", "30s, 30000", "1m, 60000",
+      "500ms, 500", "60, 60000"})
+  public void testKeepAliveTime(String configuredValue, long expectedMillis) {
+    Configuration conf = new Configuration(false);
+    if (configuredValue != null) {
+      conf.set(YarnConfiguration.ROUTER_USER_CLIENT_THREAD_POOL_KEEP_ALIVE_TIME,
+          configuredValue);
+    }
+    assertKeepAliveTime(conf, expectedMillis);
+  }
+
+  @Test
+  public void testXmlDefaultKeepAliveTime() {
+    assertKeepAliveTime(new YarnConfiguration(), 30000);
+  }
+
+  private void assertKeepAliveTime(Configuration conf, long expectedMillis) {
+    conf.set(YarnConfiguration.FEDERATION_STATESTORE_CLIENT_CLASS,
+        MemoryFederationStateStore.class.getName());
+    conf.setBoolean(
+        YarnConfiguration.ROUTER_USER_CLIENT_THREAD_POOL_ALLOW_CORE_THREAD_TIMEOUT, true);
+    FederationClientInterceptor client = new FederationClientInterceptor();
+    FederationRMAdminInterceptor admin = new FederationRMAdminInterceptor();
+    client.setConf(conf);
+    admin.setConf(conf);
+    try {
+      client.init("test-user");
+      admin.init("test-user");
+      assertEquals(expectedMillis,
+          client.getExecutorService().getKeepAliveTime(TimeUnit.MILLISECONDS));
+      assertEquals(expectedMillis,
+          admin.getExecutorService().getKeepAliveTime(TimeUnit.MILLISECONDS));
+      assertEquals(expectedMillis > 0,
+          client.getExecutorService().allowsCoreThreadTimeOut());
+      assertEquals(expectedMillis > 0,
+          admin.getExecutorService().allowsCoreThreadTimeOut());
+    } finally {
+      if (client.getExecutorService() != null) {
+        client.getExecutorService().shutdownNow();
+      }
+      if (admin.getExecutorService() != null) {
+        admin.getExecutorService().shutdownNow();
+      }
+    }
+  }
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/Federation.md

@@ -454,7 +454,7 @@ Below is the configuration for the thread pool, which can be adjusted based on t
 |:----------------------------------------------------------------------|:--------|:---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
 | `yarn.router.interceptor.user-thread-pool.minimum-pool-size`          | `5`     | This configurable is used to set the corePoolSize(minimumPoolSize) of the thread pool of the interceptor. Default is 5.                                                                                                                      |
 | `yarn.router.interceptor.user-thread-pool.maximum-pool-size`          | `5`     | This configuration is used to set the default value of maximumPoolSize of the thread pool of the interceptor. Default is 5.                                                                                                                  |
-| `yarn.router.interceptor.user-thread-pool.keep-alive-time`            | `0s`    | This configurable is used to set the keepAliveTime of the thread pool of the interceptor. Default is 0s.                                                                                                                                     |
+| `yarn.router.interceptor.user-thread-pool.keep-alive-time`            | `30s`   | This configurable is used to set the keepAliveTime of the thread pool of the interceptor. Default is 30s. Specify a duration with a time unit suffix, such as `30s` or `500ms`. Values without a suffix are interpreted as seconds. |
 | `yarn.router.interceptor.user-thread-pool.allow-core-thread-time-out` | `false` | This method configures the policy for core threads regarding termination when no tasks arrive within the keep-alive time. If set to true, We need to ensure that yarn.router.interceptor.user-thread-pool.keep-alive-time is greater than 0. |
 
 - yarn.router.rmadmin.interceptor-class.pipeline