review, comments

Author: yaooqinn

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

@@ -22,6 +22,7 @@ import java.security.PrivilegedExceptionAction
 import java.text.DateFormat
 import java.util.{Arrays, Comparator, Date, Locale}
 
+import scala.collection.immutable.Map
 import scala.collection.JavaConverters._
 import scala.collection.mutable
 import scala.collection.mutable.HashMap
@@ -75,7 +76,6 @@ class SparkHadoopUtil extends Logging {
     }
   }
 
-
   /**
    * Appends S3-specific, spark.hadoop.*, and spark.buffer.size configurations to a Hadoop
    * configuration.
@@ -106,21 +106,26 @@ class SparkHadoopUtil extends Logging {
     }
   }
 
+  /**
+   * Appends spark.hadoop.* configurations from a [[SparkConf]] to a Hadoop
+   * configuration without the spark.hadoop. prefix.
+   */
   def appendSparkHadoopConfigs(conf: SparkConf, hadoopConf: Configuration): Unit = {
-    // Copy any "spark.hadoop.foo=bar" system properties into conf as "foo=bar"
-    conf.getAll.foreach { case (key, value) =>
-      if (key.startsWith("spark.hadoop.")) {
-        hadoopConf.set(key.substring("spark.hadoop.".length), value)
-      }
+    // Copy any "spark.hadoop.foo=bar" spark properties into conf as "foo=bar"
+    conf.getAll.foreach { case (key, value) if key.startsWith("spark.hadoop.") =>
+      hadoopConf.set(key.substring("spark.hadoop.".length), value)
     }
   }
 
-  def appendSparkHadoopConfigs(propMap: HashMap[String, String]): Unit = {
-    // Copy any "spark.hadoop.foo=bar" system properties into conf as "foo=bar"
-    sys.props.foreach { case (key, value) =>
-      if (key.startsWith("spark.hadoop.")) {
-        propMap.put(key.substring("spark.hadoop.".length), value)
-      }
+  /**
+   * Appends spark.hadoop.* configurations from a Map to another without the spark.hadoop. prefix.
+   */
+  def appendSparkHadoopConfigs(
+      srcMap: Map[String, String],
+      destMap: HashMap[String, String]): Unit = {
+    // Copy any "spark.hadoop.foo=bar" system properties into destMap as "foo=bar"
+    srcMap.foreach { case (key, value) if key.startsWith("spark.hadoop.") =>
+      destMap.put(key.substring("spark.hadoop.".length), value)
     }
   }
 

sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala

@@ -405,7 +405,12 @@ private[spark] object HiveUtils extends Logging {
     propMap.put(ConfVars.METASTORE_EVENT_LISTENERS.varname, "")
     propMap.put(ConfVars.METASTORE_END_FUNCTION_LISTENERS.varname, "")
 
-    SparkHadoopUtil.get.appendSparkHadoopConfigs(propMap)
+    // SPARK-21451: Spark will gather all `spark.hadoop.*` properties from a `SparkConf` to a
+    // Hadoop Configuration internally, as long as it happens after SparkContext initialized.
+    // Some instances such as `CliSessionState` used in `SparkSQLCliDriver` may also rely on these
+    // Configuration. But it happens before SparkContext initialized, we need to take them from
+    // system properties in the form of regular hadoop configurations.
+    SparkHadoopUtil.get.appendSparkHadoopConfigs(sys.props.toMap, propMap)
 
     propMap.toMap
   }