Add spark.yarn.access.all.hadoopFileSystems

Author: wangyum

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/YarnSparkHadoopUtil.scala

@@ -200,10 +200,12 @@ object YarnSparkHadoopUtil {
       .map(new Path(_).getFileSystem(hadoopConf))
       .getOrElse(FileSystem.get(hadoopConf))
 
+    val accessAllFileSystem = sparkConf.get(FILESYSTEMS_TO_ACCESS_ALL)
+
     // Add the list of available namenodes for all namespaces in HDFS federation.
     // If ViewFS is enabled, this is skipped as ViewFS already handles delegation tokens for its
     // namespaces.
-    val hadoopFilesystems = if (stagingFS.getScheme == "viewfs") {
+    val hadoopFilesystems = if (accessAllFileSystem || stagingFS.getScheme == "viewfs") {
       Set.empty
     } else {
       val nameservices = hadoopConf.getTrimmedStrings("dfs.nameservices")

resource-managers/yarn/src/main/scala/org/apache/spark/deploy/yarn/config.scala

@@ -243,6 +243,14 @@ package object config {
       "that hosts fs.defaultFS does not need to be listed here.")
     .fallbackConf(NAMENODES_TO_ACCESS)
 
+  private[spark] val FILESYSTEMS_TO_ACCESS_ALL =
+    ConfigBuilder("spark.yarn.access.all.hadoopFileSystems")
+      .doc("Whether to get tokens of all filesystem configured in hdfs-site.xml. " +
+        "The default value is false. " +
+        "If true, a filesystem failed to get token may cause the entire job failed.")
+      .booleanConf
+      .createWithDefault(false)
+
   /* Rolled log aggregation configuration. */
 
   private[spark] val ROLLED_LOG_INCLUDE_PATTERN =