YARN-12003. Shut down the Router scheduled executor on service stop

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

@@ -48,6 +48,7 @@
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.VersionInfo;
 import org.apache.hadoop.util.GenericOptionsParser;
+import org.apache.hadoop.util.concurrent.HadoopExecutors;
 import org.apache.hadoop.yarn.YarnUncaughtExceptionHandler;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
@@ -124,6 +125,14 @@ public class Router extends CompositeService {
 
   private static final String UI2_WEBAPP_NAME = "/ui2";
 
+  /**
+   * How long serviceStop waits for the scheduled executor to drain before it
+   * interrupts what is still running, and then again for the interrupted tasks
+   * to exit, so stopping can take up to twice this. One SubClusterCleaner run
+   * is a scan of the state store, so this only has to cover a single pass.
+   */
+  private static final long SCHEDULED_EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS = 5;
+
   private ScheduledThreadPoolExecutor scheduledExecutorService;
   private SubClusterCleaner subClusterCleaner;
 
@@ -196,6 +205,12 @@ protected void serviceStop() throws Exception {
     if (isStopping.getAndSet(true)) {
       return;
     }
+    // serviceStart() may have scheduled the SubClusterCleaner on this executor.
+    // Shut it down before stopping the child services, allowing an in-flight
+    // cleaner run to finish within the timeout. This is best effort: a run that
+    // does not respond to interruption can outlive the stop.
+    HadoopExecutors.shutdown(scheduledExecutorService, LOG,
+        SCHEDULED_EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS);
     super.serviceStop();
     DefaultMetricsSystem.shutdown();
     WebServiceClient.destroy();
@@ -339,6 +354,16 @@ public RouterRMAdminService getRmAdminProxyService() {
     return rmAdminProxyService;
   }
 
+  /**
+   * Returns the executor that runs the SubClusterCleaner.
+   *
+   * @return the executor that runs the SubClusterCleaner.
+   */
+  @VisibleForTesting
+  public ScheduledThreadPoolExecutor getScheduledExecutorService() {
+    return scheduledExecutorService;
+  }
+
   /**
    * Returns the hostname for this Router. If the hostname is not
    * explicitly configured in the given config, then it is determined.

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

@@ -18,6 +18,7 @@
 package org.apache.hadoop.yarn.server.router;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 import static org.mockito.Mockito.mock;
@@ -52,6 +53,7 @@
 import java.util.HashMap;
 import java.util.Locale;
 import java.util.Map;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
 
 /**
  * Tests {@link Router}.
@@ -99,6 +101,29 @@ public void testServiceACLRefresh() {
 
   }
 
+  /**
+   * Stopping the Router shuts down the executor that runs the
+   * SubClusterCleaner, so the cleaner does not outlive the Router.
+   */
+  @Test
+  public void testServiceStopShutsDownScheduledExecutor() {
+    Configuration conf = new YarnConfiguration();
+    conf.setBoolean(YarnConfiguration.ROUTER_DEREGISTER_SUBCLUSTER_ENABLED, true);
+    Router router = new Router();
+    try {
+      router.init(conf);
+      router.start();
+      ScheduledThreadPoolExecutor executor = router.getScheduledExecutorService();
+      assertFalse(executor.isShutdown());
+
+      router.stop();
+
+      assertTrue(executor.isTerminated());
+    } finally {
+      router.stop();
+    }
+  }
+
   private void verifyServiceACLsRefresh(ServiceAuthorizationManager manager,
       Class<?> protocol, String aclString) {
     if (manager.getProtocolsWithAcls().size() == 0) {