YARN-11984. Federation mock sub-cluster and router do not stop cleanly on SIGTERM.

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();

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();
   }

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) {

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);