YARN-8064. Bugfix for clean up container-executor cmd helper file.

            Contributed by Eric Badger

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

@@ -113,6 +113,7 @@ public class LinuxContainerExecutor extends ContainerExecutor {
   private boolean containerLimitUsers;
   private ResourceHandler resourceHandlerChain;
   private LinuxContainerRuntime linuxContainerRuntime;
+  private Context nmContext;
 
   /**
    * The container exit code.
@@ -284,8 +285,9 @@ public class LinuxContainerExecutor extends ContainerExecutor {
   }
 
   @Override
-  public void init(Context nmContext) throws IOException {
+  public void init(Context context) throws IOException {
     Configuration conf = super.getConf();
+    this.nmContext = context;
 
     // Send command to executor which will just start up,
     // verify configuration/permissions and exit
@@ -931,11 +933,11 @@ public class LinuxContainerExecutor extends ContainerExecutor {
           PrivilegedOperationExecutor.getInstance(super.getConf());
       if (DockerCommandExecutor.isRemovable(
           DockerCommandExecutor.getContainerStatus(containerId,
-              super.getConf(), privOpExecutor))) {
+              super.getConf(), privOpExecutor, nmContext))) {
         LOG.info("Removing Docker container : " + containerId);
         DockerRmCommand dockerRmCommand = new DockerRmCommand(containerId);
         DockerCommandExecutor.executeDockerCommand(dockerRmCommand, containerId,
-            null, super.getConf(), privOpExecutor, false);
+            null, super.getConf(), privOpExecutor, false, nmContext);
       }
     } catch (ContainerExecutionException e) {
       LOG.warn("Unable to remove docker container: " + containerId);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/DockerLinuxContainerRuntime.java

@@ -376,7 +376,7 @@ public class DockerLinuxContainerRuntime implements LinuxContainerRuntime {
       Container container) throws ContainerExecutionException {
     try {
       String commandFile = dockerClient.writeCommandToTempFile(
-          dockerVolumeCommand, container.getContainerId().toString());
+          dockerVolumeCommand, container, nmContext);
       PrivilegedOperation privOp = new PrivilegedOperation(
           PrivilegedOperation.OperationType.RUN_DOCKER_CMD);
       privOp.appendArgs(commandFile);
@@ -895,7 +895,7 @@ public class DockerLinuxContainerRuntime implements LinuxContainerRuntime {
     }
 
     String commandFile = dockerClient.writeCommandToTempFile(runCommand,
-        containerIdStr);
+        container, nmContext);
     PrivilegedOperation launchOp = buildLaunchOp(ctx,
         commandFile, runCommand);
 
@@ -1000,7 +1000,7 @@ public class DockerLinuxContainerRuntime implements LinuxContainerRuntime {
         new DockerInspectCommand(containerId).getIpAndHost();
     try {
       String commandFile = dockerClient.writeCommandToTempFile(inspectCommand,
-          containerId);
+          container, nmContext);
       PrivilegedOperation privOp = new PrivilegedOperation(
           PrivilegedOperation.OperationType.RUN_DOCKER_CMD);
       privOp.appendArgs(commandFile);
@@ -1142,11 +1142,11 @@ public class DockerLinuxContainerRuntime implements LinuxContainerRuntime {
       throws ContainerExecutionException {
     DockerCommandExecutor.DockerContainerStatus containerStatus =
         DockerCommandExecutor.getContainerStatus(containerId, conf,
-            privilegedOperationExecutor);
+            privilegedOperationExecutor, nmContext);
     if (DockerCommandExecutor.isStoppable(containerStatus)) {
       DockerStopCommand dockerStopCommand = new DockerStopCommand(containerId);
       DockerCommandExecutor.executeDockerCommand(dockerStopCommand, containerId,
-          env, conf, privilegedOperationExecutor, false);
+          env, conf, privilegedOperationExecutor, false, nmContext);
     } else {
       if (LOG.isDebugEnabled()) {
         LOG.debug(
@@ -1160,12 +1160,12 @@ public class DockerLinuxContainerRuntime implements LinuxContainerRuntime {
       ContainerExecutor.Signal signal) throws ContainerExecutionException {
     DockerCommandExecutor.DockerContainerStatus containerStatus =
         DockerCommandExecutor.getContainerStatus(containerId, conf,
-            privilegedOperationExecutor);
+            privilegedOperationExecutor, nmContext);
     if (DockerCommandExecutor.isKillable(containerStatus)) {
       DockerKillCommand dockerKillCommand =
           new DockerKillCommand(containerId).setSignal(signal.name());
       DockerCommandExecutor.executeDockerCommand(dockerKillCommand, containerId,
-          env, conf, privilegedOperationExecutor, false);
+          env, conf, privilegedOperationExecutor, false, nmContext);
     } else {
       if (LOG.isDebugEnabled()) {
         LOG.debug(
@@ -1185,11 +1185,11 @@ public class DockerLinuxContainerRuntime implements LinuxContainerRuntime {
     } else {
       DockerCommandExecutor.DockerContainerStatus containerStatus =
           DockerCommandExecutor.getContainerStatus(containerId, conf,
-              privilegedOperationExecutor);
+              privilegedOperationExecutor, nmContext);
       if (DockerCommandExecutor.isRemovable(containerStatus)) {
         DockerRmCommand dockerRmCommand = new DockerRmCommand(containerId);
         DockerCommandExecutor.executeDockerCommand(dockerRmCommand, containerId,
-            env, conf, privilegedOperationExecutor, false);
+            env, conf, privilegedOperationExecutor, false, nmContext);
       }
     }
   }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerClient.java

@@ -23,7 +23,13 @@ package org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.server.nodemanager.Context;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ResourceLocalizationService;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.runtime.ContainerExecutionException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -97,4 +103,54 @@ public final class DockerClient {
       throw new ContainerExecutionException(e);
     }
   }
+
+  public String writeCommandToTempFile(DockerCommand cmd, Container container,
+      Context nmContext) throws ContainerExecutionException {
+    ContainerId containerId = container.getContainerId();
+    String filePrefix = containerId.toString();
+    ApplicationId appId = containerId.getApplicationAttemptId()
+        .getApplicationId();
+    File dockerCommandFile;
+    String cmdDir = null;
+
+    if(nmContext == null || nmContext.getLocalDirsHandler() == null) {
+      throw new ContainerExecutionException(
+          "Unable to write temporary docker command");
+    }
+
+    try {
+      cmdDir = nmContext.getLocalDirsHandler().getLocalPathForWrite(
+          ResourceLocalizationService.NM_PRIVATE_DIR + Path.SEPARATOR +
+          appId + Path.SEPARATOR + filePrefix + Path.SEPARATOR).toString();
+
+      dockerCommandFile = File.createTempFile(TMP_FILE_PREFIX + filePrefix,
+          TMP_FILE_SUFFIX, new File(cmdDir));
+
+      Writer writer = new OutputStreamWriter(
+          new FileOutputStream(dockerCommandFile.toString()), "UTF-8");
+      PrintWriter printWriter = new PrintWriter(writer);
+      printWriter.println("[docker-command-execution]");
+      for (Map.Entry<String, List<String>> entry :
+          cmd.getDockerCommandWithArguments().entrySet()) {
+        if (entry.getKey().contains("=")) {
+          throw new ContainerExecutionException(
+              "'=' found in entry for docker command file, key = " + entry
+                  .getKey() + "; value = " + entry.getValue());
+        }
+        if (entry.getValue().contains("\n")) {
+          throw new ContainerExecutionException(
+              "'\\n' found in entry for docker command file, key = " + entry
+                  .getKey() + "; value = " + entry.getValue());
+        }
+        printWriter.println("  " + entry.getKey() + "=" + StringUtils
+            .join(",", entry.getValue()));
+      }
+      printWriter.close();
+
+      return dockerCommandFile.toString();
+    } catch (IOException e) {
+      LOG.warn("Unable to write docker command to " + cmdDir);
+      throw new ContainerExecutionException(e);
+    }
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/DockerCommandExecutor.java

@@ -17,6 +17,8 @@
 package org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.docker;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.yarn.api.records.ContainerId;
+import org.apache.hadoop.yarn.server.nodemanager.Context;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.privileged.PrivilegedOperation;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.privileged.PrivilegedOperationException;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.privileged.PrivilegedOperationExecutor;
@@ -76,11 +78,13 @@ public final class DockerCommandExecutor {
   public static String executeDockerCommand(DockerCommand dockerCommand,
       String containerId, Map<String, String> env, Configuration conf,
       PrivilegedOperationExecutor privilegedOperationExecutor,
-      boolean disableFailureLogging)
+      boolean disableFailureLogging, Context nmContext)
       throws ContainerExecutionException {
     DockerClient dockerClient = new DockerClient(conf);
     String commandFile =
-        dockerClient.writeCommandToTempFile(dockerCommand, containerId);
+        dockerClient.writeCommandToTempFile(dockerCommand,
+        nmContext.getContainers().get(ContainerId.fromString(containerId)),
+        nmContext);
     PrivilegedOperation dockerOp = new PrivilegedOperation(
         PrivilegedOperation.OperationType.RUN_DOCKER_CMD);
     dockerOp.appendArgs(commandFile);
@@ -116,11 +120,13 @@ public final class DockerCommandExecutor {
    */
   public static DockerContainerStatus getContainerStatus(String containerId,
       Configuration conf,
-      PrivilegedOperationExecutor privilegedOperationExecutor) {
+      PrivilegedOperationExecutor privilegedOperationExecutor,
+      Context nmContext) {
     try {
       DockerContainerStatus dockerContainerStatus;
       String currentContainerStatus =
-          executeStatusCommand(containerId, conf, privilegedOperationExecutor);
+          executeStatusCommand(containerId, conf,
+          privilegedOperationExecutor, nmContext);
       if (currentContainerStatus == null) {
         dockerContainerStatus = DockerContainerStatus.UNKNOWN;
       } else if (currentContainerStatus
@@ -177,13 +183,15 @@ public final class DockerCommandExecutor {
    */
   private static String executeStatusCommand(String containerId,
       Configuration conf,
-      PrivilegedOperationExecutor privilegedOperationExecutor)
+      PrivilegedOperationExecutor privilegedOperationExecutor,
+      Context nmContext)
       throws ContainerExecutionException {
     DockerInspectCommand dockerInspectCommand =
         new DockerInspectCommand(containerId).getContainerStatus();
     try {
       return DockerCommandExecutor.executeDockerCommand(dockerInspectCommand,
-          containerId, null, conf, privilegedOperationExecutor, true);
+          containerId, null, conf, privilegedOperationExecutor, true,
+          nmContext);
     } catch (ContainerExecutionException e) {
       throw new ContainerExecutionException(e);
     }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.c

@@ -1279,6 +1279,15 @@ char *construct_docker_command(const char *command_file) {
   int ret = 0;
   size_t command_size = MIN(sysconf(_SC_ARG_MAX), 128*1024);
   char *buffer = alloc_and_clear_memory(command_size, sizeof(char));
+
+  uid_t user = geteuid();
+  gid_t group = getegid();
+  if (change_effective_user(nm_uid, nm_gid) != 0) {
+    fprintf(ERRORFILE, "Cannot change effective user to nm");
+    fflush(ERRORFILE);
+    exit(SETUID_OPER_FAILED);
+  }
+
   ret = get_docker_command(command_file, &CFG, buffer, command_size);
   if (ret != 0) {
     fprintf(ERRORFILE, "Error constructing docker command, docker error code=%d, error message='%s'\n", ret,
@@ -1286,6 +1295,13 @@ char *construct_docker_command(const char *command_file) {
     fflush(ERRORFILE);
     exit(DOCKER_RUN_FAILED);
   }
+
+  if (change_effective_user(user, group)) {
+    fprintf(ERRORFILE, "Cannot change effective user from nm back to original");
+    fflush(ERRORFILE);
+    exit(SETUID_OPER_FAILED);
+  }
+
   return buffer;
 }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/utils/docker-util.c

@@ -341,6 +341,7 @@ int docker_module_enabled(const struct configuration *conf) {
 int get_docker_command(const char *command_file, const struct configuration *conf, char *out, const size_t outlen) {
   int ret = 0;
   struct configuration command_config = {0, NULL};
+
   ret = read_config(command_file, &command_config);
   if (ret != 0) {
     return INVALID_COMMAND_FILE;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/TestDockerContainerRuntime.java

@@ -30,8 +30,10 @@ import org.apache.hadoop.registry.client.binding.RegistryPathUtils;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.util.Shell;
 import org.apache.hadoop.util.StringUtils;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
+import org.apache.hadoop.yarn.server.nodemanager.LocalDirsHandlerService;
 import org.apache.hadoop.yarn.util.DockerClientConfigHandler;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.security.TestDockerClientConfigHandler;
@@ -82,6 +84,7 @@ import java.util.List;
 import java.util.Map;
 import java.util.Random;
 import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
 
 import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.LinuxContainerRuntimeConstants.APPID;
 import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.LinuxContainerRuntimeConstants.APPLICATION_LOCAL_DIRS;
@@ -101,6 +104,7 @@ import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.r
 import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.LinuxContainerRuntimeConstants.SIGNAL;
 import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.LinuxContainerRuntimeConstants.USER;
 import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.LinuxContainerRuntimeConstants.USER_FILECACHE_DIRS;
+import static org.mockito.Matchers.anyString;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.anyBoolean;
 import static org.mockito.Mockito.anyList;
@@ -120,7 +124,9 @@ public class TestDockerContainerRuntime {
   private String containerId;
   private Container container;
   private ContainerId cId;
+  private ApplicationAttemptId appAttemptId;
   private ContainerLaunchContext context;
+  private Context nmContext;
   private HashMap<String, String> env;
   private String image;
   private String uidGidPair;
@@ -158,17 +164,20 @@ public class TestDockerContainerRuntime {
     mockExecutor = Mockito
         .mock(PrivilegedOperationExecutor.class);
     mockCGroupsHandler = Mockito.mock(CGroupsHandler.class);
-    containerId = "container_id";
+    containerId = "container_e11_1518975676334_14532816_01_000001";
     container = mock(Container.class);
     cId = mock(ContainerId.class);
+    appAttemptId = mock(ApplicationAttemptId.class);
     context = mock(ContainerLaunchContext.class);
     env = new HashMap<String, String>();
     env.put("FROM_CLIENT", "1");
     image = "busybox:latest";
+    nmContext = createMockNMContext();
 
     env.put(DockerLinuxContainerRuntime.ENV_DOCKER_CONTAINER_IMAGE, image);
     when(container.getContainerId()).thenReturn(cId);
     when(cId.toString()).thenReturn(containerId);
+    when(cId.getApplicationAttemptId()).thenReturn(appAttemptId);
     when(container.getLaunchContext()).thenReturn(context);
     when(context.getEnvironment()).thenReturn(env);
     when(container.getUser()).thenReturn(submittingUser);
@@ -257,6 +266,34 @@ public class TestDockerContainerRuntime {
         .setExecutionAttribute(RESOURCES_OPTIONS, resourcesOptions);
   }
 
+  public Context createMockNMContext() {
+    Context mockNMContext = mock(Context.class);
+    LocalDirsHandlerService localDirsHandler =
+        mock(LocalDirsHandlerService.class);
+    ResourcePluginManager resourcePluginManager =
+        mock(ResourcePluginManager.class);
+
+    String tmpPath = new StringBuffer(System.getProperty("test.build.data"))
+        .append('/').append("hadoop.tmp.dir").toString();
+
+    ConcurrentMap<ContainerId, Container> containerMap =
+        mock(ConcurrentMap.class);
+
+    when(mockNMContext.getLocalDirsHandler()).thenReturn(localDirsHandler);
+    when(mockNMContext.getResourcePluginManager())
+        .thenReturn(resourcePluginManager);
+    when(mockNMContext.getContainers()).thenReturn(containerMap);
+    when(containerMap.get(any())).thenReturn(container);
+
+    try {
+      when(localDirsHandler.getLocalPathForWrite(anyString()))
+          .thenReturn(new Path(tmpPath));
+    } catch (IOException ioe) {
+      LOG.info("LocalDirsHandler failed" + ioe);
+    }
+    return mockNMContext;
+  }
+
   @Test
   public void testSelectDockerContainerType() {
     Map<String, String> envDockerType = new HashMap<>();
@@ -333,7 +370,7 @@ public class TestDockerContainerRuntime {
       IOException {
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
     runtime.launchContainer(builder.build());
 
     PrivilegedOperation op = capturePrivilegedOperationAndVerifyArgs();
@@ -360,7 +397,9 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals(
         "  launch-command=bash,/test_container_work_dir/launch_container.sh",
         dockerCommands.get(counter++));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(counter++));
     Assert.assertEquals("  net=host", dockerCommands.get(counter++));
     Assert.assertEquals("  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
         + "/test_user_filecache_dir:/test_user_filecache_dir",
@@ -382,7 +421,7 @@ public class TestDockerContainerRuntime {
         true);
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
     runtime.launchContainer(builder.build());
 
     PrivilegedOperation op = capturePrivilegedOperationAndVerifyArgs();
@@ -408,7 +447,9 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals(
         "  launch-command=bash,/test_container_work_dir/launch_container.sh",
         dockerCommands.get(counter++));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(counter++));
     Assert
         .assertEquals("  net=host", dockerCommands.get(counter++));
     Assert.assertEquals("  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
@@ -431,7 +472,7 @@ public class TestDockerContainerRuntime {
 
     DockerLinuxContainerRuntime runtime =
         new DockerLinuxContainerRuntime(mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     //invalid default network configuration - sdn2 is included in allowed
     // networks
@@ -447,7 +488,7 @@ public class TestDockerContainerRuntime {
     try {
       runtime =
           new DockerLinuxContainerRuntime(mockExecutor, mockCGroupsHandler);
-      runtime.initialize(conf, null);
+      runtime.initialize(conf, nmContext);
       Assert.fail("Invalid default network configuration should did not "
           + "trigger initialization failure.");
     } catch (ContainerExecutionException e) {
@@ -463,7 +504,7 @@ public class TestDockerContainerRuntime {
         validDefaultNetwork);
     runtime =
         new DockerLinuxContainerRuntime(mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
   }
 
   @Test
@@ -473,7 +514,7 @@ public class TestDockerContainerRuntime {
       PrivilegedOperationException {
     DockerLinuxContainerRuntime runtime =
         new DockerLinuxContainerRuntime(mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     Random randEngine = new Random();
     String disallowedNetwork = "sdn" + Integer.toString(randEngine.nextInt());
@@ -524,7 +565,9 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals(
         "  launch-command=bash,/test_container_work_dir/launch_container.sh",
         dockerCommands.get(counter++));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(counter++));
     Assert
         .assertEquals("  net=" + allowedNetwork, dockerCommands.get(counter++));
     Assert.assertEquals("  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
@@ -549,7 +592,7 @@ public class TestDockerContainerRuntime {
     conf.setBoolean(RegistryConstants.KEY_DNS_ENABLED, true);
     DockerLinuxContainerRuntime runtime =
         new DockerLinuxContainerRuntime(mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     String expectedHostname = "test.hostname";
     env.put(DockerLinuxContainerRuntime.ENV_DOCKER_CONTAINER_HOSTNAME,
@@ -582,7 +625,9 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals(
         "  launch-command=bash,/test_container_work_dir/launch_container.sh",
         dockerCommands.get(counter++));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(counter++));
     Assert
         .assertEquals("  net=host", dockerCommands.get(counter++));
     Assert.assertEquals("  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
@@ -620,7 +665,7 @@ public class TestDockerContainerRuntime {
         customNetwork1);
 
     //this should cause no failures.
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
     runtime.launchContainer(builder.build());
     PrivilegedOperation op = capturePrivilegedOperationAndVerifyArgs();
     List<String> args = op.getArguments();
@@ -643,13 +688,17 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals("  docker-command=run", dockerCommands.get(counter++));
     Assert.assertEquals("  group-add=" + String.join(",", groups),
         dockerCommands.get(counter++));
-    Assert.assertEquals("  hostname=ctr-id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  hostname=ctr-e11-1518975676334-14532816-01-000001",
+        dockerCommands.get(counter++));
     Assert
         .assertEquals("  image=busybox:latest", dockerCommands.get(counter++));
     Assert.assertEquals(
         "  launch-command=bash,/test_container_work_dir/launch_container.sh",
         dockerCommands.get(counter++));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(counter++));
     Assert.assertEquals("  net=sdn1", dockerCommands.get(counter++));
     Assert.assertEquals("  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
             + "/test_user_filecache_dir:/test_user_filecache_dir",
@@ -688,14 +737,18 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals("  docker-command=run", dockerCommands.get(counter++));
     Assert.assertEquals("  group-add=" + String.join(",", groups),
         dockerCommands.get(counter++));
-    Assert.assertEquals("  hostname=ctr-id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  hostname=ctr-e11-1518975676334-14532816-01-000001",
+        dockerCommands.get(counter++));
     Assert
         .assertEquals("  image=busybox:latest", dockerCommands.get(counter++));
     Assert.assertEquals(
         "  launch-command=bash,/test_container_work_dir/launch_container.sh",
         dockerCommands.get(counter++));
 
-    Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(counter++));
     Assert.assertEquals("  net=sdn2", dockerCommands.get(counter++));
     Assert.assertEquals("  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
             + "/test_user_filecache_dir:/test_user_filecache_dir",
@@ -728,7 +781,7 @@ public class TestDockerContainerRuntime {
       IOException {
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(DockerLinuxContainerRuntime
         .ENV_DOCKER_CONTAINER_PID_NAMESPACE, "invalid-value");
@@ -756,7 +809,7 @@ public class TestDockerContainerRuntime {
       throws ContainerExecutionException {
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(DockerLinuxContainerRuntime
         .ENV_DOCKER_CONTAINER_PID_NAMESPACE, "host");
@@ -779,7 +832,7 @@ public class TestDockerContainerRuntime {
 
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(DockerLinuxContainerRuntime
         .ENV_DOCKER_CONTAINER_PID_NAMESPACE, "host");
@@ -809,7 +862,9 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals(
         "  launch-command=bash,/test_container_work_dir/launch_container.sh",
         dockerCommands.get(counter++));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(counter++));
     Assert.assertEquals("  net=host", dockerCommands.get(counter++));
     Assert.assertEquals("  pid=host", dockerCommands.get(counter++));
     Assert.assertEquals("  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
@@ -830,7 +885,7 @@ public class TestDockerContainerRuntime {
       IOException {
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(DockerLinuxContainerRuntime
             .ENV_DOCKER_CONTAINER_RUN_PRIVILEGED_CONTAINER, "invalid-value");
@@ -858,7 +913,7 @@ public class TestDockerContainerRuntime {
       throws ContainerExecutionException {
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(DockerLinuxContainerRuntime
             .ENV_DOCKER_CONTAINER_RUN_PRIVILEGED_CONTAINER, "true");
@@ -880,7 +935,7 @@ public class TestDockerContainerRuntime {
 
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(DockerLinuxContainerRuntime
             .ENV_DOCKER_CONTAINER_RUN_PRIVILEGED_CONTAINER, "true");
@@ -909,7 +964,7 @@ public class TestDockerContainerRuntime {
 
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(DockerLinuxContainerRuntime
             .ENV_DOCKER_CONTAINER_RUN_PRIVILEGED_CONTAINER, "true");
@@ -936,7 +991,7 @@ public class TestDockerContainerRuntime {
 
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(DockerLinuxContainerRuntime
             .ENV_DOCKER_CONTAINER_RUN_PRIVILEGED_CONTAINER, "true");
@@ -964,7 +1019,9 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals(
         "  launch-command=bash,/test_container_work_dir/launch_container.sh",
         dockerCommands.get(counter++));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(counter++));
     Assert.assertEquals("  net=host", dockerCommands.get(counter++));
     Assert.assertEquals("  privileged=true", dockerCommands.get(counter++));
     Assert.assertEquals("  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
@@ -988,7 +1045,7 @@ public class TestDockerContainerRuntime {
 
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime
         (mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     String resourceOptionsNone = "cgroups=none";
     DockerRunCommand command = Mockito.mock(DockerRunCommand.class);
@@ -1015,7 +1072,7 @@ public class TestDockerContainerRuntime {
 
     runtime = new DockerLinuxContainerRuntime
         (mockExecutor, null);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     runtime.addCGroupParentIfRequired(resourceOptionsNone, containerIdStr,
         command);
@@ -1030,7 +1087,7 @@ public class TestDockerContainerRuntime {
   public void testMountSourceOnly() throws ContainerExecutionException {
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(
         DockerLinuxContainerRuntime.ENV_DOCKER_CONTAINER_LOCAL_RESOURCE_MOUNTS,
@@ -1050,7 +1107,7 @@ public class TestDockerContainerRuntime {
       IOException {
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(
         DockerLinuxContainerRuntime.ENV_DOCKER_CONTAINER_LOCAL_RESOURCE_MOUNTS,
@@ -1081,7 +1138,9 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals(
         "  launch-command=bash,/test_container_work_dir/launch_container.sh",
         dockerCommands.get(counter++));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(counter++));
     Assert.assertEquals("  net=host", dockerCommands.get(counter++));
     Assert.assertEquals(
         "  ro-mounts=/test_filecache_dir:/test_filecache_dir,/"
@@ -1101,7 +1160,7 @@ public class TestDockerContainerRuntime {
   public void testMountInvalid() throws ContainerExecutionException {
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(
         DockerLinuxContainerRuntime.ENV_DOCKER_CONTAINER_LOCAL_RESOURCE_MOUNTS,
@@ -1121,7 +1180,7 @@ public class TestDockerContainerRuntime {
       IOException {
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(
         DockerLinuxContainerRuntime.ENV_DOCKER_CONTAINER_LOCAL_RESOURCE_MOUNTS,
@@ -1153,7 +1212,9 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals(
         "  launch-command=bash,/test_container_work_dir/launch_container.sh",
         dockerCommands.get(counter++));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(counter++));
     Assert.assertEquals("  net=host", dockerCommands.get(counter++));
     Assert.assertEquals(
         "  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
@@ -1176,7 +1237,7 @@ public class TestDockerContainerRuntime {
       IOException {
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(
         DockerLinuxContainerRuntime.ENV_DOCKER_CONTAINER_MOUNTS,
@@ -1207,7 +1268,9 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals(
         "  launch-command=bash,/test_container_work_dir/launch_container.sh",
         dockerCommands.get(counter++));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(counter++));
     Assert.assertEquals("  net=host", dockerCommands.get(counter++));
     Assert.assertEquals("  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
             + "/test_user_filecache_dir:/test_user_filecache_dir,"
@@ -1227,7 +1290,7 @@ public class TestDockerContainerRuntime {
   public void testUserMountInvalid() throws ContainerExecutionException {
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(
         DockerLinuxContainerRuntime.ENV_DOCKER_CONTAINER_MOUNTS,
@@ -1245,7 +1308,7 @@ public class TestDockerContainerRuntime {
   public void testUserMountModeInvalid() throws ContainerExecutionException {
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(
         DockerLinuxContainerRuntime.ENV_DOCKER_CONTAINER_MOUNTS,
@@ -1263,7 +1326,7 @@ public class TestDockerContainerRuntime {
   public void testUserMountModeNulInvalid() throws ContainerExecutionException {
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     env.put(
         DockerLinuxContainerRuntime.ENV_DOCKER_CONTAINER_MOUNTS,
@@ -1310,7 +1373,9 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals(3, dockerCommands.size());
     Assert.assertEquals("[docker-command-execution]", dockerCommands.get(0));
     Assert.assertEquals("  docker-command=stop", dockerCommands.get(1));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(2));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(2));
   }
 
   @Test
@@ -1323,7 +1388,9 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals(3, dockerCommands.size());
     Assert.assertEquals("[docker-command-execution]", dockerCommands.get(0));
     Assert.assertEquals("  docker-command=stop", dockerCommands.get(1));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(2));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(2));
   }
 
   @Test
@@ -1334,7 +1401,9 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals(4, dockerCommands.size());
     Assert.assertEquals("[docker-command-execution]", dockerCommands.get(0));
     Assert.assertEquals("  docker-command=kill", dockerCommands.get(1));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(2));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(2));
     Assert.assertEquals("  signal=QUIT", dockerCommands.get(3));
   }
 
@@ -1553,7 +1622,7 @@ public class TestDockerContainerRuntime {
             any(File.class), anyMap(), anyBoolean(), anyBoolean())).thenReturn(
         dockerVolumeListOutput);
 
-    Context nmContext = mock(Context.class);
+    Context mockNMContext = createMockNMContext();
     ResourcePluginManager rpm = mock(ResourcePluginManager.class);
     Map<String, ResourcePlugin> pluginsMap = new HashMap<>();
     ResourcePlugin plugin1 = mock(ResourcePlugin.class);
@@ -1570,9 +1639,9 @@ public class TestDockerContainerRuntime {
 
     when(rpm.getNameToPlugins()).thenReturn(pluginsMap);
 
-    when(nmContext.getResourcePluginManager()).thenReturn(rpm);
+    when(mockNMContext.getResourcePluginManager()).thenReturn(rpm);
 
-    runtime.initialize(conf, nmContext);
+    runtime.initialize(conf, mockNMContext);
 
     ContainerRuntimeContext containerRuntimeContext = builder.build();
 
@@ -1650,7 +1719,7 @@ public class TestDockerContainerRuntime {
             any(File.class), anyMap(), anyBoolean(), anyBoolean())).thenReturn(
         "volume1,local");
 
-    Context nmContext = mock(Context.class);
+    Context mockNMContext = createMockNMContext();
     ResourcePluginManager rpm = mock(ResourcePluginManager.class);
     Map<String, ResourcePlugin> pluginsMap = new HashMap<>();
     ResourcePlugin plugin1 = mock(ResourcePlugin.class);
@@ -1667,9 +1736,9 @@ public class TestDockerContainerRuntime {
 
     when(rpm.getNameToPlugins()).thenReturn(pluginsMap);
 
-    when(nmContext.getResourcePluginManager()).thenReturn(rpm);
+    when(mockNMContext.getResourcePluginManager()).thenReturn(rpm);
 
-    runtime.initialize(conf, nmContext);
+    runtime.initialize(conf, mockNMContext);
 
     ContainerRuntimeContext containerRuntimeContext = builder.build();
 
@@ -1701,7 +1770,9 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals(
         "  launch-command=bash,/test_container_work_dir/launch_container.sh",
         dockerCommands.get(counter++));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(counter++));
     Assert.assertEquals("  net=host", dockerCommands.get(counter++));
     Assert.assertEquals("  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
             + "/test_user_filecache_dir:/test_user_filecache_dir,"
@@ -1727,7 +1798,7 @@ public class TestDockerContainerRuntime {
     try {
       conf.setStrings(YarnConfiguration.NM_DOCKER_CONTAINER_CAPABILITIES,
           "none", "CHOWN", "DAC_OVERRIDE");
-      runtime.initialize(conf, null);
+      runtime.initialize(conf, nmContext);
       Assert.fail("Initialize didn't fail with invalid capabilities " +
           "'none', 'CHOWN', 'DAC_OVERRIDE'");
     } catch (ContainerExecutionException e) {
@@ -1736,7 +1807,7 @@ public class TestDockerContainerRuntime {
     try {
       conf.setStrings(YarnConfiguration.NM_DOCKER_CONTAINER_CAPABILITIES,
           "CHOWN", "DAC_OVERRIDE", "NONE");
-      runtime.initialize(conf, null);
+      runtime.initialize(conf, nmContext);
       Assert.fail("Initialize didn't fail with invalid capabilities " +
           "'CHOWN', 'DAC_OVERRIDE', 'NONE'");
     } catch (ContainerExecutionException e) {
@@ -1744,17 +1815,17 @@ public class TestDockerContainerRuntime {
 
     conf.setStrings(YarnConfiguration.NM_DOCKER_CONTAINER_CAPABILITIES,
         "NONE");
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
     Assert.assertEquals(0, runtime.getCapabilities().size());
 
     conf.setStrings(YarnConfiguration.NM_DOCKER_CONTAINER_CAPABILITIES,
         "none");
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
     Assert.assertEquals(0, runtime.getCapabilities().size());
 
     conf.setStrings(YarnConfiguration.NM_DOCKER_CONTAINER_CAPABILITIES,
         "CHOWN", "DAC_OVERRIDE");
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
     Iterator<String> it = runtime.getCapabilities().iterator();
     Assert.assertEquals("CHOWN", it.next());
     Assert.assertEquals("DAC_OVERRIDE", it.next());
@@ -1782,7 +1853,7 @@ public class TestDockerContainerRuntime {
     when(context.getTokens()).thenReturn(tokens);
     DockerLinuxContainerRuntime runtime =
         new DockerLinuxContainerRuntime(mockExecutor, mockCGroupsHandler);
-    runtime.initialize(conf, null);
+    runtime.initialize(conf, nmContext);
 
     Set<PosixFilePermission> perms =
         PosixFilePermissions.fromString("rwxr-xr--");
@@ -1842,7 +1913,9 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals(
         "  launch-command=bash,/test_container_work_dir/launch_container.sh",
         dockerCommands.get(counter++));
-    Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
+    Assert.assertEquals(
+        "  name=container_e11_1518975676334_14532816_01_000001",
+        dockerCommands.get(counter++));
     Assert.assertEquals("  net=host", dockerCommands.get(counter++));
     Assert.assertEquals("  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
             + "/test_user_filecache_dir:/test_user_filecache_dir",
@@ -1885,7 +1958,8 @@ public class TestDockerContainerRuntime {
             DockerStopCommand dockerStopCommand =
                 new DockerStopCommand(containerName);
             DockerCommandExecutor.executeDockerCommand(dockerStopCommand,
-                containerName, environment, conf, mockExecutor, false);
+                containerName, environment, conf, mockExecutor, false,
+                nmContext);
           }
         } else {
           if (DockerCommandExecutor.isKillable(containerStatus)) {
@@ -1893,7 +1967,8 @@ public class TestDockerContainerRuntime {
                 new DockerKillCommand(containerName);
             dockerKillCommand.setSignal(signal.name());
             DockerCommandExecutor.executeDockerCommand(dockerKillCommand,
-                containerName, environment, conf, mockExecutor, false);
+                containerName, environment, conf, mockExecutor, false,
+                nmContext);
           }
         }
       } catch (ContainerExecutionException e) {
@@ -1916,7 +1991,7 @@ public class TestDockerContainerRuntime {
           DockerRmCommand dockerRmCommand = new DockerRmCommand(containerId);
           DockerCommandExecutor
               .executeDockerCommand(dockerRmCommand, containerId, env, conf,
-                  privilegedOperationExecutor, false);
+                  privilegedOperationExecutor, false, nmContext);
         }
       }
     }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/linux/runtime/docker/TestDockerCommandExecutor.java

@@ -17,9 +17,13 @@
 package org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.docker;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
 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.container.Container;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.privileged.MockPrivilegedOperationCaptor;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.privileged.PrivilegedOperation;
@@ -39,13 +43,16 @@ import java.nio.file.Paths;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.List;
+import java.util.concurrent.ConcurrentMap;
 
 import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.LinuxContainerRuntimeConstants.CONTAINER_ID_STR;
 import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.docker.DockerCommandExecutor.DockerContainerStatus;
+import static org.eclipse.jetty.server.handler.gzip.GzipHttpOutputInterceptor.LOG;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -55,7 +62,8 @@ import static org.mockito.Mockito.when;
  */
 public class TestDockerCommandExecutor {
 
-  private static final String MOCK_CONTAINER_ID = "container_id";
+  private static final String MOCK_CONTAINER_ID =
+      "container_e11_1861047502093_13763105_01_000001";
   private static final String MOCK_LOCAL_IMAGE_NAME = "local_image_name";
   private static final String MOCK_IMAGE_NAME = "image_name";
 
@@ -68,21 +76,29 @@ public class TestDockerCommandExecutor {
   private ContainerId cId;
   private ContainerLaunchContext context;
   private HashMap<String, String> env;
+  private Context nmContext;
+  private ApplicationAttemptId appAttemptId;
 
   @Before
   public void setUp() throws Exception {
     mockExecutor = mock(PrivilegedOperationExecutor.class);
     mockCGroupsHandler = mock(CGroupsHandler.class);
     configuration = new Configuration();
+    String tmpPath = new StringBuffer(System.getProperty("test.build.data"))
+        .append('/').append("hadoop.tmp.dir").toString();
+    configuration.set("hadoop.tmp.dir", tmpPath);
     runtime = new DockerLinuxContainerRuntime(mockExecutor, mockCGroupsHandler);
     container = mock(Container.class);
     cId = mock(ContainerId.class);
     context = mock(ContainerLaunchContext.class);
     env = new HashMap<>();
     builder = new ContainerRuntimeContext.Builder(container);
+    nmContext = createMockNMContext();
+    appAttemptId = mock(ApplicationAttemptId.class);
 
     when(container.getContainerId()).thenReturn(cId);
     when(cId.toString()).thenReturn(MOCK_CONTAINER_ID);
+    when(cId.getApplicationAttemptId()).thenReturn(appAttemptId);
     when(container.getLaunchContext()).thenReturn(context);
     when(context.getEnvironment()).thenReturn(env);
 
@@ -92,12 +108,37 @@ public class TestDockerCommandExecutor {
         null);
   }
 
+
+  public Context createMockNMContext() {
+    Context mockNMContext = mock(Context.class);
+    LocalDirsHandlerService localDirsHandler =
+        mock(LocalDirsHandlerService.class);
+
+    String tmpPath = new StringBuffer(System.getProperty("test.build.data"))
+        .append('/').append("hadoop.tmp.dir").toString();
+
+    ConcurrentMap<ContainerId, Container> containerMap =
+        mock(ConcurrentMap.class);
+
+    when(mockNMContext.getLocalDirsHandler()).thenReturn(localDirsHandler);
+    when(mockNMContext.getContainers()).thenReturn(containerMap);
+    when(containerMap.get(any())).thenReturn(container);
+
+    try {
+      when(localDirsHandler.getLocalPathForWrite(anyString()))
+          .thenReturn(new Path(tmpPath));
+    } catch (IOException ioe) {
+      LOG.info("LocalDirsHandler failed" + ioe);
+    }
+    return mockNMContext;
+  }
+
   @Test
   public void testExecuteDockerCommand() throws Exception {
     DockerStopCommand dockerStopCommand =
         new DockerStopCommand(MOCK_CONTAINER_ID);
     DockerCommandExecutor.executeDockerCommand(dockerStopCommand,
-        cId.toString(), env, configuration, mockExecutor, false);
+        cId.toString(), env, configuration, mockExecutor, false, nmContext);
     List<PrivilegedOperation> ops = MockPrivilegedOperationCaptor
         .capturePrivilegedOperations(mockExecutor, 1, true);
     assertEquals(1, ops.size());
@@ -109,7 +150,7 @@ public class TestDockerCommandExecutor {
   public void testExecuteDockerRm() throws Exception {
     DockerRmCommand dockerCommand = new DockerRmCommand(MOCK_CONTAINER_ID);
     DockerCommandExecutor.executeDockerCommand(dockerCommand, MOCK_CONTAINER_ID,
-        env, configuration, mockExecutor, false);
+        env, configuration, mockExecutor, false, nmContext);
     List<PrivilegedOperation> ops = MockPrivilegedOperationCaptor
         .capturePrivilegedOperations(mockExecutor, 1, true);
     List<String> dockerCommands = getValidatedDockerCommands(ops);
@@ -126,7 +167,7 @@ public class TestDockerCommandExecutor {
   public void testExecuteDockerStop() throws Exception {
     DockerStopCommand dockerCommand = new DockerStopCommand(MOCK_CONTAINER_ID);
     DockerCommandExecutor.executeDockerCommand(dockerCommand, MOCK_CONTAINER_ID,
-        env, configuration, mockExecutor, false);
+        env, configuration, mockExecutor, false, nmContext);
     List<PrivilegedOperation> ops = MockPrivilegedOperationCaptor
         .capturePrivilegedOperations(mockExecutor, 1, true);
     List<String> dockerCommands = getValidatedDockerCommands(ops);
@@ -144,7 +185,7 @@ public class TestDockerCommandExecutor {
     DockerInspectCommand dockerCommand =
         new DockerInspectCommand(MOCK_CONTAINER_ID).getContainerStatus();
     DockerCommandExecutor.executeDockerCommand(dockerCommand, MOCK_CONTAINER_ID,
-        env, configuration, mockExecutor, false);
+        env, configuration, mockExecutor, false, nmContext);
     List<PrivilegedOperation> ops = MockPrivilegedOperationCaptor
         .capturePrivilegedOperations(mockExecutor, 1, true);
     List<String> dockerCommands = getValidatedDockerCommands(ops);
@@ -164,7 +205,7 @@ public class TestDockerCommandExecutor {
     DockerPullCommand dockerCommand =
         new DockerPullCommand(MOCK_IMAGE_NAME);
     DockerCommandExecutor.executeDockerCommand(dockerCommand, MOCK_CONTAINER_ID,
-        env, configuration, mockExecutor, false);
+        env, configuration, mockExecutor, false, nmContext);
     List<PrivilegedOperation> ops = MockPrivilegedOperationCaptor
         .capturePrivilegedOperations(mockExecutor, 1, true);
     List<String> dockerCommands = getValidatedDockerCommands(ops);
@@ -182,7 +223,7 @@ public class TestDockerCommandExecutor {
     DockerLoadCommand dockerCommand =
         new DockerLoadCommand(MOCK_LOCAL_IMAGE_NAME);
     DockerCommandExecutor.executeDockerCommand(dockerCommand, MOCK_CONTAINER_ID,
-        env, configuration, mockExecutor, false);
+        env, configuration, mockExecutor, false, nmContext);
     List<PrivilegedOperation> ops = MockPrivilegedOperationCaptor
         .capturePrivilegedOperations(mockExecutor, 1, true);
     List<String> dockerCommands = getValidatedDockerCommands(ops);
@@ -204,7 +245,7 @@ public class TestDockerCommandExecutor {
           any(PrivilegedOperation.class), eq(null), any(), eq(true), eq(false)))
           .thenReturn(status.getName());
       assertEquals(status, DockerCommandExecutor.getContainerStatus(
-          MOCK_CONTAINER_ID, configuration, mockExecutor));
+          MOCK_CONTAINER_ID, configuration, mockExecutor, nmContext));
     }
   }
 
@@ -214,7 +255,7 @@ public class TestDockerCommandExecutor {
         new DockerKillCommand(MOCK_CONTAINER_ID)
             .setSignal(ContainerExecutor.Signal.QUIT.name());
     DockerCommandExecutor.executeDockerCommand(dockerKillCommand,
-        MOCK_CONTAINER_ID, env, configuration, mockExecutor, false);
+        MOCK_CONTAINER_ID, env, configuration, mockExecutor, false, nmContext);
     List<PrivilegedOperation> ops = MockPrivilegedOperationCaptor
         .capturePrivilegedOperations(mockExecutor, 1, true);
     List<String> dockerCommands = getValidatedDockerCommands(ops);
@@ -235,7 +276,7 @@ public class TestDockerCommandExecutor {
         new DockerKillCommand(MOCK_CONTAINER_ID)
             .setSignal(ContainerExecutor.Signal.KILL.name());
     DockerCommandExecutor.executeDockerCommand(dockerKillCommand,
-        MOCK_CONTAINER_ID, env, configuration, mockExecutor, false);
+        MOCK_CONTAINER_ID, env, configuration, mockExecutor, false, nmContext);
     List<PrivilegedOperation> ops = MockPrivilegedOperationCaptor
         .capturePrivilegedOperations(mockExecutor, 1, true);
     List<String> dockerCommands = getValidatedDockerCommands(ops);
@@ -256,7 +297,7 @@ public class TestDockerCommandExecutor {
         new DockerKillCommand(MOCK_CONTAINER_ID)
             .setSignal(ContainerExecutor.Signal.TERM.name());
     DockerCommandExecutor.executeDockerCommand(dockerKillCommand,
-        MOCK_CONTAINER_ID, env, configuration, mockExecutor, false);
+        MOCK_CONTAINER_ID, env, configuration, mockExecutor, false, nmContext);
     List<PrivilegedOperation> ops = MockPrivilegedOperationCaptor
         .capturePrivilegedOperations(mockExecutor, 1, true);
     List<String> dockerCommands = getValidatedDockerCommands(ops);