diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/Router.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/Router.java index 8f3c4d0fe577e0..1865d561eeec2e 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/Router.java +++ b/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. diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/TestRouter.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/TestRouter.java index 10ef06c8125f56..245e10f43b3197 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/TestRouter.java +++ b/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) {