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..8d66977a4f4de3 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,13 @@ 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. 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 = 10; + private ScheduledThreadPoolExecutor scheduledExecutorService; private SubClusterCleaner subClusterCleaner; @@ -196,6 +204,12 @@ protected void serviceStop() throws Exception { if (isStopping.getAndSet(true)) { return; } + // serviceStart() may have scheduled the SubClusterCleaner on this executor. + // Shut it down before the services below, and wait for an in-flight run to + // finish: the cleaner reaches the federation state store on every run, so + // it must not still be running once the Router has stopped. + HadoopExecutors.shutdown(scheduledExecutorService, LOG, + SCHEDULED_EXECUTOR_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS); super.serviceStop(); DefaultMetricsSystem.shutdown(); WebServiceClient.destroy(); diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/RouterClientRMService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/RouterClientRMService.java index 3d45ad207cf403..490b8ff6480f15 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/RouterClientRMService.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/main/java/org/apache/hadoop/yarn/server/router/clientrm/RouterClientRMService.java @@ -202,6 +202,13 @@ protected void serviceStop() throws Exception { if (this.server != null) { this.server.stop(); } + // serviceStart() started the secret manager's ExpiredTokenRemover; stop it + // here, once the server is down and no request can reach it, so it does not + // outlive this service and keep using the federation state store. Mirrors + // RMSecretManagerService#serviceStop. + if (this.routerDTSecretManager != null) { + this.routerDTSecretManager.stopThreads(); + } userPipelineMap.clear(); super.serviceStop(); } diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/subcluster/TestMockRouter.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/subcluster/TestMockRouter.java index 751a3fed6487d2..8dca6484a5a57a 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/subcluster/TestMockRouter.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/subcluster/TestMockRouter.java @@ -19,7 +19,8 @@ import org.apache.commons.lang3.ArrayUtils; import org.apache.hadoop.fs.CommonConfigurationKeys; -import org.apache.hadoop.io.retry.RetryPolicy; +import org.apache.hadoop.service.CompositeService.CompositeServiceShutdownHook; +import org.apache.hadoop.util.ShutdownHookManager; import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.hadoop.yarn.exceptions.YarnException; import org.apache.hadoop.yarn.server.MiniYARNCluster; @@ -30,7 +31,10 @@ import org.slf4j.LoggerFactory; /** - * Tests {@link Router}. + * Process entry point that starts a {@link Router} for the federation + * end-to-end suites. This is not a test: it declares no test methods. + * {@link TestFederationSubCluster} launches it as a separate JVM and stops it + * with Process#destroy. */ public class TestMockRouter { @@ -73,18 +77,43 @@ public static void main(String[] args) throws YarnException { conf.set(YarnConfiguration.ROUTER_WEBAPP_ADDRESS, getHostNameAndPort(pRouterWebAddressPort)); - RetryPolicy retryPolicy = FederationStateStoreFacade.createRetryPolicy(conf); - + // This class is launched as its own JVM by JavaProcess, and the parent test + // terminates it with Process#destroy: SIGTERM on Linux and macOS, which runs + // shutdown hooks. (On Windows destroy maps to TerminateProcess, which runs + // none, so there this is inert.) Router#main registers this hook itself; + // starting the Router directly, as here, skips it, so without this the JVM + // exits with the Router's services still running and none of them ever + // stopped. + ShutdownHookManager.get().addShutdownHook( + new CompositeServiceShutdownHook(router), Router.SHUTDOWN_HOOK_PRIORITY); router.init(conf); router.start(); - FederationStateStore stateStore = (FederationStateStore) - FederationStateStoreFacade.createRetryInstance(conf, - YarnConfiguration.FEDERATION_STATESTORE_CLIENT_CLASS, - YarnConfiguration.DEFAULT_FEDERATION_STATESTORE_CLIENT_CLASS, - FederationStateStore.class, retryPolicy); - stateStore.init(conf); - FederationStateStoreFacade.getInstance().reinitialize(stateStore, conf); + // Starting the Router has already created and initialized the facade's + // store: RouterClientRMService#serviceStart builds a + // RouterDelegationTokenSecretManager whose constructor calls + // FederationStateStoreFacade#getInstance(Configuration). Building a second + // store here and handing it to reinitialize() would orphan that first one - + // reinitialize swaps the reference, it does not close what it replaces - so + // take the store the facade is already using. + FederationStateStore stateStore = + FederationStateStoreFacade.getInstance(conf).getStateStore(); + + // The facade holds this store but never closes it, so its ZooKeeper + // connection outlives the process unless we close it ourselves. Registered + // at a lower priority than the Router hook above: ShutdownHookManager runs + // the highest priority first, so the Router is fully stopped before the + // store it may still be using goes away. That ordering is only sound + // because RouterClientRMService#serviceStop stops the delegation token + // secret manager's ExpiredTokenRemover, which would otherwise still be + // reaching the facade after the Router has stopped. + ShutdownHookManager.get().addShutdownHook(() -> { + try { + stateStore.close(); + } catch (Exception e) { + LOG.warn("Error closing FederationStateStore.", e); + } + }, Router.SHUTDOWN_HOOK_PRIORITY - 5); } private static String getHostNameAndPort(int port) { diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/subcluster/TestMockSubCluster.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/subcluster/TestMockSubCluster.java index d3ab90f3105e78..8a154ee83a54d5 100644 --- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/subcluster/TestMockSubCluster.java +++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-router/src/test/java/org/apache/hadoop/yarn/server/router/subcluster/TestMockSubCluster.java @@ -20,20 +20,33 @@ import org.apache.commons.lang3.ArrayUtils; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.CommonConfigurationKeys; +import org.apache.hadoop.service.CompositeService.CompositeServiceShutdownHook; import org.apache.hadoop.test.GenericTestUtils; +import org.apache.hadoop.util.ShutdownHookManager; import org.apache.hadoop.util.Time; import org.apache.hadoop.yarn.conf.YarnConfiguration; import org.apache.hadoop.yarn.server.MiniYARNCluster; +import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.io.File; +import java.util.concurrent.TimeUnit; import static org.apache.hadoop.yarn.server.router.subcluster.TestFederationSubCluster.ZK_FEDERATION_STATESTORE; public class TestMockSubCluster { private static final Logger LOG = LoggerFactory.getLogger(TestMockSubCluster.class); + + /** + * Shutdown budget for the whole sub-cluster. Each NodeManager spends a + * hardcoded 10s in DeletionService#serviceStop on top of its container grace + * period, so three of them do not fit ShutdownHookManager's 30s default and + * the hook would be abandoned part-way through the teardown. + */ + private static final long SHUTDOWN_TIMEOUT_SECONDS = 60; + private Configuration conf; private String subClusterId; @@ -51,6 +64,16 @@ private static String getHostNameAndPort(int port) { public void startYarnSubCluster() { MiniYARNCluster yrCluster = new MiniYARNCluster(subClusterId, 3, 1, 1, false); + // This class is launched as its own JVM by JavaProcess, and the parent test + // terminates it with Process#destroy: SIGTERM on Linux and macOS, which runs + // shutdown hooks. (On Windows destroy maps to TerminateProcess, which runs + // none, so there this is inert.) MiniYARNCluster registers no hook of its + // own, so without this the JVM exits with its ResourceManager and + // NodeManagers still running and no service ever stopped. Registered before + // init/start so a failure part-way through startup still tears down + // whatever came up. + ShutdownHookManager.get().addShutdownHook(new CompositeServiceShutdownHook(yrCluster), + ResourceManager.SHUTDOWN_HOOK_PRIORITY, SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS); yrCluster.init(conf); yrCluster.start(); } @@ -87,6 +110,12 @@ public static void main(String[] args) { getHostNameAndPort(pRmTrackerAddressPort)); conf.set(YarnConfiguration.RM_WEBAPP_ADDRESS, getHostNameAndPort(pRmWebAddressPort)); conf.setBoolean(YarnConfiguration.YARN_MINICLUSTER_FIXED_PORTS, true); + // These sub-clusters keep applications running for the lifetime of the + // suite, so on shutdown every NodeManager would otherwise wait out its full + // container grace period - 250ms + 5000ms + 1s of slop, rounded up to 7s by + // the one-second poll in ContainerManagerImpl - before it stops. + conf.setLong(YarnConfiguration.NM_SLEEP_DELAY_BEFORE_SIGKILL_MS, 0); + conf.setLong(YarnConfiguration.NM_PROCESS_KILL_WAIT_MS, 0); conf.setBoolean(YarnConfiguration.FEDERATION_ENABLED, true); conf.set(YarnConfiguration.FEDERATION_STATESTORE_CLIENT_CLASS, ZK_FEDERATION_STATESTORE); conf.set(CommonConfigurationKeys.ZK_ADDRESS, pZkAddress);