YARN-7815. Make the YARN mounts added to Docker containers more restrictive. Contributed by Shane Kumpf

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

@@ -651,6 +651,9 @@ public class LinuxContainerExecutor extends ContainerExecutor {
       .setExecutionAttribute(FILECACHE_DIRS, ctx.getFilecacheDirs())
       .setExecutionAttribute(USER_LOCAL_DIRS, ctx.getUserLocalDirs())
       .setExecutionAttribute(CONTAINER_LOCAL_DIRS, ctx.getContainerLocalDirs())
+      .setExecutionAttribute(USER_FILECACHE_DIRS, ctx.getUserFilecacheDirs())
+      .setExecutionAttribute(APPLICATION_LOCAL_DIRS,
+          ctx.getApplicationLocalDirs())
       .setExecutionAttribute(CONTAINER_LOG_DIRS, ctx.getContainerLogDirs())
       .setExecutionAttribute(RESOURCES_OPTIONS, resourcesOptions);
 

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

@@ -169,6 +169,17 @@ public class ContainerLaunch implements Callable<Integer> {
     return var;
   }
 
+  private Map<String, String> expandAllEnvironmentVars(
+      ContainerLaunchContext launchContext, Path containerLogDir) {
+    Map<String, String> environment = launchContext.getEnvironment();
+    for (Entry<String, String> entry : environment.entrySet()) {
+      String value = entry.getValue();
+      value = expandEnvironment(value, containerLogDir);
+      entry.setValue(value);
+    }
+    return environment;
+  }
+
   @Override
   @SuppressWarnings("unchecked") // dispatcher not typed
   public Integer call() {
@@ -202,13 +213,8 @@ public class ContainerLaunch implements Callable<Integer> {
       }
       launchContext.setCommands(newCmds);
 
-      Map<String, String> environment = launchContext.getEnvironment();
+      Map<String, String> environment = expandAllEnvironmentVars(
-      // Make a copy of env to iterate & do variable expansion
+          launchContext, containerLogDir);
-      for (Entry<String, String> entry : environment.entrySet()) {
-        String value = entry.getValue();
-        value = expandEnvironment(value, containerLogDir);
-        entry.setValue(value);
-      }
       // /////////////////////////// End of variable expansion
 
       FileContext lfs = FileContext.getLocalFSFileContext();
@@ -237,6 +243,9 @@ public class ContainerLaunch implements Callable<Integer> {
       List<String> userLocalDirs = getUserLocalDirs(localDirs);
       List<String> containerLocalDirs = getContainerLocalDirs(localDirs);
       List<String> containerLogDirs = getContainerLogDirs(logDirs);
+      List<String> userFilecacheDirs = getUserFilecacheDirs(localDirs);
+      List<String> applicationLocalDirs = getApplicationLocalDirs(localDirs,
+          appIdStr);
 
       if (!dirsHandler.areDisksHealthy()) {
         ret = ContainerExitStatus.DISKS_FAILED;
@@ -295,7 +304,9 @@ public class ContainerLaunch implements Callable<Integer> {
           .setFilecacheDirs(filecacheDirs)
           .setUserLocalDirs(userLocalDirs)
           .setContainerLocalDirs(containerLocalDirs)
-          .setContainerLogDirs(containerLogDirs).build());
+          .setContainerLogDirs(containerLogDirs)
+          .setUserFilecacheDirs(userFilecacheDirs)
+          .setApplicationLocalDirs(applicationLocalDirs).build());
     } catch (ConfigurationException e) {
       LOG.error("Failed to launch container due to configuration error.", e);
       dispatcher.getEventHandler().handle(new ContainerExitEvent(
@@ -426,6 +437,31 @@ public class ContainerLaunch implements Callable<Integer> {
     return filecacheDirs;
   }
 
+  protected List<String> getUserFilecacheDirs(List<String> localDirs) {
+    List<String> userFilecacheDirs = new ArrayList<>(localDirs.size());
+    String user = container.getUser();
+    for (String localDir : localDirs) {
+      String userFilecacheDir = localDir + Path.SEPARATOR +
+          ContainerLocalizer.USERCACHE + Path.SEPARATOR + user
+          + Path.SEPARATOR + ContainerLocalizer.FILECACHE;
+      userFilecacheDirs.add(userFilecacheDir);
+    }
+    return userFilecacheDirs;
+  }
+
+  protected List<String> getApplicationLocalDirs(List<String> localDirs,
+      String appIdStr) {
+    List<String> applicationLocalDirs = new ArrayList<>(localDirs.size());
+    String user = container.getUser();
+    for (String localDir : localDirs) {
+      String appLocalDir = localDir + Path.SEPARATOR +
+          ContainerLocalizer.USERCACHE + Path.SEPARATOR + user
+          + Path.SEPARATOR + ContainerLocalizer.APPCACHE
+          + Path.SEPARATOR + appIdStr;
+      applicationLocalDirs.add(appLocalDir);
+    }
+    return applicationLocalDirs;
+  }
 
   protected Map<Path, List<String>> getLocalizedResources()
       throws YarnException {

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

@@ -98,6 +98,9 @@ public class ContainerRelaunch extends ContainerLaunch {
       List<String> containerLogDirs = getContainerLogDirs(logDirs);
       List<String> filecacheDirs = getNMFilecacheDirs(localDirs);
       List<String> userLocalDirs = getUserLocalDirs(localDirs);
+      List<String> userFilecacheDirs = getUserFilecacheDirs(localDirs);
+      List<String> applicationLocalDirs = getApplicationLocalDirs(localDirs,
+          appIdStr);
 
       if (!dirsHandler.areDisksHealthy()) {
         ret = ContainerExitStatus.DISKS_FAILED;
@@ -119,6 +122,8 @@ public class ContainerRelaunch extends ContainerLaunch {
           .setUserLocalDirs(userLocalDirs)
           .setContainerLocalDirs(containerLocalDirs)
           .setContainerLogDirs(containerLogDirs)
+          .setUserFilecacheDirs(userFilecacheDirs)
+          .setApplicationLocalDirs(applicationLocalDirs)
           .build());
     } catch (ConfigurationException e) {
       LOG.error("Failed to launch container due to configuration error.", 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/DockerLinuxContainerRuntime.java

@@ -769,16 +769,17 @@ public class DockerLinuxContainerRuntime implements LinuxContainerRuntime {
     @SuppressWarnings("unchecked")
     List<String> filecacheDirs = ctx.getExecutionAttribute(FILECACHE_DIRS);
     @SuppressWarnings("unchecked")
-    List<String> containerLocalDirs = ctx.getExecutionAttribute(
-        CONTAINER_LOCAL_DIRS);
-    @SuppressWarnings("unchecked")
     List<String> containerLogDirs = ctx.getExecutionAttribute(
         CONTAINER_LOG_DIRS);
     @SuppressWarnings("unchecked")
+    List<String> userFilecacheDirs =
+        ctx.getExecutionAttribute(USER_FILECACHE_DIRS);
+    @SuppressWarnings("unchecked")
+    List<String> applicationLocalDirs =
+        ctx.getExecutionAttribute(APPLICATION_LOCAL_DIRS);
+    @SuppressWarnings("unchecked")
     Map<Path, List<String>> localizedResources = ctx.getExecutionAttribute(
         LOCALIZED_RESOURCES);
-    @SuppressWarnings("unchecked")
-    List<String> userLocalDirs = ctx.getExecutionAttribute(USER_LOCAL_DIRS);
 
     @SuppressWarnings("unchecked")
     DockerRunCommand runCommand = new DockerRunCommand(containerIdStr,
@@ -789,14 +790,10 @@ public class DockerLinuxContainerRuntime implements LinuxContainerRuntime {
     setHostname(runCommand, containerIdStr, hostname);
     runCommand.setCapabilities(capabilities);
 
-    List<String> allDirs = new ArrayList<>(containerLocalDirs);
+    runCommand.addAllReadWriteMountLocations(containerLogDirs);
-    allDirs.addAll(filecacheDirs);
+    runCommand.addAllReadWriteMountLocations(applicationLocalDirs);
-    allDirs.add(containerWorkDir.toString());
+    runCommand.addAllReadOnlyMountLocations(filecacheDirs);
-    allDirs.addAll(containerLogDirs);
+    runCommand.addAllReadOnlyMountLocations(userFilecacheDirs);
-    allDirs.addAll(userLocalDirs);
-    for (String dir: allDirs) {
-      runCommand.addMountLocation(dir, dir, true);
-    }
 
     if (environment.containsKey(ENV_DOCKER_CONTAINER_LOCAL_RESOURCE_MOUNTS)) {
       String mounts = environment.get(

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

@@ -70,6 +70,10 @@ public final class LinuxContainerRuntimeConstants {
       List.class, "user_local_dirs");
   public static final Attribute<List> CONTAINER_LOCAL_DIRS = Attribute
       .attribute(List.class, "container_local_dirs");
+  public static final Attribute<List> USER_FILECACHE_DIRS = Attribute
+      .attribute(List.class, "user_filecache_dirs");
+  public static final Attribute<List> APPLICATION_LOCAL_DIRS = Attribute
+      .attribute(List.class, "application_local_dirs");
   public static final Attribute<List> CONTAINER_LOG_DIRS = Attribute.attribute(
       List.class, "container_log_dirs");
   public static final Attribute<String> RESOURCES_OPTIONS = Attribute.attribute(

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/DockerRunCommand.java

@@ -77,6 +77,13 @@ public class DockerRunCommand extends DockerCommand {
     return this;
   }
 
+  public DockerRunCommand addAllReadWriteMountLocations(List<String> paths) {
+    for (String dir: paths) {
+      this.addReadWriteMountLocation(dir, dir);
+    }
+    return this;
+  }
+
   public DockerRunCommand addReadOnlyMountLocation(String sourcePath, String
       destinationPath, boolean createSource) {
     boolean sourceExists = new File(sourcePath).exists();
@@ -93,6 +100,13 @@ public class DockerRunCommand extends DockerCommand {
     return this;
   }
 
+  public DockerRunCommand addAllReadOnlyMountLocations(List<String> paths) {
+    for (String dir: paths) {
+      this.addReadOnlyMountLocation(dir, dir);
+    }
+    return this;
+  }
+
   public DockerRunCommand setVolumeDriver(String volumeDriver) {
     super.addCommandArguments("volume-driver", volumeDriver);
     return this;

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

@@ -49,6 +49,8 @@ public final class ContainerStartContext {
   private final List<String> userLocalDirs;
   private final List<String> containerLocalDirs;
   private final List<String> containerLogDirs;
+  private final List<String> userFilecacheDirs;
+  private final List<String> applicationLocalDirs;
 
   public static final class Builder {
     private Container container;
@@ -64,6 +66,8 @@ public final class ContainerStartContext {
     private List<String> userLocalDirs;
     private List<String> containerLocalDirs;
     private List<String> containerLogDirs;
+    private List<String> userFilecacheDirs;
+    private List<String> applicationLocalDirs;
 
     public Builder() {
     }
@@ -135,6 +139,18 @@ public final class ContainerStartContext {
       return this;
     }
 
+    @SuppressWarnings("checkstyle:hiddenfield")
+    public Builder setUserFilecacheDirs(List<String> userFilecacheDirs) {
+      this.userFilecacheDirs = userFilecacheDirs;
+      return this;
+    }
+
+    @SuppressWarnings("checkstyle:hiddenfield")
+    public Builder setApplicationLocalDirs(List<String> applicationLocalDirs) {
+      this.applicationLocalDirs = applicationLocalDirs;
+      return this;
+    }
+
     public ContainerStartContext build() {
       return new ContainerStartContext(this);
     }
@@ -154,6 +170,8 @@ public final class ContainerStartContext {
     this.userLocalDirs = builder.userLocalDirs;
     this.containerLocalDirs = builder.containerLocalDirs;
     this.containerLogDirs = builder.containerLogDirs;
+    this.userFilecacheDirs = builder.userFilecacheDirs;
+    this.applicationLocalDirs = builder.applicationLocalDirs;
   }
 
   public Container getContainer() {
@@ -212,4 +230,12 @@ public final class ContainerStartContext {
     return Collections.unmodifiableList(this
         .containerLogDirs);
   }
+
+  public List<String> getUserFilecacheDirs() {
+    return Collections.unmodifiableList(this.userFilecacheDirs);
+  }
+
+  public List<String> getApplicationLocalDirs() {
+    return Collections.unmodifiableList(this.applicationLocalDirs);
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestLinuxContainerExecutorWithMocks.java

@@ -209,6 +209,8 @@ public class TestLinuxContainerExecutorWithMocks {
         .setUserLocalDirs(new ArrayList<>())
         .setContainerLocalDirs(new ArrayList<>())
         .setContainerLogDirs(new ArrayList<>())
+        .setUserFilecacheDirs(new ArrayList<>())
+        .setApplicationLocalDirs(new ArrayList<>())
         .build());
     assertEquals(0, ret);
     assertEquals(Arrays.asList(YarnConfiguration.DEFAULT_NM_NONSECURE_MODE_LOCAL_USER,
@@ -398,6 +400,8 @@ public class TestLinuxContainerExecutorWithMocks {
             .setUserLocalDirs(new ArrayList<>())
             .setContainerLocalDirs(new ArrayList<>())
             .setContainerLogDirs(new ArrayList<>())
+            .setUserFilecacheDirs(new ArrayList<>())
+            .setApplicationLocalDirs(new ArrayList<>())
             .build());
 
         Assert.assertNotSame(0, ret);
@@ -611,6 +615,8 @@ public class TestLinuxContainerExecutorWithMocks {
         .setUserLocalDirs(new ArrayList<>())
         .setContainerLocalDirs(new ArrayList<>())
         .setContainerLogDirs(new ArrayList<>())
+        .setUserFilecacheDirs(new ArrayList<>())
+        .setApplicationLocalDirs(new ArrayList<>())
         .build());
     lce.deleteAsUser(new DeletionAsUserContext.Builder()
         .setUser(appSubmitter)
@@ -665,6 +671,8 @@ public class TestLinuxContainerExecutorWithMocks {
         .setUserLocalDirs(new ArrayList<>())
         .setContainerLocalDirs(new ArrayList<>())
         .setContainerLogDirs(new ArrayList<>())
+        .setUserFilecacheDirs(new ArrayList<>())
+        .setApplicationLocalDirs(new ArrayList<>())
         .build());
     ArgumentCaptor<PrivilegedOperation> opCaptor = ArgumentCaptor.forClass(
         PrivilegedOperation.class);

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/launcher/TestContainerRelaunch.java

@@ -93,5 +93,7 @@ public class TestContainerRelaunch {
     assertNotNull("tokens path null", csc.getNmPrivateTokensPath());
     assertNotNull("user null", csc.getUser());
     assertNotNull("user local dirs null", csc.getUserLocalDirs());
+    assertNotNull("user filecache dirs null", csc.getUserFilecacheDirs());
+    assertNotNull("application local dirs null", csc.getApplicationLocalDirs());
   }
 }

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

@@ -24,7 +24,6 @@ import org.apache.commons.io.IOUtils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.registry.client.binding.RegistryPathUtils;
 import org.apache.hadoop.util.Shell;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.yarn.api.records.ContainerId;
@@ -64,19 +63,16 @@ import java.nio.charset.Charset;
 import java.nio.file.Files;
 import java.nio.file.Paths;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashMap;
-import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Random;
-import java.util.Set;
 
 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;
 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.LinuxContainerRuntimeConstants.CONTAINER_LOCAL_DIRS;
 import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.LinuxContainerRuntimeConstants.CONTAINER_LOG_DIRS;
 import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.LinuxContainerRuntimeConstants.CONTAINER_WORK_DIR;
 import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.LinuxContainerRuntimeConstants.FILECACHE_DIRS;
@@ -91,7 +87,7 @@ import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.r
 import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.LinuxContainerRuntimeConstants.RUN_AS_USER;
 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_LOCAL_DIRS;
+import static org.apache.hadoop.yarn.server.nodemanager.containermanager.linux.runtime.LinuxContainerRuntimeConstants.USER_FILECACHE_DIRS;
 import static org.mockito.Mockito.any;
 import static org.mockito.Mockito.anyBoolean;
 import static org.mockito.Mockito.anyList;
@@ -109,7 +105,6 @@ public class TestDockerContainerRuntime {
   private PrivilegedOperationExecutor mockExecutor;
   private CGroupsHandler mockCGroupsHandler;
   private String containerId;
-  private String defaultHostname;
   private Container container;
   private ContainerId cId;
   private ContainerLaunchContext context;
@@ -128,8 +123,8 @@ public class TestDockerContainerRuntime {
   private List<String> localDirs;
   private List<String> logDirs;
   private List<String> filecacheDirs;
-  private List<String> userLocalDirs;
+  private List<String> userFilecacheDirs;
-  private List<String> containerLocalDirs;
+  private List<String> applicationLocalDirs;
   private List<String> containerLogDirs;
   private Map<Path, List<String>> localizedResources;
   private String resourcesOptions;
@@ -151,7 +146,6 @@ public class TestDockerContainerRuntime {
         .mock(PrivilegedOperationExecutor.class);
     mockCGroupsHandler = Mockito.mock(CGroupsHandler.class);
     containerId = "container_id";
-    defaultHostname = RegistryPathUtils.encodeYarnID(containerId);
     container = mock(Container.class);
     cId = mock(ContainerId.class);
     context = mock(ContainerLaunchContext.class);
@@ -208,16 +202,16 @@ public class TestDockerContainerRuntime {
     logDirs = new ArrayList<>();
     filecacheDirs = new ArrayList<>();
     resourcesOptions = "cgroups=none";
-    userLocalDirs = new ArrayList<>();
+    userFilecacheDirs = new ArrayList<>();
-    containerLocalDirs = new ArrayList<>();
+    applicationLocalDirs = new ArrayList<>();
     containerLogDirs = new ArrayList<>();
     localizedResources = new HashMap<>();
 
     localDirs.add("/test_local_dir");
     logDirs.add("/test_log_dir");
     filecacheDirs.add("/test_filecache_dir");
-    userLocalDirs.add("/test_user_local_dir");
+    userFilecacheDirs.add("/test_user_filecache_dir");
-    containerLocalDirs.add("/test_container_local_dir");
+    applicationLocalDirs.add("/test_application_local_dir");
     containerLogDirs.add("/test_container_log_dir");
     localizedResources.put(new Path("/test_local_dir/test_resource_file"),
         Collections.singletonList("test_dir/test_resource_file"));
@@ -241,8 +235,8 @@ public class TestDockerContainerRuntime {
         .setExecutionAttribute(LOCAL_DIRS, localDirs)
         .setExecutionAttribute(LOG_DIRS, logDirs)
         .setExecutionAttribute(FILECACHE_DIRS, filecacheDirs)
-        .setExecutionAttribute(USER_LOCAL_DIRS, userLocalDirs)
+        .setExecutionAttribute(USER_FILECACHE_DIRS, userFilecacheDirs)
-        .setExecutionAttribute(CONTAINER_LOCAL_DIRS, containerLocalDirs)
+        .setExecutionAttribute(APPLICATION_LOCAL_DIRS, applicationLocalDirs)
         .setExecutionAttribute(CONTAINER_LOG_DIRS, containerLogDirs)
         .setExecutionAttribute(LOCALIZED_RESOURCES, localizedResources)
         .setExecutionAttribute(RESOURCES_OPTIONS, resourcesOptions);
@@ -296,41 +290,28 @@ public class TestDockerContainerRuntime {
     List<String> args = op.getArguments();
 
     //This invocation of container-executor should use 13 arguments in a
-    // specific order (sigh.)
+    // specific order
-    Assert.assertEquals(13, args.size());
+    int expected = 13;
-
+    int counter = 1;
-    //verify arguments
+    Assert.assertEquals(expected, args.size());
-    Assert.assertEquals(user, args.get(1));
+    Assert.assertEquals(user, args.get(counter++));
     Assert.assertEquals(Integer.toString(PrivilegedOperation.RunAsUserCommand
-        .LAUNCH_DOCKER_CONTAINER.getValue()), args.get(2));
+        .LAUNCH_DOCKER_CONTAINER.getValue()), args.get(counter++));
-    Assert.assertEquals(appId, args.get(3));
+    Assert.assertEquals(appId, args.get(counter++));
-    Assert.assertEquals(containerId, args.get(4));
+    Assert.assertEquals(containerId, args.get(counter++));
-    Assert.assertEquals(containerWorkDir.toString(), args.get(5));
+    Assert.assertEquals(containerWorkDir.toString(), args.get(counter++));
     Assert.assertEquals(nmPrivateContainerScriptPath.toUri()
-        .toString(), args.get(6));
+        .toString(), args.get(counter++));
-    Assert.assertEquals(nmPrivateTokensPath.toUri().getPath(), args.get(7));
+    Assert.assertEquals(nmPrivateTokensPath.toUri().getPath(),
-    Assert.assertEquals(pidFilePath.toString(), args.get(8));
+        args.get(counter++));
-    Assert.assertEquals(localDirs.get(0), args.get(9));
+    Assert.assertEquals(pidFilePath.toString(), args.get(counter++));
-    Assert.assertEquals(logDirs.get(0), args.get(10));
+    Assert.assertEquals(localDirs.get(0), args.get(counter++));
-    Assert.assertEquals(resourcesOptions, args.get(12));
+    Assert.assertEquals(logDirs.get(0), args.get(counter++));
+    Assert.assertEquals(resourcesOptions, args.get(++counter));
 
     return op;
   }
 
-  private String getExpectedTestCapabilitiesArgumentString()  {
-    /* Ordering of capabilities depends on HashSet ordering. */
-    Set<String> capabilitySet = new HashSet<>(Arrays.asList(testCapabilities));
-    StringBuilder expectedCapabilitiesString = new StringBuilder(
-        "--cap-drop=ALL ");
-
-    for(String capability : capabilitySet) {
-      expectedCapabilitiesString.append("--cap-add=").append(capability)
-          .append(" ");
-    }
-
-    return expectedCapabilitiesString.toString();
-  }
-
   @Test
   public void testDockerContainerLaunch()
       throws ContainerExecutionException, PrivilegedOperationException,
@@ -347,7 +328,7 @@ public class TestDockerContainerRuntime {
     List<String> dockerCommands = Files.readAllLines(Paths.get
             (dockerCommandFile), Charset.forName("UTF-8"));
 
-    int expected = 14;
+    int expected = 15;
     int counter = 0;
     Assert.assertEquals(expected, dockerCommands.size());
     Assert.assertEquals("[docker-command-execution]",
@@ -367,16 +348,16 @@ public class TestDockerContainerRuntime {
         dockerCommands.get(counter++));
     Assert.assertEquals("  name=container_id", 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",
+        dockerCommands.get(counter++));
     Assert.assertEquals(
-        "  rw-mounts=/test_container_local_dir:/test_container_local_dir,"
+        "  rw-mounts=/test_container_log_dir:/test_container_log_dir,"
-            + "/test_filecache_dir:/test_filecache_dir,"
+            + "/test_application_local_dir:/test_application_local_dir",
-            + "/test_container_work_dir:/test_container_work_dir,"
-            + "/test_container_log_dir:/test_container_log_dir,"
-            + "/test_user_local_dir:/test_user_local_dir",
         dockerCommands.get(counter++));
     Assert.assertEquals("  user=" + uidGidPair, dockerCommands.get(counter++));
     Assert.assertEquals("  workdir=/test_container_work_dir",
-        dockerCommands.get(counter++));
+        dockerCommands.get(counter));
   }
 
   @Test
@@ -397,7 +378,7 @@ public class TestDockerContainerRuntime {
     List<String> dockerCommands = Files.readAllLines(
         Paths.get(dockerCommandFile), Charset.forName("UTF-8"));
 
-    Assert.assertEquals(14, dockerCommands.size());
+    Assert.assertEquals(15, dockerCommands.size());
     int counter = 0;
     Assert.assertEquals("[docker-command-execution]",
         dockerCommands.get(counter++));
@@ -418,16 +399,16 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals("  name=container_id", 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",
+        dockerCommands.get(counter++));
     Assert.assertEquals(
-        "  rw-mounts=/test_container_local_dir:/test_container_local_dir,"
+        "  rw-mounts=/test_container_log_dir:/test_container_log_dir,"
-            + "/test_filecache_dir:/test_filecache_dir,"
+            + "/test_application_local_dir:/test_application_local_dir",
-            + "/test_container_work_dir:/test_container_work_dir,"
-            + "/test_container_log_dir:/test_container_log_dir,"
-            + "/test_user_local_dir:/test_user_local_dir",
         dockerCommands.get(counter++));
     Assert.assertEquals("  user=" + uidGidPair, dockerCommands.get(counter++));
     Assert.assertEquals("  workdir=/test_container_work_dir",
-        dockerCommands.get(counter++));
+        dockerCommands.get(counter));
   }
 
   @Test
@@ -515,7 +496,7 @@ public class TestDockerContainerRuntime {
     //This is the expected docker invocation for this case
     List<String> dockerCommands = Files
         .readAllLines(Paths.get(dockerCommandFile), Charset.forName("UTF-8"));
-    int expected = 14;
+    int expected = 15;
     int counter = 0;
     Assert.assertEquals(expected, dockerCommands.size());
     Assert.assertEquals("[docker-command-execution]",
@@ -537,16 +518,16 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
     Assert
         .assertEquals("  net=" + allowedNetwork, dockerCommands.get(counter++));
+    Assert.assertEquals("  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
+            + "/test_user_filecache_dir:/test_user_filecache_dir",
+        dockerCommands.get(counter++));
     Assert.assertEquals(
-        "  rw-mounts=/test_container_local_dir:/test_container_local_dir,"
+        "  rw-mounts=/test_container_log_dir:/test_container_log_dir,"
-            + "/test_filecache_dir:/test_filecache_dir,"
+            + "/test_application_local_dir:/test_application_local_dir",
-            + "/test_container_work_dir:/test_container_work_dir,"
-            + "/test_container_log_dir:/test_container_log_dir,"
-            + "/test_user_local_dir:/test_user_local_dir",
         dockerCommands.get(counter++));
     Assert.assertEquals("  user=" + uidGidPair, dockerCommands.get(counter++));
     Assert.assertEquals("  workdir=/test_container_work_dir",
-        dockerCommands.get(counter++));
+        dockerCommands.get(counter));
   }
 
   @Test
@@ -583,7 +564,7 @@ public class TestDockerContainerRuntime {
     List<String> dockerCommands = Files
         .readAllLines(Paths.get(dockerCommandFile), Charset.forName("UTF-8"));
 
-    int expected = 14;
+    int expected = 15;
     int counter = 0;
     Assert.assertEquals(expected, dockerCommands.size());
     Assert.assertEquals("[docker-command-execution]",
@@ -603,16 +584,16 @@ public class TestDockerContainerRuntime {
         dockerCommands.get(counter++));
     Assert.assertEquals("  name=container_id", 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",
+        dockerCommands.get(counter++));
     Assert.assertEquals(
-        "  rw-mounts=/test_container_local_dir:/test_container_local_dir,"
+        "  rw-mounts=/test_container_log_dir:/test_container_log_dir,"
-            + "/test_filecache_dir:/test_filecache_dir,"
+            + "/test_application_local_dir:/test_application_local_dir",
-            + "/test_container_work_dir:/test_container_work_dir,"
-            + "/test_container_log_dir:/test_container_log_dir,"
-            + "/test_user_local_dir:/test_user_local_dir",
         dockerCommands.get(counter++));
     Assert.assertEquals("  user=" + uidGidPair, dockerCommands.get(counter++));
     Assert.assertEquals("  workdir=/test_container_work_dir",
-        dockerCommands.get(counter++));
+        dockerCommands.get(counter));
 
     //now set an explicit (non-default) allowedNetwork and ensure that it is
     // used.
@@ -649,16 +630,16 @@ public class TestDockerContainerRuntime {
 
     Assert.assertEquals("  name=container_id", 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",
+        dockerCommands.get(counter++));
     Assert.assertEquals(
-        "  rw-mounts=/test_container_local_dir:/test_container_local_dir,"
+        "  rw-mounts=/test_container_log_dir:/test_container_log_dir,"
-            + "/test_filecache_dir:/test_filecache_dir,"
+            + "/test_application_local_dir:/test_application_local_dir",
-            + "/test_container_work_dir:/test_container_work_dir,"
-            + "/test_container_log_dir:/test_container_log_dir,"
-            + "/test_user_local_dir:/test_user_local_dir",
         dockerCommands.get(counter++));
     Assert.assertEquals("  user=" + uidGidPair, dockerCommands.get(counter++));
     Assert.assertEquals("  workdir=/test_container_work_dir",
-        dockerCommands.get(counter++));
+        dockerCommands.get(counter));
 
 
     //disallowed network should trigger a launch failure
@@ -693,7 +674,7 @@ public class TestDockerContainerRuntime {
     List<String> dockerCommands = Files.readAllLines(Paths.get
         (dockerCommandFile), Charset.forName("UTF-8"));
 
-    int expected = 14;
+    int expected = 15;
     Assert.assertEquals(expected, dockerCommands.size());
 
     String command = dockerCommands.get(0);
@@ -744,7 +725,7 @@ public class TestDockerContainerRuntime {
     List<String> dockerCommands = Files.readAllLines(
         Paths.get(dockerCommandFile), Charset.forName("UTF-8"));
 
-    int expected = 15;
+    int expected = 16;
     int counter = 0;
     Assert.assertEquals(expected, dockerCommands.size());
     Assert.assertEquals("[docker-command-execution]",
@@ -765,16 +746,16 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals("  name=container_id", 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,"
+            + "/test_user_filecache_dir:/test_user_filecache_dir",
+        dockerCommands.get(counter++));
     Assert.assertEquals(
-        "  rw-mounts=/test_container_local_dir:/test_container_local_dir,"
+        "  rw-mounts=/test_container_log_dir:/test_container_log_dir,"
-            + "/test_filecache_dir:/test_filecache_dir,"
+            + "/test_application_local_dir:/test_application_local_dir",
-            + "/test_container_work_dir:/test_container_work_dir,"
-            + "/test_container_log_dir:/test_container_log_dir,"
-            + "/test_user_local_dir:/test_user_local_dir",
         dockerCommands.get(counter++));
     Assert.assertEquals("  user=" + uidGidPair, dockerCommands.get(counter++));
     Assert.assertEquals("  workdir=/test_container_work_dir",
-        dockerCommands.get(counter++));
+        dockerCommands.get(counter));
   }
 
   @Test
@@ -796,7 +777,7 @@ public class TestDockerContainerRuntime {
     List<String> dockerCommands = Files.readAllLines(
         Paths.get(dockerCommandFile), Charset.forName("UTF-8"));
 
-    int expected = 14;
+    int expected = 15;
     Assert.assertEquals(expected, dockerCommands.size());
 
     String command = dockerCommands.get(0);
@@ -808,8 +789,7 @@ public class TestDockerContainerRuntime {
 
   @Test
   public void testLaunchPrivilegedContainersWithDisabledSetting()
-      throws ContainerExecutionException, PrivilegedOperationException,
+      throws ContainerExecutionException {
-      IOException{
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
     runtime.initialize(conf, null);
@@ -827,8 +807,7 @@ public class TestDockerContainerRuntime {
 
   @Test
   public void testLaunchPrivilegedContainersWithEnabledSettingAndDefaultACL()
-      throws ContainerExecutionException, PrivilegedOperationException,
+      throws ContainerExecutionException {
-      IOException{
     //Enable privileged containers.
     conf.setBoolean(YarnConfiguration.NM_DOCKER_ALLOW_PRIVILEGED_CONTAINERS,
         true);
@@ -854,8 +833,7 @@ public class TestDockerContainerRuntime {
   @Test
   public void
   testLaunchPrivilegedContainersEnabledAndUserNotInWhitelist()
-      throws ContainerExecutionException, PrivilegedOperationException,
+      throws ContainerExecutionException {
-      IOException{
     //Enable privileged containers.
     conf.setBoolean(YarnConfiguration.NM_DOCKER_ALLOW_PRIVILEGED_CONTAINERS,
         true);
@@ -905,7 +883,7 @@ public class TestDockerContainerRuntime {
     List<String> dockerCommands = Files.readAllLines(Paths.get
         (dockerCommandFile), Charset.forName("UTF-8"));
 
-    int expected = 15;
+    int expected = 16;
     int counter = 0;
     Assert.assertEquals(expected, dockerCommands.size());
     Assert.assertEquals("[docker-command-execution]",
@@ -926,16 +904,16 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals("  name=container_id", 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,"
+            + "/test_user_filecache_dir:/test_user_filecache_dir",
+        dockerCommands.get(counter++));
     Assert.assertEquals(
-        "  rw-mounts=/test_container_local_dir:/test_container_local_dir,"
+        "  rw-mounts=/test_container_log_dir:/test_container_log_dir,"
-            + "/test_filecache_dir:/test_filecache_dir,"
+            + "/test_application_local_dir:/test_application_local_dir",
-            + "/test_container_work_dir:/test_container_work_dir,"
-            + "/test_container_log_dir:/test_container_log_dir,"
-            + "/test_user_local_dir:/test_user_local_dir",
         dockerCommands.get(counter++));
     Assert.assertEquals("  user=" + uidGidPair, dockerCommands.get(counter++));
     Assert.assertEquals("  workdir=/test_container_work_dir",
-        dockerCommands.get(counter++));
+        dockerCommands.get(counter));
   }
 
   @Test
@@ -985,9 +963,7 @@ public class TestDockerContainerRuntime {
   }
 
   @Test
-  public void testMountSourceOnly()
+  public void testMountSourceOnly() throws ContainerExecutionException {
-      throws ContainerExecutionException, PrivilegedOperationException,
-      IOException{
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
     runtime.initialize(conf, null);
@@ -1045,24 +1021,21 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
     Assert.assertEquals("  net=host", dockerCommands.get(counter++));
     Assert.assertEquals(
-        "  ro-mounts=/test_local_dir/test_resource_file:test_mount",
+        "  ro-mounts=/test_filecache_dir:/test_filecache_dir,/"
+            + "test_user_filecache_dir:/test_user_filecache_dir,"
+            + "/test_local_dir/test_resource_file:test_mount",
         dockerCommands.get(counter++));
     Assert.assertEquals(
-        "  rw-mounts=/test_container_local_dir:/test_container_local_dir,"
+        "  rw-mounts=/test_container_log_dir:/test_container_log_dir,"
-            + "/test_filecache_dir:/test_filecache_dir,"
+            + "/test_application_local_dir:/test_application_local_dir",
-            + "/test_container_work_dir:/test_container_work_dir,"
-            + "/test_container_log_dir:/test_container_log_dir,"
-            + "/test_user_local_dir:/test_user_local_dir",
         dockerCommands.get(counter++));
     Assert.assertEquals("  user=" + uidGidPair, dockerCommands.get(counter++));
     Assert.assertEquals("  workdir=/test_container_work_dir",
-        dockerCommands.get(counter++));
+        dockerCommands.get(counter));
   }
 
   @Test
-  public void testMountInvalid()
+  public void testMountInvalid() throws ContainerExecutionException {
-      throws ContainerExecutionException, PrivilegedOperationException,
-      IOException{
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
     runtime.initialize(conf, null);
@@ -1121,20 +1094,18 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
     Assert.assertEquals("  net=host", dockerCommands.get(counter++));
     Assert.assertEquals(
-        "  ro-mounts=/test_local_dir/test_resource_file:test_mount1,"
+        "  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
+            + "/test_user_filecache_dir:/test_user_filecache_dir,"
+            + "/test_local_dir/test_resource_file:test_mount1,"
             + "/test_local_dir/test_resource_file:test_mount2",
         dockerCommands.get(counter++));
     Assert.assertEquals(
-        "  rw-mounts=/test_container_local_dir:/test_container_local_dir,"
+        "  rw-mounts=/test_container_log_dir:/test_container_log_dir,"
-            + "/test_filecache_dir:/test_filecache_dir,"
+            + "/test_application_local_dir:/test_application_local_dir",
-            + "/test_container_work_dir:/test_container_work_dir,"
-            + "/test_container_log_dir:/test_container_log_dir,"
-            + "/test_user_local_dir:/test_user_local_dir",
         dockerCommands.get(counter++));
     Assert.assertEquals("  user=" + uidGidPair, dockerCommands.get(counter++));
     Assert.assertEquals("  workdir=/test_container_work_dir",
-        dockerCommands.get(counter++));
+        dockerCommands.get(counter));
-
   }
 
   @Test
@@ -1177,25 +1148,22 @@ public class TestDockerContainerRuntime {
         dockerCommands.get(counter++));
     Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
     Assert.assertEquals("  net=host", dockerCommands.get(counter++));
-    Assert.assertEquals("  ro-mounts=/tmp/foo:/tmp/foo",
+    Assert.assertEquals("  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
+            + "/test_user_filecache_dir:/test_user_filecache_dir,"
+            + "/tmp/foo:/tmp/foo",
         dockerCommands.get(counter++));
     Assert.assertEquals(
-        "  rw-mounts=/test_container_local_dir:/test_container_local_dir,"
+        "  rw-mounts=/test_container_log_dir:/test_container_log_dir,"
-            + "/test_filecache_dir:/test_filecache_dir,"
+            + "/test_application_local_dir:/test_application_local_dir,"
-            + "/test_container_work_dir:/test_container_work_dir,"
-            + "/test_container_log_dir:/test_container_log_dir,"
-            + "/test_user_local_dir:/test_user_local_dir,"
             + "/tmp/bar:/tmp/bar",
         dockerCommands.get(counter++));
     Assert.assertEquals("  user=" + uidGidPair, dockerCommands.get(counter++));
     Assert.assertEquals("  workdir=/test_container_work_dir",
-        dockerCommands.get(counter++));
+        dockerCommands.get(counter));
   }
 
   @Test
-  public void testUserMountInvalid()
+  public void testUserMountInvalid() throws ContainerExecutionException {
-      throws ContainerExecutionException, PrivilegedOperationException,
-      IOException{
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
     runtime.initialize(conf, null);
@@ -1213,9 +1181,7 @@ public class TestDockerContainerRuntime {
   }
 
   @Test
-  public void testUserMountModeInvalid()
+  public void testUserMountModeInvalid() throws ContainerExecutionException {
-      throws ContainerExecutionException, PrivilegedOperationException,
-      IOException{
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
     runtime.initialize(conf, null);
@@ -1233,9 +1199,7 @@ public class TestDockerContainerRuntime {
   }
 
   @Test
-  public void testUserMountModeNulInvalid()
+  public void testUserMountModeNulInvalid() throws ContainerExecutionException {
-      throws ContainerExecutionException, PrivilegedOperationException,
-      IOException{
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
     runtime.initialize(conf, null);
@@ -1679,14 +1643,13 @@ public class TestDockerContainerRuntime {
         dockerCommands.get(counter++));
     Assert.assertEquals("  name=container_id", dockerCommands.get(counter++));
     Assert.assertEquals("  net=host", dockerCommands.get(counter++));
-    Assert.assertEquals("  ro-mounts=/source/path:/destination/path",
+    Assert.assertEquals("  ro-mounts=/test_filecache_dir:/test_filecache_dir,"
+            + "/test_user_filecache_dir:/test_user_filecache_dir,"
+            + "/source/path:/destination/path",
         dockerCommands.get(counter++));
     Assert.assertEquals(
-        "  rw-mounts=/test_container_local_dir:/test_container_local_dir,"
+        "  rw-mounts=/test_container_log_dir:/test_container_log_dir,"
-            + "/test_filecache_dir:/test_filecache_dir,"
+            + "/test_application_local_dir:/test_application_local_dir",
-            + "/test_container_work_dir:/test_container_work_dir,"
-            + "/test_container_log_dir:/test_container_log_dir,"
-            + "/test_user_local_dir:/test_user_local_dir",
         dockerCommands.get(counter++));
     Assert.assertEquals("  user=" + uidGidPair, dockerCommands.get(counter++));
 
@@ -1694,13 +1657,11 @@ public class TestDockerContainerRuntime {
     Assert.assertEquals("  volume-driver=driver-1",
         dockerCommands.get(counter++));
     Assert.assertEquals("  workdir=/test_container_work_dir",
-        dockerCommands.get(counter++));
+        dockerCommands.get(counter));
   }
 
   @Test
-  public void testDockerCapabilities()
+  public void testDockerCapabilities() throws ContainerExecutionException {
-      throws ContainerExecutionException, PrivilegedOperationException,
-      IOException {
     DockerLinuxContainerRuntime runtime = new DockerLinuxContainerRuntime(
         mockExecutor, mockCGroupsHandler);
     try {