[SPARK-25200][YARN] Allow specifying HADOOP_CONF_DIR as spark property (apache-spark-on-k8s#410)

Author: robert3005

core/src/main/scala/org/apache/spark/deploy/SparkSubmitArguments.scala

@@ -286,9 +286,11 @@ private[deploy] class SparkSubmitArguments(args: Seq[String], env: Map[String, S
 
     if (master.startsWith("yarn")) {
       val hasHadoopEnv = env.contains("HADOOP_CONF_DIR") || env.contains("YARN_CONF_DIR")
-      if (!hasHadoopEnv && !Utils.isTesting) {
+      val hasHadoopProp = sparkProperties.contains("spark.yarn.conf.dir")
+      if (!hasHadoopEnv && !hasHadoopProp && !Utils.isTesting) {
         error(s"When running with master '$master' " +
-          "either HADOOP_CONF_DIR or YARN_CONF_DIR must be set in the environment.")
+          "either HADOOP_CONF_DIR or YARN_CONF_DIR must be set in the environment, +" +
+          "or spark.yarn.conf.dir in the spark properties.")
       }
     }
 

launcher/src/main/java/org/apache/spark/launcher/AbstractCommandBuilder.java

@@ -200,6 +200,7 @@ List<String> buildClassPath(String appClassPath) throws IOException {
 
     addToClassPath(cp, getenv("HADOOP_CONF_DIR"));
     addToClassPath(cp, getenv("YARN_CONF_DIR"));
+    addToClassPath(cp, getEffectiveConfig().get("spark.yarn.conf.dir"));
     addToClassPath(cp, getenv("SPARK_DIST_CLASSPATH"));
     return new ArrayList<>(cp);
   }

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

@@ -697,21 +697,21 @@ private[spark] class Client(
     // SPARK-23630: during testing, Spark scripts filter out hadoop conf dirs so that user's
     // environments do not interfere with tests. This allows a special env variable during
     // tests so that custom conf dirs can be used by unit tests.
-    val confDirs = Seq("HADOOP_CONF_DIR", "YARN_CONF_DIR") ++
-      (if (Utils.isTesting) Seq("SPARK_TEST_HADOOP_CONF_DIR") else Nil)
-
-    confDirs.foreach { envKey =>
-      sys.env.get(envKey).foreach { path =>
-        val dir = new File(path)
-        if (dir.isDirectory()) {
-          val files = dir.listFiles()
-          if (files == null) {
-            logWarning("Failed to list files under directory " + dir)
-          } else {
-            files.foreach { file =>
-              if (file.isFile && !hadoopConfFiles.contains(file.getName())) {
-                hadoopConfFiles(file.getName()) = file
-              }
+    val confDirs = Seq("HADOOP_CONF_DIR", "YARN_CONF_DIR").map(sys.env.get)
+    val confDirProp = sparkConf.getOption("spark.yarn.conf.dir")
+
+    val confDirPaths = (confDirs :+ confDirProp).flatten
+    confDirPaths.foreach { path =>
+      logDebug("Reading config files from " + path)
+      val dir = new File(path)
+      if (dir.isDirectory) {
+        val files = dir.listFiles()
+        if (files == null) {
+          logWarning("Failed to list files under directory " + dir)
+        } else {
+          files.foreach { file =>
+            if (file.isFile && !hadoopConfFiles.contains(file.getName)) {
+              hadoopConfFiles(file.getName) = file
             }
           }
         }

resource-managers/yarn/src/test/scala/org/apache/spark/deploy/yarn/YarnClusterSuite.scala

@@ -168,8 +168,9 @@ class YarnClusterSuite extends BaseYarnClusterSuite {
     val finalState = runSpark(false,
       mainClassName(YarnClusterDriverUseSparkHadoopUtilConf.getClass),
       appArgs = Seq("key=value", "spark.test.key=testvalue", result.getAbsolutePath()),
-      extraConf = Map("spark.hadoop.key" -> "value"),
-      extraEnv = Map("SPARK_TEST_HADOOP_CONF_DIR" -> customConf.getAbsolutePath()))
+      extraConf = Map(
+        "spark.hadoop.key" -> "value",
+        "spark.yarn.conf.dir" -> customConf.getAbsolutePath))
     checkResult(finalState, result)
   }