YARN-9559. Create AbstractContainersLauncher for pluggable ContainersLauncher logic. (Contributed by Jonathan Hung)

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

@@ -4053,6 +4053,12 @@ public class YarnConfiguration extends Configuration {
   public static final int
       DEFAULT_RM_ACTIVITIES_MANAGER_APP_ACTIVITIES_MAX_QUEUE_LENGTH = 100;
 
+  /**
+   * Containers launcher implementation to use.
+   */
+  public static final String NM_CONTAINERS_LAUNCHER_CLASS =
+      NM_PREFIX + "containers-launcher.class";
+
   public YarnConfiguration() {
     super();
   }

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

@@ -4230,4 +4230,13 @@
     <name>yarn.resourcemanager.activities-manager.app-activities.max-queue-length</name>
     <value>100</value>
   </property>
+
+  <property>
+    <description>
+      Containers launcher implementation for determining how containers
+      are launched within NodeManagers.
+    </description>
+    <name>yarn.nodemanager.containers-launcher.class</name>
+    <value>org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainersLauncher</value>
+  </property>
 </configuration>

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java

@@ -20,6 +20,7 @@ package org.apache.hadoop.yarn.server.nodemanager.containermanager;
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.protobuf.ByteString;
+import org.apache.hadoop.util.ReflectionUtils;
 import org.apache.hadoop.yarn.api.protocolrecords.GetLocalizationStatusesRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.GetLocalizationStatusesResponse;
 import org.apache.hadoop.yarn.api.records.LocalizationStatus;
@@ -131,6 +132,7 @@ import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Cont
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerImpl;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerKillEvent;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerReInitEvent;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.AbstractContainersLauncher;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainersLauncher;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainersLauncherEventType;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.SignalContainersLauncherEvent;
@@ -208,7 +210,7 @@ public class ContainerManagerImpl extends CompositeService implements
   private final ContainersMonitor containersMonitor;
   private Server server;
   private final ResourceLocalizationService rsrcLocalizationSrvc;
-  private final ContainersLauncher containersLauncher;
+  private final AbstractContainersLauncher containersLauncher;
   private final AuxServices auxiliaryServices;
   private final NodeManagerMetrics metrics;
 
@@ -567,9 +569,21 @@ public class ContainerManagerImpl extends CompositeService implements
     return nmTimelinePublisherLocal;
   }
 
-  protected ContainersLauncher createContainersLauncher(Context context,
-      ContainerExecutor exec) {
-    return new ContainersLauncher(context, this.dispatcher, exec, dirsHandler, this);
+  protected AbstractContainersLauncher createContainersLauncher(
+      Context ctxt, ContainerExecutor exec) {
+    Class<? extends AbstractContainersLauncher> containersLauncherClass =
+        ctxt.getConf()
+            .getClass(YarnConfiguration.NM_CONTAINERS_LAUNCHER_CLASS,
+                ContainersLauncher.class, AbstractContainersLauncher.class);
+    AbstractContainersLauncher launcher;
+    try {
+      launcher = ReflectionUtils.newInstance(containersLauncherClass,
+          ctxt.getConf());
+      launcher.init(ctxt, this.dispatcher, exec, dirsHandler, this);
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+    return launcher;
   }
 
   protected EventHandler<ApplicationEvent> createApplicationEventDispatcher() {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/AbstractContainersLauncher.java

@@ -0,0 +1,41 @@
+/**
+ * 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.nodemanager.containermanager.launcher;
+
+import org.apache.hadoop.service.Service;
+import org.apache.hadoop.yarn.event.Dispatcher;
+import org.apache.hadoop.yarn.event.EventHandler;
+import org.apache.hadoop.yarn.server.nodemanager.ContainerExecutor;
+import org.apache.hadoop.yarn.server.nodemanager.Context;
+import org.apache.hadoop.yarn.server.nodemanager.LocalDirsHandlerService;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.ContainerManagerImpl;
+
+
+/**
+ * Pluggable ContainersLauncher interface for processing
+ * ContainersLauncherEvents.
+ */
+public interface AbstractContainersLauncher extends Service,
+    EventHandler<ContainersLauncherEvent> {
+
+  void init(Context context, Dispatcher dispatcher,
+      ContainerExecutor exec, LocalDirsHandlerService dirsHandler,
+      ContainerManagerImpl containerManager);
+
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/ContainersLauncher.java

@@ -38,7 +38,6 @@ import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.util.concurrent.HadoopExecutors;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.event.Dispatcher;
-import org.apache.hadoop.yarn.event.EventHandler;
 import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
 import org.apache.hadoop.yarn.server.nodemanager.ContainerExecutor;
 import org.apache.hadoop.yarn.server.nodemanager.Context;
@@ -58,15 +57,15 @@ import com.google.common.util.concurrent.ThreadFactoryBuilder;
  * 
  */
 public class ContainersLauncher extends AbstractService
-    implements EventHandler<ContainersLauncherEvent> {
+    implements AbstractContainersLauncher {
 
   private static final Logger LOG =
        LoggerFactory.getLogger(ContainersLauncher.class);
 
-  private final Context context;
-  private final ContainerExecutor exec;
-  private final Dispatcher dispatcher;
-  private final ContainerManagerImpl containerManager;
+  private Context context;
+  private ContainerExecutor exec;
+  private Dispatcher dispatcher;
+  private ContainerManagerImpl containerManager;
 
   private LocalDirsHandlerService dirsHandler;
   @VisibleForTesting
@@ -79,15 +78,27 @@ public class ContainersLauncher extends AbstractService
   public final Map<ContainerId, ContainerLaunch> running =
     Collections.synchronizedMap(new HashMap<ContainerId, ContainerLaunch>());
 
+  public ContainersLauncher() {
+    super("containers-launcher");
+  }
+
+  @VisibleForTesting
   public ContainersLauncher(Context context, Dispatcher dispatcher,
       ContainerExecutor exec, LocalDirsHandlerService dirsHandler,
       ContainerManagerImpl containerManager) {
-    super("containers-launcher");
-    this.exec = exec;
-    this.context = context;
-    this.dispatcher = dispatcher;
-    this.dirsHandler = dirsHandler;
-    this.containerManager = containerManager;
+    this();
+    init(context, dispatcher, exec, dirsHandler, containerManager);
+  }
+
+  @Override
+  public void init(Context nmContext, Dispatcher nmDispatcher,
+      ContainerExecutor containerExec, LocalDirsHandlerService nmDirsHandler,
+      ContainerManagerImpl nmContainerManager) {
+    this.exec = containerExec;
+    this.context = nmContext;
+    this.dispatcher = nmDispatcher;
+    this.dirsHandler = nmDirsHandler;
+    this.containerManager = nmContainerManager;
   }
 
   @Override

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/package-info.java

@@ -0,0 +1,22 @@
+/*
+ * 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.
+ */
+
+/**
+ * This package contains classes related to NM container launch.
+ */
+package org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher;