diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/api/records/ConfigFile.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/api/records/ConfigFile.java
index c09373f5483..060e2045278 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/api/records/ConfigFile.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/api/records/ConfigFile.java
@@ -24,6 +24,7 @@ import io.swagger.annotations.ApiModel;
 import io.swagger.annotations.ApiModelProperty;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
 
 import javax.xml.bind.annotation.XmlElement;
 import javax.xml.bind.annotation.XmlEnum;
@@ -73,6 +74,7 @@ public class ConfigFile implements Serializable {
   private TypeEnum type = null;
   private String destFile = null;
   private String srcFile = null;
+  private LocalResourceVisibility visibility = null;
   private Map<String, String> properties = new HashMap<>();
 
   public ConfigFile copy() {
@@ -80,6 +82,7 @@ public class ConfigFile implements Serializable {
     copy.setType(this.getType());
     copy.setSrcFile(this.getSrcFile());
     copy.setDestFile(this.getDestFile());
+    copy.setVisibility(this.visibility);
     if (this.getProperties() != null && !this.getProperties().isEmpty()) {
       copy.getProperties().putAll(this.getProperties());
     }
@@ -150,6 +153,26 @@ public class ConfigFile implements Serializable {
     this.srcFile = srcFile;
   }
 
+
+  /**
+   * Visibility of the Config file.
+   **/
+  public ConfigFile visibility(LocalResourceVisibility localrsrcVisibility) {
+    this.visibility = localrsrcVisibility;
+    return this;
+  }
+
+  @ApiModelProperty(example = "null", value = "Visibility of the Config file")
+  @JsonProperty("visibility")
+  public LocalResourceVisibility getVisibility() {
+    return visibility;
+  }
+
+  @XmlElement(name = "visibility", defaultValue="APPLICATION")
+  public void setVisibility(LocalResourceVisibility localrsrcVisibility) {
+    this.visibility = localrsrcVisibility;
+  }
+
   /**
    A blob of key value pairs that will be dumped in the dest_file in the format
    as specified in type. If src_file is specified, src_file content are dumped
@@ -200,12 +223,13 @@ public class ConfigFile implements Serializable {
     return Objects.equals(this.type, configFile.type)
         && Objects.equals(this.destFile, configFile.destFile)
         && Objects.equals(this.srcFile, configFile.srcFile)
+        && Objects.equals(this.visibility, configFile.visibility)
         && Objects.equals(this.properties, configFile.properties);
   }
 
   @Override
   public int hashCode() {
-    return Objects.hash(type, destFile, srcFile, properties);
+    return Objects.hash(type, destFile, srcFile, visibility, properties);
   }
 
   @Override
@@ -217,6 +241,8 @@ public class ConfigFile implements Serializable {
         .append("    destFile: ").append(toIndentedString(destFile))
         .append("\n")
         .append("    srcFile: ").append(toIndentedString(srcFile)).append("\n")
+        .append("    visibility: ").append(toIndentedString(visibility))
+        .append("\n")
         .append("    properties: ").append(toIndentedString(properties))
         .append("\n")
         .append("}");
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/client/ServiceClient.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/client/ServiceClient.java
index 1276022f25f..46bfa7a4564 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/client/ServiceClient.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/client/ServiceClient.java
@@ -817,6 +817,21 @@ public class ServiceClient extends AppAdminClient implements SliderExitCodes,
           + appDir);
       ret = EXIT_NOT_FOUND;
     }
+
+    // Delete Public Resource Dir
+    Path publicResourceDir = new Path(fs.getBasePath(), serviceName);
+    if (fileSystem.exists(publicResourceDir)) {
+      if (fileSystem.delete(publicResourceDir, true)) {
+        LOG.info("Successfully deleted public resource dir for "
+            + serviceName + ": " + publicResourceDir);
+      } else {
+        String message = "Failed to delete public resource dir for service "
+            + serviceName + " at:  " + publicResourceDir;
+        LOG.info(message);
+        throw new YarnException(message);
+      }
+    }
+
     try {
       deleteZKNode(serviceName);
       // don't set destroySucceed to false if no ZK node exists because not
@@ -1315,7 +1330,8 @@ public class ServiceClient extends AppAdminClient implements SliderExitCodes,
             new Path(remoteConfPath, YarnServiceConstants.YARN_SERVICE_LOG4J_FILENAME);
         copy(conf, localFilePath, remoteFilePath);
         LocalResource localResource =
-            fs.createAmResource(remoteConfPath, LocalResourceType.FILE);
+            fs.createAmResource(remoteConfPath, LocalResourceType.FILE,
+            LocalResourceVisibility.APPLICATION);
         localResources.put(localFilePath.getName(), localResource);
         hasAMLog4j = true;
       } else {
@@ -1465,7 +1481,7 @@ public class ServiceClient extends AppAdminClient implements SliderExitCodes,
         return;
       }
       LocalResource keytabRes = fileSystem.createAmResource(keytabOnhdfs,
-          LocalResourceType.FILE);
+          LocalResourceType.FILE, LocalResourceVisibility.PRIVATE);
       localResource.put(String.format(YarnServiceConstants.KEYTAB_LOCATION,
           service.getName()), keytabRes);
       LOG.info("Adding " + service.getName() + "'s keytab for "
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/conf/YarnServiceConstants.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/conf/YarnServiceConstants.java
index 05135fe6196..dd940650c8a 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/conf/YarnServiceConstants.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/conf/YarnServiceConstants.java
@@ -47,6 +47,8 @@ public interface YarnServiceConstants {
 
   String SERVICES_DIRECTORY = "services";
 
+  String SERVICES_PUBLIC_DIRECTORY = "/tmp/hadoop-yarn/staging/";
+
   /**
    * JVM property to define the service lib directory;
    * this is set by the yarn.sh script
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/provider/ProviderUtils.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/provider/ProviderUtils.java
index 5fc96a09df2..0b091e22419 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/provider/ProviderUtils.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/provider/ProviderUtils.java
@@ -27,6 +27,7 @@ import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.LocalResource;
 import org.apache.hadoop.yarn.api.records.LocalResourceType;
+import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
 import org.apache.hadoop.yarn.service.ServiceContext;
 import org.apache.hadoop.yarn.service.api.records.ConfigFile;
 import org.apache.hadoop.yarn.service.api.records.ConfigFormat;
@@ -191,6 +192,17 @@ public class ProviderUtils implements YarnServiceConstants {
     return compInstanceDir;
   }
 
+  public static Path initCompPublicResourceDir(SliderFileSystem fs,
+      ContainerLaunchService.ComponentLaunchContext compLaunchContext,
+      ComponentInstance instance) {
+    Path compDir = fs.getComponentPublicResourceDir(
+        compLaunchContext.getServiceVersion(), compLaunchContext.getName());
+    Path compPublicResourceDir = new Path(compDir,
+        instance.getCompInstanceName());
+    return compPublicResourceDir;
+  }
+
+
   // 1. Create all config files for a component on hdfs for localization
   // 2. Add the config file to localResource
   public static synchronized void createConfigFileAndAddLocalResource(
@@ -212,6 +224,20 @@ public class ProviderUtils implements YarnServiceConstants {
       log.info("Component instance conf dir already exists: " + compInstanceDir);
     }
 
+    Path compPublicResourceDir = initCompPublicResourceDir(fs,
+        compLaunchContext, instance);
+    if (!fs.getFileSystem().exists(compPublicResourceDir)) {
+      log.info("{} version {} : Creating Public Resource dir on hdfs: {}",
+          instance.getCompInstanceId(), compLaunchContext.getServiceVersion(),
+          compPublicResourceDir);
+      fs.getFileSystem().mkdirs(compPublicResourceDir,
+          new FsPermission(FsAction.ALL, FsAction.READ_EXECUTE,
+          FsAction.EXECUTE));
+    } else {
+      log.info("Component instance public resource dir already exists: "
+          + compPublicResourceDir);
+    }
+
     log.debug("Tokens substitution for component instance: {}{}{}" + instance
         .getCompInstanceName(), System.lineSeparator(), tokensForSubstitution);
 
@@ -236,7 +262,14 @@ public class ProviderUtils implements YarnServiceConstants {
        * substitution and merges in new configs, and writes a new file to
        * compInstanceDir/fileName.
        */
-      Path remoteFile = new Path(compInstanceDir, fileName);
+      Path remoteFile = null;
+      LocalResourceVisibility visibility = configFile.getVisibility();
+      if (visibility != null &&
+          visibility.equals(LocalResourceVisibility.PUBLIC)) {
+        remoteFile = new Path(compPublicResourceDir, fileName);
+      } else {
+        remoteFile = new Path(compInstanceDir, fileName);
+      }
 
       if (!fs.getFileSystem().exists(remoteFile)) {
         log.info("Saving config file on hdfs for component " + instance
@@ -268,7 +301,8 @@ public class ProviderUtils implements YarnServiceConstants {
 
       // Add resource for localization
       LocalResource configResource =
-          fs.createAmResource(remoteFile, LocalResourceType.FILE);
+          fs.createAmResource(remoteFile, LocalResourceType.FILE,
+          configFile.getVisibility());
       Path destFile = new Path(configFile.getDestFile());
       String symlink = APP_CONF_DIR + "/" + fileName;
       addLocalResource(launcher, symlink, configResource, destFile,
@@ -311,7 +345,8 @@ public class ProviderUtils implements YarnServiceConstants {
       LocalResource localResource = fs.createAmResource(sourceFile,
           (staticFile.getType() == ConfigFile.TypeEnum.ARCHIVE ?
               LocalResourceType.ARCHIVE :
-              LocalResourceType.FILE));
+              LocalResourceType.FILE), staticFile.getVisibility());
+
       Path destFile = new Path(sourceFile.getName());
       if (staticFile.getDestFile() != null && !staticFile.getDestFile()
           .isEmpty()) {
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/provider/tarball/TarballProviderService.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/provider/tarball/TarballProviderService.java
index 87406f79228..cd783e77f76 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/provider/tarball/TarballProviderService.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/provider/tarball/TarballProviderService.java
@@ -20,6 +20,7 @@ package org.apache.hadoop.yarn.service.provider.tarball;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.yarn.api.records.LocalResource;
 import org.apache.hadoop.yarn.api.records.LocalResourceType;
+import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
 import org.apache.hadoop.yarn.service.api.records.Service;
 import org.apache.hadoop.yarn.service.component.instance.ComponentInstance;
 import org.apache.hadoop.yarn.service.containerlaunch.ContainerLaunchService;
@@ -43,7 +44,8 @@ public class TarballProviderService extends AbstractProviderService {
     }
     log.info("Adding resource {}", artifact);
     LocalResourceType type = LocalResourceType.ARCHIVE;
-    LocalResource packageResource = fileSystem.createAmResource(artifact, type);
+    LocalResource packageResource = fileSystem.createAmResource(artifact, type,
+        LocalResourceVisibility.APPLICATION);
     launcher.addLocalResource(APP_LIB_DIR, packageResource);
   }
 }
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/utils/CoreFileSystem.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/utils/CoreFileSystem.java
index b9a464960d5..0ee8e839807 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/utils/CoreFileSystem.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/utils/CoreFileSystem.java
@@ -384,13 +384,19 @@ public class CoreFileSystem {
    * @param resourceType resource type
    * @return the local resource for AM
    */
-  public LocalResource createAmResource(Path destPath, LocalResourceType resourceType) throws IOException {
+  public LocalResource createAmResource(Path destPath,
+      LocalResourceType resourceType,
+      LocalResourceVisibility visibility) throws IOException {
+
     FileStatus destStatus = fileSystem.getFileStatus(destPath);
     LocalResource amResource = Records.newRecord(LocalResource.class);
     amResource.setType(resourceType);
     // Set visibility of the resource
     // Setting to most private option
-    amResource.setVisibility(LocalResourceVisibility.APPLICATION);
+    if (visibility == null) {
+      visibility = LocalResourceVisibility.APPLICATION;
+    }
+    amResource.setVisibility(visibility);
     // Set the resource to be copied over
     amResource.setResource(
         URL.fromPath(fileSystem.resolvePath(destStatus.getPath())));
@@ -419,7 +425,7 @@ public class CoreFileSystem {
     for (FileStatus entry : fileset) {
 
       LocalResource resource = createAmResource(entry.getPath(),
-              LocalResourceType.FILE);
+              LocalResourceType.FILE, LocalResourceVisibility.APPLICATION);
       String relativePath = destRelativeDir + "/" + entry.getPath().getName();
       localResources.put(relativePath, resource);
     }
@@ -465,7 +471,8 @@ public class CoreFileSystem {
     // Set the type of resource - file or archive
     // archives are untarred at destination
     // we don't need the jar file to be untarred for now
-    return createAmResource(destPath, LocalResourceType.FILE);
+    return createAmResource(destPath, LocalResourceType.FILE,
+        LocalResourceVisibility.APPLICATION);
   }
 
   /**
@@ -483,7 +490,7 @@ public class CoreFileSystem {
       BadClusterStateException {
     Path dependencyLibTarGzip = getDependencyTarGzip();
     LocalResource lc = createAmResource(dependencyLibTarGzip,
-        LocalResourceType.ARCHIVE);
+        LocalResourceType.ARCHIVE, LocalResourceVisibility.APPLICATION);
     providerResources.put(YarnServiceConstants.DEPENDENCY_LOCALIZED_DIR_LINK, lc);
   }
 
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/utils/SliderFileSystem.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/utils/SliderFileSystem.java
index c7764764be8..4af97502269 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/utils/SliderFileSystem.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/main/java/org/apache/hadoop/yarn/service/utils/SliderFileSystem.java
@@ -21,6 +21,7 @@ package org.apache.hadoop.yarn.service.utils;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.yarn.service.conf.YarnServiceConstants;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -63,6 +64,26 @@ public class SliderFileSystem extends CoreFileSystem {
         serviceVersion + "/" + compName);
   }
 
+  public Path getBasePath() {
+    String tmpDir = configuration.get("hadoop.tmp.dir");
+    String basePath = YarnServiceConstants.SERVICE_BASE_DIRECTORY
+        + "/" + YarnServiceConstants.SERVICES_DIRECTORY;
+    return new Path(tmpDir, basePath);
+  }
+
+  /**
+   * Returns the component public resource directory path.
+   *
+   * @param serviceVersion service version
+   * @param compName       component name
+   * @return component public resource directory
+   */
+  public Path getComponentPublicResourceDir(String serviceVersion,
+      String compName) {
+    return new Path(new Path(getBasePath(), getAppDir().getName() + "/"
+        + "components"), serviceVersion + "/" + compName);
+  }
+
   /**
    * Deletes the component directory.
    *
@@ -77,6 +98,12 @@ public class SliderFileSystem extends CoreFileSystem {
       fileSystem.delete(path, true);
       LOG.debug("deleted dir {}", path);
     }
+    Path publicResourceDir = getComponentPublicResourceDir(serviceVersion,
+        compName);
+    if (fileSystem.exists(publicResourceDir)) {
+      fileSystem.delete(publicResourceDir, true);
+      LOG.debug("deleted public resource dir {}", publicResourceDir);
+    }
   }
 
   /**
@@ -92,6 +119,13 @@ public class SliderFileSystem extends CoreFileSystem {
       fileSystem.delete(path, true);
       LOG.info("deleted dir {}", path);
     }
+    Path publicResourceDir = new Path(new Path(getBasePath(),
+        getAppDir().getName() + "/" + "components"), serviceVersion);
+    if (fileSystem.exists(publicResourceDir)
+        && fileSystem.listStatus(publicResourceDir).length == 0) {
+      fileSystem.delete(publicResourceDir, true);
+      LOG.info("deleted public resource dir {}", publicResourceDir);
+    }
   }
 
 
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/test/java/org/apache/hadoop/yarn/service/provider/TestProviderUtils.java b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/test/java/org/apache/hadoop/yarn/service/provider/TestProviderUtils.java
index 84c3b6e020d..bfdcccd268c 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/test/java/org/apache/hadoop/yarn/service/provider/TestProviderUtils.java
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-applications/hadoop-yarn-services/hadoop-yarn-services-core/src/test/java/org/apache/hadoop/yarn/service/provider/TestProviderUtils.java
@@ -63,95 +63,100 @@ public class TestProviderUtils {
     List<ConfigFile> configFileList = new ArrayList<>();
     when(conf.getFiles()).thenReturn(configFileList);
     when(compLaunchCtx.getConfiguration()).thenReturn(conf);
-    when(sfs.createAmResource(any(Path.class), any(LocalResourceType.class)))
-        .thenAnswer(invocationOnMock -> new LocalResource() {
-          @Override
-          public URL getResource() {
-            return URL.fromPath(((Path) invocationOnMock.getArguments()[0]));
-          }
+    when(sfs.createAmResource(any(Path.class), any(LocalResourceType.class),
+        any(LocalResourceVisibility.class))).thenAnswer(
+          invocationOnMock -> new LocalResource() {
+            @Override
+            public URL getResource() {
+              return URL.fromPath(((Path) invocationOnMock.getArguments()[0]));
+            }
 
-          @Override
-          public void setResource(URL resource) {
+            @Override
+            public void setResource(URL resource) {
 
-          }
+            }
 
-          @Override
-          public long getSize() {
-            return 0;
-          }
+            @Override
+            public long getSize() {
+              return 0;
+            }
 
-          @Override
-          public void setSize(long size) {
+            @Override
+            public void setSize(long size) {
 
-          }
+            }
 
-          @Override
-          public long getTimestamp() {
-            return 0;
-          }
+            @Override
+            public long getTimestamp() {
+              return 0;
+            }
 
-          @Override
-          public void setTimestamp(long timestamp) {
+            @Override
+            public void setTimestamp(long timestamp) {
 
-          }
+            }
 
-          @Override
-          public LocalResourceType getType() {
-            return (LocalResourceType) invocationOnMock.getArguments()[1];
-          }
+            @Override
+            public LocalResourceType getType() {
+              return (LocalResourceType) invocationOnMock.getArguments()[1];
+            }
 
-          @Override
-          public void setType(LocalResourceType type) {
+            @Override
+            public void setType(LocalResourceType type) {
 
-          }
+            }
 
-          @Override
-          public LocalResourceVisibility getVisibility() {
-            return null;
-          }
+            @Override
+            public LocalResourceVisibility getVisibility() {
+              return LocalResourceVisibility.APPLICATION;
+            }
 
-          @Override
-          public void setVisibility(LocalResourceVisibility visibility) {
+            @Override
+            public void setVisibility(LocalResourceVisibility visibility) {
 
-          }
+            }
 
-          @Override
-          public String getPattern() {
-            return null;
-          }
+            @Override
+            public String getPattern() {
+              return null;
+            }
 
-          @Override
-          public void setPattern(String pattern) {
+            @Override
+            public void setPattern(String pattern) {
 
-          }
+            }
 
-          @Override
-          public boolean getShouldBeUploadedToSharedCache() {
-            return false;
-          }
+            @Override
+            public boolean getShouldBeUploadedToSharedCache() {
+              return false;
+            }
 
-          @Override
-          public void setShouldBeUploadedToSharedCache(
-              boolean shouldBeUploadedToSharedCache) {
+            @Override
+            public void setShouldBeUploadedToSharedCache(
+                boolean shouldBeUploadedToSharedCache) {
 
-          }
-        });
+            }
+          });
 
     // Initialize list of files.
     //archive
     configFileList.add(new ConfigFile().srcFile("hdfs://default/sourceFile1")
-        .destFile("destFile1").type(ConfigFile.TypeEnum.ARCHIVE));
+        .destFile("destFile1").type(ConfigFile.TypeEnum.ARCHIVE)
+        .visibility(LocalResourceVisibility.APPLICATION));
 
     //static file
     configFileList.add(new ConfigFile().srcFile("hdfs://default/sourceFile2")
-        .destFile("folder/destFile_2").type(ConfigFile.TypeEnum.STATIC));
+        .destFile("folder/destFile_2").type(ConfigFile.TypeEnum.STATIC)
+        .visibility(LocalResourceVisibility.APPLICATION));
 
     //This will be ignored since type is JSON
     configFileList.add(new ConfigFile().srcFile("hdfs://default/sourceFile3")
-        .destFile("destFile3").type(ConfigFile.TypeEnum.JSON));
+        .destFile("destFile3").type(ConfigFile.TypeEnum.JSON)
+        .visibility(LocalResourceVisibility.APPLICATION));
     //No destination file specified
     configFileList.add(new ConfigFile().srcFile("hdfs://default/sourceFile4")
-        .type(ConfigFile.TypeEnum.STATIC));
+        .type(ConfigFile.TypeEnum.STATIC)
+        .visibility(LocalResourceVisibility.APPLICATION));
 
     ProviderService.ResolvedLaunchParams resolved =
         new ProviderService.ResolvedLaunchParams();
diff --git a/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 b/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
index 50721debe5e..dce24908a51 100644
--- a/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
+++ b/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
@@ -235,6 +235,9 @@ public class DockerLinuxContainerRuntime extends OCIContainerRuntime {
   @InterfaceAudience.Private
   public static final String ENV_DOCKER_CONTAINER_DOCKER_RUNTIME =
       "YARN_CONTAINER_RUNTIME_DOCKER_RUNTIME";
+  @InterfaceAudience.Private
+  public static final String ENV_DOCKER_CONTAINER_DOCKER_SERVICE_MODE =
+      "YARN_CONTAINER_RUNTIME_DOCKER_SERVICE_MODE";
 
   @InterfaceAudience.Private
   private static final String RUNTIME_TYPE = "DOCKER";
@@ -588,7 +591,9 @@ public class DockerLinuxContainerRuntime extends OCIContainerRuntime {
     String network = environment.get(ENV_DOCKER_CONTAINER_NETWORK);
     String hostname = environment.get(ENV_DOCKER_CONTAINER_HOSTNAME);
     String runtime = environment.get(ENV_DOCKER_CONTAINER_DOCKER_RUNTIME);
-    boolean useEntryPoint = checkUseEntryPoint(environment);
+    boolean serviceMode = Boolean.parseBoolean(environment.get(
+        ENV_DOCKER_CONTAINER_DOCKER_SERVICE_MODE));
+    boolean useEntryPoint = serviceMode || checkUseEntryPoint(environment);
 
     if (imageName == null || imageName.isEmpty()) {
       imageName = defaultImageName;
@@ -679,10 +684,12 @@ public class DockerLinuxContainerRuntime extends OCIContainerRuntime {
       runCommand.addRuntime(runtime);
     }
 
-    runCommand.addAllReadWriteMountLocations(containerLogDirs);
-    runCommand.addAllReadWriteMountLocations(applicationLocalDirs);
-    runCommand.addAllReadOnlyMountLocations(filecacheDirs);
-    runCommand.addAllReadOnlyMountLocations(userFilecacheDirs);
+    if (!serviceMode) {
+      runCommand.addAllReadWriteMountLocations(containerLogDirs);
+      runCommand.addAllReadWriteMountLocations(applicationLocalDirs);
+      runCommand.addAllReadOnlyMountLocations(filecacheDirs);
+      runCommand.addAllReadOnlyMountLocations(userFilecacheDirs);
+    }
 
     if (environment.containsKey(ENV_DOCKER_CONTAINER_MOUNTS)) {
       Matcher parsedMounts = USER_MOUNT_PATTERN.matcher(
@@ -800,11 +807,20 @@ public class DockerLinuxContainerRuntime extends OCIContainerRuntime {
       runCommand.setYarnSysFS(true);
     }
 
+    // In service mode, the YARN log dirs are not mounted into the container.
+    // As a result, the container fails to start due to stdout and stderr output
+    // being sent to a file in a directory that does not exist. In service mode,
+    // only supply the command with no stdout or stderr redirection.
+    List<String> commands = container.getLaunchContext().getCommands();
+    if (serviceMode) {
+      commands = Arrays.asList(
+          String.join(" ", commands).split("1>")[0].split(" "));
+    }
+
     if (useEntryPoint) {
       runCommand.setOverrideDisabled(true);
       runCommand.addEnv(environment);
-      runCommand.setOverrideCommandWithArgs(container.getLaunchContext()
-          .getCommands());
+      runCommand.setOverrideCommandWithArgs(commands);
       runCommand.disableDetach();
       runCommand.setLogDir(container.getLogDir());
     } else {
@@ -818,6 +834,10 @@ public class DockerLinuxContainerRuntime extends OCIContainerRuntime {
       runCommand.detachOnRun();
     }
 
+    if (serviceMode) {
+      runCommand.setServiceMode(serviceMode);
+    }
+
     if(enableUserReMapping) {
       if (!allowPrivilegedContainerExecution(container)) {
         runCommand.groupAdd(groups);
@@ -1279,11 +1299,14 @@ public class DockerLinuxContainerRuntime extends OCIContainerRuntime {
       throw new ContainerExecutionException(e);
     }
 
+    boolean serviceMode = Boolean.parseBoolean(env.get(
+        ENV_DOCKER_CONTAINER_DOCKER_SERVICE_MODE));
+
     // Only need to check whether the container was asked to be privileged.
     // If the container had failed the permissions checks upon launch, it
     // would have never been launched and thus we wouldn't be here
     // attempting to signal it.
-    if (isContainerRequestedAsPrivileged(container)) {
+    if (isContainerRequestedAsPrivileged(container) || serviceMode) {
       String containerId = container.getContainerId().toString();
       DockerCommandExecutor.DockerContainerStatus containerStatus =
           DockerCommandExecutor.getContainerStatus(containerId,
diff --git a/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 b/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
index b0603a3a220..7fb0e40c442 100644
--- a/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
+++ b/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
@@ -199,6 +199,12 @@ public class DockerRunCommand extends DockerCommand {
     return this;
   }
 
+  public DockerRunCommand setServiceMode(boolean serviceMode) {
+    String value = Boolean.toString(serviceMode);
+    super.addCommandArguments("service-mode", value);
+    return this;
+  }
+
   /**
    * Check if user defined environment variables are empty.
    *
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.h b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.h
index b215af72a87..757bd16c63a 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.h
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/container-executor.h
@@ -325,12 +325,6 @@ int sync_yarn_sysfs(char* const* local_dirs, const char *running_user,
  */
 int execute_regex_match(const char *regex_str, const char *input);
 
-/**
- * Validate the docker image name matches the expected input.
- * Return 0 on success.
- */
-int validate_docker_image_name(const char *image_name);
-
 struct configuration* get_cfg();
 
 /**
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/utils/docker-util.c b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/utils/docker-util.c
index 17114338e72..3ef571fdeff 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/utils/docker-util.c
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/utils/docker-util.c
@@ -28,6 +28,7 @@
 #include "docker-util.h"
 #include "string-utils.h"
 #include "util.h"
+#include "container-executor.h"
 #include <grp.h>
 #include <pwd.h>
 #include <errno.h>
@@ -374,6 +375,8 @@ const char *get_docker_error_message(const int error_code) {
       return "Invalid docker tmpfs mount";
     case INVALID_DOCKER_RUNTIME:
       return "Invalid docker runtime";
+    case SERVICE_MODE_DISABLED:
+      return "Service mode disabled";
     default:
       return "Unknown error";
   }
@@ -987,6 +990,22 @@ static int set_runtime(const struct configuration *command_config,
   return ret;
 }
 
+int is_service_mode_enabled(const struct configuration *command_config,
+                            const struct configuration *executor_cfg, args *args) {
+    int ret = 0;
+    struct section *section = get_configuration_section(CONTAINER_EXECUTOR_CFG_DOCKER_SECTION, executor_cfg);
+    char *value = get_configuration_value("service-mode", DOCKER_COMMAND_FILE_SECTION, command_config);
+    if (value != NULL && strcasecmp(value, "true") == 0) {
+      if (is_feature_enabled(DOCKER_SERVICE_MODE_ENABLED_KEY, ret, section)) {
+        ret = 1;
+      } else {
+        ret = SERVICE_MODE_DISABLED;
+      }
+     }
+    free(value);
+    return ret;
+}
+
 static int add_ports_mapping_to_command(const struct configuration *command_config, args *args) {
   int i = 0, ret = 0;
   char *network_type = (char*) malloc(128);
@@ -1595,12 +1614,19 @@ int get_docker_run_command(const char *command_file, const struct configuration
   char *privileged = NULL;
   char *no_new_privileges_enabled = NULL;
   char *use_entry_point = NULL;
+  int service_mode_enabled = 0;
   struct configuration command_config = {0, NULL};
   ret = read_and_verify_command_file(command_file, DOCKER_RUN_COMMAND, &command_config);
   if (ret != 0) {
     goto free_and_exit;
   }
 
+  service_mode_enabled = is_service_mode_enabled(&command_config, conf, args);
+  if (service_mode_enabled == SERVICE_MODE_DISABLED) {
+    ret = SERVICE_MODE_DISABLED;
+    goto free_and_exit;
+  }
+
   use_entry_point = get_configuration_value("use-entry-point", DOCKER_COMMAND_FILE_SECTION, &command_config);
   if (use_entry_point != NULL && strcasecmp(use_entry_point, "true") == 0) {
     entry_point = 1;
@@ -1612,10 +1638,13 @@ int get_docker_run_command(const char *command_file, const struct configuration
     ret = INVALID_DOCKER_CONTAINER_NAME;
     goto free_and_exit;
   }
-  user = get_configuration_value("user", DOCKER_COMMAND_FILE_SECTION, &command_config);
-  if (user == NULL) {
-    ret = INVALID_DOCKER_USER_NAME;
-    goto free_and_exit;
+
+  if (!service_mode_enabled) {
+    user = get_configuration_value("user", DOCKER_COMMAND_FILE_SECTION, &command_config);
+    if (user == NULL) {
+      ret = INVALID_DOCKER_USER_NAME;
+      goto free_and_exit;
+    }
   }
   image = get_configuration_value("image", DOCKER_COMMAND_FILE_SECTION, &command_config);
   if (image == NULL || validate_docker_image_name(image) != 0) {
@@ -1640,12 +1669,14 @@ int get_docker_run_command(const char *command_file, const struct configuration
   privileged = get_configuration_value("privileged", DOCKER_COMMAND_FILE_SECTION, &command_config);
 
   if (privileged == NULL || strcmp(privileged, "false") == 0) {
-    char *user_buffer = make_string("--user=%s", user);
-    ret = add_to_args(args, user_buffer);
-    free(user_buffer);
-    if (ret != 0) {
-      ret = BUFFER_TOO_SMALL;
-      goto free_and_exit;
+    if (!service_mode_enabled) {
+      char *user_buffer = make_string("--user=%s", user);
+      ret = add_to_args(args, user_buffer);
+      free(user_buffer);
+      if (ret != 0) {
+        ret = BUFFER_TOO_SMALL;
+        goto free_and_exit;
+      }
     }
     no_new_privileges_enabled =
         get_configuration_value("docker.no-new-privileges.enabled",
@@ -1725,9 +1756,11 @@ int get_docker_run_command(const char *command_file, const struct configuration
     goto free_and_exit;
   }
 
-  ret = set_group_add(&command_config, args);
-  if (ret != 0) {
-    goto free_and_exit;
+  if (!service_mode_enabled) {
+    ret = set_group_add(&command_config, args);
+    if (ret != 0) {
+      goto free_and_exit;
+    }
   }
 
   ret = set_devices(&command_config, conf, args);
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/utils/docker-util.h b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/utils/docker-util.h
index 07da195629a..d9d34a0640a 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/utils/docker-util.h
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/native/container-executor/impl/utils/docker-util.h
@@ -36,6 +36,7 @@
 #define DOCKER_START_COMMAND "start"
 #define DOCKER_EXEC_COMMAND "exec"
 #define DOCKER_IMAGES_COMMAND "images"
+#define DOCKER_SERVICE_MODE_ENABLED_KEY "docker.service-mode.enabled"
 #define DOCKER_ARG_MAX 1024
 #define ARGS_INITIAL_VALUE { 0 };
 
@@ -71,7 +72,8 @@ enum docker_error_codes {
     INVALID_PID_NAMESPACE,
     INVALID_DOCKER_IMAGE_TRUST,
     INVALID_DOCKER_TMPFS_MOUNT,
-    INVALID_DOCKER_RUNTIME
+    INVALID_DOCKER_RUNTIME,
+    SERVICE_MODE_DISABLED
 };
 
 /**
diff --git a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/DockerContainers.md b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/DockerContainers.md
index e30ac9808eb..db9c56d99ee 100644
--- a/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/DockerContainers.md
+++ b/hadoop-yarn-project/hadoop-yarn/hadoop-yarn-site/src/site/markdown/DockerContainers.md
@@ -285,6 +285,7 @@ are allowed. It contains the following properties:
 | `docker.inspect.max.retries` | Integer value to check docker container readiness.  Each inspection is set with 3 seconds delay.  Default value of 10 will wait 30 seconds for docker container to become ready before marked as container failed. |
 | `docker.no-new-privileges.enabled` | Enable/disable the no-new-privileges flag for docker run. Set to "true" to enable, disabled by default. |
 | `docker.allowed.runtimes` | Comma seperated runtimes that containers are allowed to use. By default no runtimes are allowed to be added.|
+| `docker.service-mode.enabled` | Set to "true" or "false" to enable or disable docker container service mode. Default value is "false". |
 
 Please note that if you wish to run Docker containers that require access to the YARN local directories, you must add them to the docker.allowed.rw-mounts list.
 
@@ -436,6 +437,7 @@ environment variables in the application's environment:
 | `YARN_CONTAINER_RUNTIME_DOCKER_TMPFS_MOUNTS` | Adds additional tmpfs mounts to the Docker container. The value of the environment variable should be a comma-separated list of absolute mount points within the container. |
 | `YARN_CONTAINER_RUNTIME_DOCKER_DELAYED_REMOVAL` | Allows a user to request delayed deletion of the Docker container on a per container basis. If true, Docker containers will not be removed until the duration defined by yarn.nodemanager.delete.debug-delay-sec has elapsed. Administrators can disable this feature through the yarn-site property yarn.nodemanager.runtime.linux.docker.delayed-removal.allowed. This feature is disabled by default. When this feature is disabled or set to false, the container will be removed as soon as it exits. |
 | `YARN_CONTAINER_RUNTIME_YARN_SYSFS_ENABLE` | Enable mounting of container working directory sysfs sub-directory into Docker container /hadoop/yarn/sysfs.  This is useful for populating cluster information into container. |
+| `YARN_CONTAINER_RUNTIME_DOCKER_SERVICE_MODE` | Enable Service Mode which runs the docker container as defined by the image but does not set the user (--user and --group-add). |
 
 The first two are required. The remainder can be set as needed. While
 controlling the container type through environment variables is somewhat less
@@ -1080,3 +1082,24 @@ YARN service framework automatically populates cluster information
 to /hadoop/yarn/sysfs/app.json.  For more information about
 YARN service, see: [YARN Service](./yarn-service/Overview.html).
 
+Docker Container Service Mode
+-----------------------------
+
+Docker Container Service Mode runs the container as defined by the image
+but does not set the user (--user and --group-add). This mode is disabled
+by default. The administrator sets docker.service-mode.enabled to true
+in container-executor.cfg under docker section to enable.
+
+Part of a container-executor.cfg which allows docker service mode is below:
+
+```
+yarn.nodemanager.linux-container-executor.group=yarn
+[docker]
+  module.enabled=true
+  docker.privileged-containers.enabled=true
+  docker.service-mode.enabled=true
+```
+
+Application User can enable or disable service mode at job level by exporting
+environment variable YARN_CONTAINER_RUNTIME_DOCKER_SERVICE_MODE in the application's
+environment with value true or false respectively.