From 394b4716daefd76cf99b6eb0f90c37b23cfa12d1 Mon Sep 17 00:00:00 2001
From: Kent Yao <yaooqinn@hotmail.com>
Date: Sun, 16 Jul 2017 23:52:37 +0800
Subject: [PATCH 1/9] set isolateOn to false while using builtin hive jars

---
 .../src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala    | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
index a29d7a7565ee..43522daed455 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
@@ -312,7 +312,7 @@ private[spark] object HiveUtils extends Logging {
         hadoopConf = hadoopConf,
         execJars = jars.toSeq,
         config = configurations,
-        isolationOn = true,
+        isolationOn = false,
         barrierPrefixes = hiveMetastoreBarrierPrefixes,
         sharedPrefixes = hiveMetastoreSharedPrefixes)
     } else if (hiveMetastoreJars == "maven") {

From 62049f50c6decbec90bf72fa7ea8cfd0a1347ce2 Mon Sep 17 00:00:00 2001
From: Kent Yao <yaooqinn@hotmail.com>
Date: Mon, 17 Jul 2017 11:13:48 +0800
Subject: [PATCH 2/9] add unit tests

---
 sql/hive/pom.xml                              |  5 ++
 .../spark/sql/hive/HiveUtilsSuite.scala       | 48 +++++++++++++++++++
 2 files changed, 53 insertions(+)

diff --git a/sql/hive/pom.xml b/sql/hive/pom.xml
index f9462e79a69f..538329c21b46 100644
--- a/sql/hive/pom.xml
+++ b/sql/hive/pom.xml
@@ -98,6 +98,11 @@
       <groupId>${hive.group}</groupId>
       <artifactId>hive-metastore</artifactId>
     </dependency>
+    <dependency>
+      <groupId>${hive.group}</groupId>
+      <artifactId>hive-cli</artifactId>
+      <scope>test</scope>
+    </dependency>
     <!--
         <dependency>
           <groupId>${hive.group}</groupId>
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveUtilsSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveUtilsSuite.scala
index 667a7ddd8bb6..586e299321c1 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveUtilsSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveUtilsSuite.scala
@@ -17,8 +17,13 @@
 
 package org.apache.spark.sql.hive
 
+import org.apache.hadoop.hive.cli.CliSessionState
+import org.apache.hadoop.hive.conf.HiveConf
 import org.apache.hadoop.hive.conf.HiveConf.ConfVars
+import org.apache.hadoop.hive.ql.session.SessionState
 
+import org.apache.spark.SparkConf
+import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.sql.hive.test.TestHiveSingleton
 import org.apache.spark.sql.test.SQLTestUtils
 import org.apache.spark.sql.QueryTest
@@ -33,4 +38,47 @@ class HiveUtilsSuite extends QueryTest with SQLTestUtils with TestHiveSingleton
       assert(conf(ConfVars.METASTORE_END_FUNCTION_LISTENERS.varname) === "")
     }
   }
+
+  test("CliSessionState will be reused") {
+    val hiveConf = new HiveConf(classOf[SessionState])
+    val sessionState: SessionState = new CliSessionState(hiveConf)
+    SessionState.start(sessionState)
+    val s1 = SessionState.get()
+    assert(s1.isInstanceOf[CliSessionState])
+
+    val sparkConf = new SparkConf()
+    val hadoopConf = SparkHadoopUtil.get.newConfiguration(sparkConf)
+    val metaHive = HiveUtils.newClientForMetadata(sparkConf, hadoopConf)
+    assert(metaHive.getConf("spark.sql.hive.jars", "builtin") === "builtin")
+
+    val s2 = SessionState.get()
+    assert(s2.isInstanceOf[CliSessionState])
+    assert(s1 === s2, "CliSessionState should be reused")
+  }
+
+  test("SessionState will not be reused") {
+    val hiveConf = new HiveConf(classOf[SessionState])
+    val sessionState: SessionState = new SessionState(hiveConf)
+    SessionState.start(sessionState)
+
+    val s1 = SessionState.get()
+    assert(!s1.isInstanceOf[CliSessionState])
+
+    val sparkConf = new SparkConf().set("spark.sql.hive.jars", "builtin")
+    val hadoopConf = SparkHadoopUtil.get.newConfiguration(sparkConf)
+    val metaHive = HiveUtils.newClientForMetadata(sparkConf, hadoopConf)
+
+    assert(metaHive.getConf("spark.sql.hive.jars", "builtin") === "builtin")
+    val s2 = SessionState.get()
+    assert(s1 !== s2)
+
+    val sparkConf2 = sparkConf.set("spark.sql.hive.jars", "maven")
+    val hadoopConf2 = SparkHadoopUtil.get.newConfiguration(sparkConf2)
+    val metaHive2 = HiveUtils.newClientForMetadata(sparkConf2, hadoopConf2)
+
+    assert(metaHive2.getConf("spark.sql.hive.jars", "builtin") === "maven")
+    val s3 = SessionState.get()
+    assert(s1 !== s3)
+    assert(s2 !== s3)
+  }
 }

From d127f96e2926222599255f687260b3bbaa86bd87 Mon Sep 17 00:00:00 2001
From: hzyaoqin <hzyaoqin@corp.netease.com>
Date: Fri, 4 Aug 2017 16:27:55 +0800
Subject: [PATCH 3/9] refacting code

---
 .../scala/org/apache/spark/sql/hive/HiveUtils.scala | 13 ++++++++++++-
 .../spark/sql/hive/client/HiveClientImpl.scala      | 13 ++-----------
 2 files changed, 14 insertions(+), 12 deletions(-)

diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
index 43522daed455..a586baf2d5cd 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
@@ -32,6 +32,7 @@ import org.apache.hadoop.conf.Configuration
 import org.apache.hadoop.hive.common.`type`.HiveDecimal
 import org.apache.hadoop.hive.conf.HiveConf
 import org.apache.hadoop.hive.conf.HiveConf.ConfVars
+import org.apache.hadoop.hive.ql.session.SessionState
 import org.apache.hadoop.hive.serde2.io.{DateWritable, TimestampWritable}
 import org.apache.hadoop.util.VersionInfo
 
@@ -229,6 +230,16 @@ private[spark] object HiveUtils extends Logging {
     }.toMap
   }
 
+  def isCliSessionState(state: SessionState = SessionState.get): Boolean = {
+    var temp: Class[_] = if (state != null) state.getClass else null
+    var found = false
+    while (temp != null && !found) {
+      found = temp.getName == "org.apache.hadoop.hive.cli.CliSessionState"
+      temp = temp.getSuperclass
+    }
+    found
+  }
+
   /**
    * Create a [[HiveClient]] used for execution.
    *
@@ -312,7 +323,7 @@ private[spark] object HiveUtils extends Logging {
         hadoopConf = hadoopConf,
         execJars = jars.toSeq,
         config = configurations,
-        isolationOn = false,
+        isolationOn = isCliSessionState(),
         barrierPrefixes = hiveMetastoreBarrierPrefixes,
         sharedPrefixes = hiveMetastoreSharedPrefixes)
     } else if (hiveMetastoreJars == "maven") {
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index be024adac8eb..1d825a9ad808 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -49,6 +49,7 @@ import org.apache.spark.sql.catalyst.expressions.Expression
 import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException}
 import org.apache.spark.sql.execution.QueryExecutionException
 import org.apache.spark.sql.execution.command.DDLUtils
+import org.apache.spark.sql.hive.HiveUtils
 import org.apache.spark.sql.hive.client.HiveClientImpl._
 import org.apache.spark.sql.types._
 import org.apache.spark.util.{CircularBuffer, Utils}
@@ -123,20 +124,10 @@ private[hive] class HiveClientImpl(
       }
     }
 
-    def isCliSessionState(state: SessionState): Boolean = {
-      var temp: Class[_] = if (state != null) state.getClass else null
-      var found = false
-      while (temp != null && !found) {
-        found = temp.getName == "org.apache.hadoop.hive.cli.CliSessionState"
-        temp = temp.getSuperclass
-      }
-      found
-    }
-
     val ret = try {
       // originState will be created if not exists, will never be null
       val originalState = SessionState.get()
-      if (isCliSessionState(originalState)) {
+      if (HiveUtils.isCliSessionState(originalState)) {
         // In `SparkSQLCLIDriver`, we have already started a `CliSessionState`,
         // which contains information like configurations from command line. Later
         // we call `SparkSQLEnv.init()` there, which would run into this part again.

From d18bcc0e4cd8d80cf5731c3b58bf8659f0290423 Mon Sep 17 00:00:00 2001
From: hzyaoqin <hzyaoqin@corp.netease.com>
Date: Tue, 8 Aug 2017 10:16:27 +0800
Subject: [PATCH 4/9] move suites to thriftserver module for bannedDependencies

---
 .../apache/spark/deploy/SparkHadoopUtil.scala |  13 +-
 .../org/apache/spark/deploy/SparkSubmit.scala |  22 ++-
 .../HiveCliSessionStateSuite.scala            |  58 ++++++++
 sql/hive/pom.xml                              |   5 -
 .../org/apache/spark/sql/hive/HiveUtils.scala |   5 +-
 .../sql/hive/client/HiveClientImpl.scala      | 140 +++++++-----------
 .../spark/sql/hive/HiveUtilsSuite.scala       |  48 ------
 7 files changed, 138 insertions(+), 153 deletions(-)
 create mode 100644 sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala

diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala b/core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala
index 6afe58bff522..d623cd9c4a5f 100644
--- a/core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala
@@ -17,7 +17,7 @@
 
 package org.apache.spark.deploy
 
-import java.io.IOException
+import java.io.{File, IOException}
 import java.security.PrivilegedExceptionAction
 import java.text.DateFormat
 import java.util.{Arrays, Comparator, Date, Locale}
@@ -136,8 +136,15 @@ class SparkHadoopUtil extends Logging {
 
   def getSecretKeyFromUserCredentials(key: String): Array[Byte] = { null }
 
-  def loginUserFromKeytab(principalName: String, keytabFilename: String) {
-    UserGroupInformation.loginUserFromKeytab(principalName, keytabFilename)
+  def loginUserFromKeytab(principalName: String, keytabFilename: String): Unit = {
+    if (!new File(keytabFilename).exists()) {
+      throw new SparkException(s"Keytab file: ${keytabFilename}" +
+        " specified in spark.yarn.keytab does not exist")
+    } else {
+      logInfo("Attempting to login to Kerberos" +
+        s" using principal: ${principalName} and keytab: ${keytabFilename}")
+      UserGroupInformation.loginUserFromKeytab(principalName, keytabFilename)
+    }
   }
 
   /**
diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
index 0ea14361b2f7..765283d02319 100644
--- a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
@@ -28,7 +28,7 @@ import javax.net.ssl._
 
 import scala.annotation.tailrec
 import scala.collection.mutable.{ArrayBuffer, HashMap, Map}
-import scala.util.Properties
+import scala.util.{Failure, Properties, Success, Try}
 
 import com.google.common.io.ByteStreams
 import org.apache.commons.io.FileUtils
@@ -581,17 +581,15 @@ object SparkSubmit extends CommandLineUtils {
     if (clusterManager == YARN || clusterManager == LOCAL) {
       if (args.principal != null) {
         require(args.keytab != null, "Keytab must be specified when principal is specified")
-        if (!new File(args.keytab).exists()) {
-          throw new SparkException(s"Keytab file: ${args.keytab} does not exist")
-        } else {
-          // Add keytab and principal configurations in sysProps to make them available
-          // for later use; e.g. in spark sql, the isolated class loader used to talk
-          // to HiveMetastore will use these settings. They will be set as Java system
-          // properties and then loaded by SparkConf
-          sysProps.put("spark.yarn.keytab", args.keytab)
-          sysProps.put("spark.yarn.principal", args.principal)
-
-          UserGroupInformation.loginUserFromKeytab(args.principal, args.keytab)
+        Try { SparkHadoopUtil.get.loginUserFromKeytab(args.principal, args.keytab)} match {
+          case Success(_) =>
+            // Add keytab and principal configurations in sysProps to make them available
+            // for later use; e.g. in spark sql, the isolated class loader used to talk
+            // to HiveMetastore will use these settings. They will be set as Java system
+            // properties and then loaded by SparkConf
+            sysProps.put("spark.yarn.keytab", args.keytab)
+            sysProps.put("spark.yarn.principal", args.principal)
+          case Failure(exception) => throw exception
         }
       }
     }
diff --git a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala
new file mode 100644
index 000000000000..abb974833c46
--- /dev/null
+++ b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala
@@ -0,0 +1,58 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.hive.thriftserver
+
+import org.apache.hadoop.hive.cli.CliSessionState
+import org.apache.hadoop.hive.conf.HiveConf
+import org.apache.hadoop.hive.ql.session.SessionState
+
+import org.apache.spark.{SparkConf, SparkFunSuite}
+import org.apache.spark.deploy.SparkHadoopUtil
+import org.apache.spark.sql.hive.HiveUtils
+
+class HiveCliSessionStateSuite extends SparkFunSuite {
+
+  test("CliSessionState will be reused") {
+    val hiveConf = new HiveConf(classOf[SessionState])
+    val sessionState: SessionState = new CliSessionState(hiveConf)
+    SessionState.start(sessionState)
+    doTest(sessionState, true)
+  }
+
+  test("SessionState will not be reused") {
+    val hiveConf = new HiveConf(classOf[SessionState])
+    HiveUtils.newTemporaryConfiguration(useInMemoryDerby = false).foreach {
+      case (key, value) => hiveConf.set(key, value)
+    }
+    val sessionState: SessionState = new SessionState(hiveConf)
+    doTest(sessionState, false)
+
+  }
+
+  def doTest(state: SessionState, expected: Boolean): Unit = {
+
+    SessionState.start(state)
+    val s1 = SessionState.get
+
+    val sparkConf = new SparkConf()
+    val hadoopConf = SparkHadoopUtil.get.newConfiguration(sparkConf)
+    val hiveClient = HiveUtils.newClientForMetadata(sparkConf, hadoopConf)
+
+    assert((hiveClient.toString == s1.toString) === expected)
+  }
+}
diff --git a/sql/hive/pom.xml b/sql/hive/pom.xml
index 538329c21b46..f9462e79a69f 100644
--- a/sql/hive/pom.xml
+++ b/sql/hive/pom.xml
@@ -98,11 +98,6 @@
       <groupId>${hive.group}</groupId>
       <artifactId>hive-metastore</artifactId>
     </dependency>
-    <dependency>
-      <groupId>${hive.group}</groupId>
-      <artifactId>hive-cli</artifactId>
-      <scope>test</scope>
-    </dependency>
     <!--
         <dependency>
           <groupId>${hive.group}</groupId>
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
index a586baf2d5cd..aacdac0b933c 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
@@ -230,7 +230,8 @@ private[spark] object HiveUtils extends Logging {
     }.toMap
   }
 
-  def isCliSessionState(state: SessionState = SessionState.get): Boolean = {
+  def isCliSessionState(): Boolean = {
+    val state = SessionState.get
     var temp: Class[_] = if (state != null) state.getClass else null
     var found = false
     while (temp != null && !found) {
@@ -323,7 +324,7 @@ private[spark] object HiveUtils extends Logging {
         hadoopConf = hadoopConf,
         execJars = jars.toSeq,
         config = configurations,
-        isolationOn = isCliSessionState(),
+        isolationOn = !isCliSessionState(),
         barrierPrefixes = hiveMetastoreBarrierPrefixes,
         sharedPrefixes = hiveMetastoreSharedPrefixes)
     } else if (hiveMetastoreJars == "maven") {
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index 1d825a9ad808..4d68567a71da 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -35,9 +35,9 @@ import org.apache.hadoop.hive.ql.metadata.{Hive, Partition => HivePartition, Tab
 import org.apache.hadoop.hive.ql.parse.BaseSemanticAnalyzer.HIVE_COLUMN_ORDER_ASC
 import org.apache.hadoop.hive.ql.processors._
 import org.apache.hadoop.hive.ql.session.SessionState
-import org.apache.hadoop.security.UserGroupInformation
 
 import org.apache.spark.{SparkConf, SparkException}
+import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.internal.Logging
 import org.apache.spark.metrics.source.HiveCatalogMetrics
 import org.apache.spark.sql.AnalysisException
@@ -49,7 +49,6 @@ import org.apache.spark.sql.catalyst.expressions.Expression
 import org.apache.spark.sql.catalyst.parser.{CatalystSqlParser, ParseException}
 import org.apache.spark.sql.execution.QueryExecutionException
 import org.apache.spark.sql.execution.command.DDLUtils
-import org.apache.spark.sql.hive.HiveUtils
 import org.apache.spark.sql.hive.client.HiveClientImpl._
 import org.apache.spark.sql.types._
 import org.apache.spark.util.{CircularBuffer, Utils}
@@ -103,93 +102,28 @@ private[hive] class HiveClientImpl(
     case hive.v2_1 => new Shim_v2_1()
   }
 
+
   // Create an internal session state for this HiveClientImpl.
   val state: SessionState = {
     val original = Thread.currentThread().getContextClassLoader
-    // Switch to the initClassLoader.
-    Thread.currentThread().setContextClassLoader(initClassLoader)
-
-    // Set up kerberos credentials for UserGroupInformation.loginUser within
-    // current class loader
-    if (sparkConf.contains("spark.yarn.principal") && sparkConf.contains("spark.yarn.keytab")) {
-      val principalName = sparkConf.get("spark.yarn.principal")
-      val keytabFileName = sparkConf.get("spark.yarn.keytab")
-      if (!new File(keytabFileName).exists()) {
-        throw new SparkException(s"Keytab file: ${keytabFileName}" +
-          " specified in spark.yarn.keytab does not exist")
-      } else {
-        logInfo("Attempting to login to Kerberos" +
-          s" using principal: ${principalName} and keytab: ${keytabFileName}")
-        UserGroupInformation.loginUserFromKeytab(principalName, keytabFileName)
+    if (clientLoader.isolationOn) {
+      // Switch to the initClassLoader.
+      Thread.currentThread().setContextClassLoader(initClassLoader)
+      // Set up kerberos credentials for UserGroupInformation.loginUser within
+      // current class loader
+      if (sparkConf.contains("spark.yarn.principal") && sparkConf.contains("spark.yarn.keytab")) {
+        val principal = sparkConf.get("spark.yarn.principal")
+        val keytab = sparkConf.get("spark.yarn.keytab")
+        SparkHadoopUtil.get.loginUserFromKeytab(principal, keytab)
       }
-    }
-
-    val ret = try {
-      // originState will be created if not exists, will never be null
-      val originalState = SessionState.get()
-      if (HiveUtils.isCliSessionState(originalState)) {
-        // In `SparkSQLCLIDriver`, we have already started a `CliSessionState`,
-        // which contains information like configurations from command line. Later
-        // we call `SparkSQLEnv.init()` there, which would run into this part again.
-        // so we should keep `conf` and reuse the existing instance of `CliSessionState`.
-        originalState
-      } else {
-        val hiveConf = new HiveConf(classOf[SessionState])
-        // 1: we set all confs in the hadoopConf to this hiveConf.
-        // This hadoopConf contains user settings in Hadoop's core-site.xml file
-        // and Hive's hive-site.xml file. Note, we load hive-site.xml file manually in
-        // SharedState and put settings in this hadoopConf instead of relying on HiveConf
-        // to load user settings. Otherwise, HiveConf's initialize method will override
-        // settings in the hadoopConf. This issue only shows up when spark.sql.hive.metastore.jars
-        // is not set to builtin. When spark.sql.hive.metastore.jars is builtin, the classpath
-        // has hive-site.xml. So, HiveConf will use that to override its default values.
-        hadoopConf.iterator().asScala.foreach { entry =>
-          val key = entry.getKey
-          val value = entry.getValue
-          if (key.toLowerCase(Locale.ROOT).contains("password")) {
-            logDebug(s"Applying Hadoop and Hive config to Hive Conf: $key=xxx")
-          } else {
-            logDebug(s"Applying Hadoop and Hive config to Hive Conf: $key=$value")
-          }
-          hiveConf.set(key, value)
-        }
-        // HiveConf is a Hadoop Configuration, which has a field of classLoader and
-        // the initial value will be the current thread's context class loader
-        // (i.e. initClassLoader at here).
-        // We call initialConf.setClassLoader(initClassLoader) at here to make
-        // this action explicit.
-        hiveConf.setClassLoader(initClassLoader)
-        // 2: we set all spark confs to this hiveConf.
-        sparkConf.getAll.foreach { case (k, v) =>
-          if (k.toLowerCase(Locale.ROOT).contains("password")) {
-            logDebug(s"Applying Spark config to Hive Conf: $k=xxx")
-          } else {
-            logDebug(s"Applying Spark config to Hive Conf: $k=$v")
-          }
-          hiveConf.set(k, v)
-        }
-        // 3: we set all entries in config to this hiveConf.
-        extraConfig.foreach { case (k, v) =>
-          if (k.toLowerCase(Locale.ROOT).contains("password")) {
-            logDebug(s"Applying extra config to HiveConf: $k=xxx")
-          } else {
-            logDebug(s"Applying extra config to HiveConf: $k=$v")
-          }
-          hiveConf.set(k, v)
-        }
-        val state = new SessionState(hiveConf)
-        if (clientLoader.cachedHive != null) {
-          Hive.set(clientLoader.cachedHive.asInstanceOf[Hive])
-        }
-        SessionState.start(state)
-        state.out = new PrintStream(outputBuffer, true, "UTF-8")
-        state.err = new PrintStream(outputBuffer, true, "UTF-8")
-        state
+      try {
+        newState()
+      } finally {
+        Thread.currentThread().setContextClassLoader(original)
       }
-    } finally {
-      Thread.currentThread().setContextClassLoader(original)
+    } else {
+      Option(SessionState.get()).getOrElse(newState())
     }
-    ret
   }
 
   // Log the default warehouse location.
@@ -197,6 +131,44 @@ private[hive] class HiveClientImpl(
     s"Warehouse location for Hive client " +
       s"(version ${version.fullVersion}) is ${conf.get("hive.metastore.warehouse.dir")}")
 
+  private def newState(): SessionState = {
+    val hiveConf = new HiveConf(classOf[SessionState])
+    // HiveConf is a Hadoop Configuration, which has a field of classLoader and
+    // the initial value will be the current thread's context class loader
+    // (i.e. initClassLoader at here).
+    // We call initialConf.setClassLoader(initClassLoader) at here to make
+    // this action explicit.
+    hiveConf.setClassLoader(initClassLoader)
+
+    // 1: Take all from the hadoopConf to this hiveConf.
+    // This hadoopConf contains user settings in Hadoop's core-site.xml file
+    // and Hive's hive-site.xml file. Note, we load hive-site.xml file manually in
+    // SharedState and put settings in this hadoopConf instead of relying on HiveConf
+    // to load user settings. Otherwise, HiveConf's initialize method will override
+    // settings in the hadoopConf. This issue only shows up when spark.sql.hive.metastore.jars
+    // is not set to builtin. When spark.sql.hive.metastore.jars is builtin, the classpath
+    // has hive-site.xml. So, HiveConf will use that to override its default values.
+    // 2: we set all spark confs to this hiveConf.
+    // 3: we set all entries in config to this hiveConf.
+    (hadoopConf.iterator().asScala.map(kv => kv.getKey -> kv.getValue)
+      ++ sparkConf.getAll.toMap ++ extraConfig).foreach { case (k, v) =>
+      if (k.toLowerCase(Locale.ROOT).contains("password")) {
+        logDebug(s"Applying Spark config to Hive Conf: $k=xxx")
+      } else {
+        logDebug(s"Applying Spark config to Hive Conf: $k=$v")
+      }
+      hiveConf.set(k, v)
+    }
+    val state = new SessionState(hiveConf)
+    if (clientLoader.cachedHive != null) {
+      Hive.set(clientLoader.cachedHive.asInstanceOf[Hive])
+    }
+    SessionState.start(state)
+    state.out = new PrintStream(outputBuffer, true, "UTF-8")
+    state.err = new PrintStream(outputBuffer, true, "UTF-8")
+    state
+  }
+
   /** Returns the configuration for the current session. */
   def conf: HiveConf = state.getConf
 
@@ -260,6 +232,8 @@ private[hive] class HiveClientImpl(
     }
   }
 
+  override def toString: String = state.toString
+
   /**
    * Runs `f` with ThreadLocal session state and classloaders configured for this version of hive.
    */
diff --git a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveUtilsSuite.scala b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveUtilsSuite.scala
index 586e299321c1..667a7ddd8bb6 100644
--- a/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveUtilsSuite.scala
+++ b/sql/hive/src/test/scala/org/apache/spark/sql/hive/HiveUtilsSuite.scala
@@ -17,13 +17,8 @@
 
 package org.apache.spark.sql.hive
 
-import org.apache.hadoop.hive.cli.CliSessionState
-import org.apache.hadoop.hive.conf.HiveConf
 import org.apache.hadoop.hive.conf.HiveConf.ConfVars
-import org.apache.hadoop.hive.ql.session.SessionState
 
-import org.apache.spark.SparkConf
-import org.apache.spark.deploy.SparkHadoopUtil
 import org.apache.spark.sql.hive.test.TestHiveSingleton
 import org.apache.spark.sql.test.SQLTestUtils
 import org.apache.spark.sql.QueryTest
@@ -38,47 +33,4 @@ class HiveUtilsSuite extends QueryTest with SQLTestUtils with TestHiveSingleton
       assert(conf(ConfVars.METASTORE_END_FUNCTION_LISTENERS.varname) === "")
     }
   }
-
-  test("CliSessionState will be reused") {
-    val hiveConf = new HiveConf(classOf[SessionState])
-    val sessionState: SessionState = new CliSessionState(hiveConf)
-    SessionState.start(sessionState)
-    val s1 = SessionState.get()
-    assert(s1.isInstanceOf[CliSessionState])
-
-    val sparkConf = new SparkConf()
-    val hadoopConf = SparkHadoopUtil.get.newConfiguration(sparkConf)
-    val metaHive = HiveUtils.newClientForMetadata(sparkConf, hadoopConf)
-    assert(metaHive.getConf("spark.sql.hive.jars", "builtin") === "builtin")
-
-    val s2 = SessionState.get()
-    assert(s2.isInstanceOf[CliSessionState])
-    assert(s1 === s2, "CliSessionState should be reused")
-  }
-
-  test("SessionState will not be reused") {
-    val hiveConf = new HiveConf(classOf[SessionState])
-    val sessionState: SessionState = new SessionState(hiveConf)
-    SessionState.start(sessionState)
-
-    val s1 = SessionState.get()
-    assert(!s1.isInstanceOf[CliSessionState])
-
-    val sparkConf = new SparkConf().set("spark.sql.hive.jars", "builtin")
-    val hadoopConf = SparkHadoopUtil.get.newConfiguration(sparkConf)
-    val metaHive = HiveUtils.newClientForMetadata(sparkConf, hadoopConf)
-
-    assert(metaHive.getConf("spark.sql.hive.jars", "builtin") === "builtin")
-    val s2 = SessionState.get()
-    assert(s1 !== s2)
-
-    val sparkConf2 = sparkConf.set("spark.sql.hive.jars", "maven")
-    val hadoopConf2 = SparkHadoopUtil.get.newConfiguration(sparkConf2)
-    val metaHive2 = HiveUtils.newClientForMetadata(sparkConf2, hadoopConf2)
-
-    assert(metaHive2.getConf("spark.sql.hive.jars", "builtin") === "maven")
-    val s3 = SessionState.get()
-    assert(s1 !== s3)
-    assert(s2 !== s3)
-  }
 }

From 341964e1c9fe4f22c5aeb9a9e17f0d03a6b2bbf2 Mon Sep 17 00:00:00 2001
From: hzyaoqin <hzyaoqin@corp.netease.com>
Date: Tue, 8 Aug 2017 11:10:30 +0800
Subject: [PATCH 5/9] fix ut reduplicated derby metastore_db

---
 .../sql/hive/thriftserver/HiveCliSessionStateSuite.scala | 9 ++++-----
 1 file changed, 4 insertions(+), 5 deletions(-)

diff --git a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala
index abb974833c46..83164bf453c2 100644
--- a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala
+++ b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala
@@ -29,8 +29,10 @@ class HiveCliSessionStateSuite extends SparkFunSuite {
 
   test("CliSessionState will be reused") {
     val hiveConf = new HiveConf(classOf[SessionState])
+    HiveUtils.newTemporaryConfiguration(useInMemoryDerby = false).foreach {
+      case (key, value) => hiveConf.set(key, value)
+    }
     val sessionState: SessionState = new CliSessionState(hiveConf)
-    SessionState.start(sessionState)
     doTest(sessionState, true)
   }
 
@@ -41,18 +43,15 @@ class HiveCliSessionStateSuite extends SparkFunSuite {
     }
     val sessionState: SessionState = new SessionState(hiveConf)
     doTest(sessionState, false)
-
   }
 
   def doTest(state: SessionState, expected: Boolean): Unit = {
-
     SessionState.start(state)
     val s1 = SessionState.get
-
     val sparkConf = new SparkConf()
     val hadoopConf = SparkHadoopUtil.get.newConfiguration(sparkConf)
     val hiveClient = HiveUtils.newClientForMetadata(sparkConf, hadoopConf)
-
     assert((hiveClient.toString == s1.toString) === expected)
+    SessionState.get().close()
   }
 }

From 6c0bf709f95dec3ee3bce3e905e51f31fc5b0e64 Mon Sep 17 00:00:00 2001
From: Kent Yao <yaooqinn@hotmail.com>
Date: Tue, 8 Aug 2017 16:27:46 +0800
Subject: [PATCH 6/9] review change

---
 .../org/apache/spark/deploy/SparkSubmit.scala | 17 +++++-------
 .../HiveCliSessionStateSuite.scala            | 27 +++++++++----------
 .../spark/sql/hive/client/HiveClient.scala    |  6 +++++
 .../sql/hive/client/HiveClientImpl.scala      |  4 +--
 4 files changed, 28 insertions(+), 26 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
index 765283d02319..2a5364ab8a3d 100644
--- a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
@@ -581,16 +581,13 @@ object SparkSubmit extends CommandLineUtils {
     if (clusterManager == YARN || clusterManager == LOCAL) {
       if (args.principal != null) {
         require(args.keytab != null, "Keytab must be specified when principal is specified")
-        Try { SparkHadoopUtil.get.loginUserFromKeytab(args.principal, args.keytab)} match {
-          case Success(_) =>
-            // Add keytab and principal configurations in sysProps to make them available
-            // for later use; e.g. in spark sql, the isolated class loader used to talk
-            // to HiveMetastore will use these settings. They will be set as Java system
-            // properties and then loaded by SparkConf
-            sysProps.put("spark.yarn.keytab", args.keytab)
-            sysProps.put("spark.yarn.principal", args.principal)
-          case Failure(exception) => throw exception
-        }
+        SparkHadoopUtil.get.loginUserFromKeytab(args.principal, args.keytab)
+        // Add keytab and principal configurations in sysProps to make them available
+        // for later use; e.g. in spark sql, the isolated class loader used to talk
+        // to HiveMetastore will use these settings. They will be set as Java system
+        // properties and then loaded by SparkConf
+        sysProps.put("spark.yarn.keytab", args.keytab)
+        sysProps.put("spark.yarn.principal", args.principal)
       }
     }
 
diff --git a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala
index 83164bf453c2..c24d85576a6d 100644
--- a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala
+++ b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala
@@ -33,25 +33,24 @@ class HiveCliSessionStateSuite extends SparkFunSuite {
       case (key, value) => hiveConf.set(key, value)
     }
     val sessionState: SessionState = new CliSessionState(hiveConf)
-    doTest(sessionState, true)
+    SessionState.start(sessionState)
+    val s1 = SessionState.get
+    val sparkConf = new SparkConf()
+    val hadoopConf = SparkHadoopUtil.get.newConfiguration(sparkConf)
+    val s2 = HiveUtils.newClientForMetadata(sparkConf, hadoopConf).getState
+    assert(s1 === s2)
+    assert(s2.isInstanceOf[CliSessionState])
   }
 
   test("SessionState will not be reused") {
-    val hiveConf = new HiveConf(classOf[SessionState])
-    HiveUtils.newTemporaryConfiguration(useInMemoryDerby = false).foreach {
-      case (key, value) => hiveConf.set(key, value)
-    }
-    val sessionState: SessionState = new SessionState(hiveConf)
-    doTest(sessionState, false)
-  }
-
-  def doTest(state: SessionState, expected: Boolean): Unit = {
-    SessionState.start(state)
-    val s1 = SessionState.get
     val sparkConf = new SparkConf()
     val hadoopConf = SparkHadoopUtil.get.newConfiguration(sparkConf)
+    HiveUtils.newTemporaryConfiguration(useInMemoryDerby = false).foreach {
+      case (key, value) => hadoopConf.set(key, value)
+    }
     val hiveClient = HiveUtils.newClientForMetadata(sparkConf, hadoopConf)
-    assert((hiveClient.toString == s1.toString) === expected)
-    SessionState.get().close()
+    val s1 = hiveClient.getState
+    val s2 = hiveClient.newSession().getState
+    assert(s1 !== s2)
   }
 }
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClient.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClient.scala
index 16a80f9fff45..8cff0ca0963b 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClient.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClient.scala
@@ -38,6 +38,12 @@ private[hive] trait HiveClient {
   /** Returns the configuration for the given key in the current session. */
   def getConf(key: String, defaultValue: String): String
 
+  /**
+   * Return the associated Hive SessionState of this [[HiveClientImpl]]
+   * @return [[Any]] not SessionState to avoid linkage error
+   */
+  def getState: Any
+
   /**
    * Runs a HiveQL command using Hive, returning the results as a list of strings.  Each row will
    * result in one string.
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index 4d68567a71da..1093c4104809 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -102,7 +102,6 @@ private[hive] class HiveClientImpl(
     case hive.v2_1 => new Shim_v2_1()
   }
 
-
   // Create an internal session state for this HiveClientImpl.
   val state: SessionState = {
     val original = Thread.currentThread().getContextClassLoader
@@ -232,7 +231,8 @@ private[hive] class HiveClientImpl(
     }
   }
 
-  override def toString: String = state.toString
+  /** Return the associated Hive [[SessionState]] of this [[HiveClientImpl]] */
+  override def getState: SessionState = withHiveState(state)
 
   /**
    * Runs `f` with ThreadLocal session state and classloaders configured for this version of hive.

From 51fac11abf4726e2a67426e00104886f1072525c Mon Sep 17 00:00:00 2001
From: Kent Yao <yaooqinn@hotmail.com>
Date: Wed, 9 Aug 2017 16:10:05 +0800
Subject: [PATCH 7/9] ut affect each other

---
 .../HiveCliSessionStateSuite.scala            | 46 +++++++++++--------
 1 file changed, 27 insertions(+), 19 deletions(-)

diff --git a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala
index c24d85576a6d..5f9ea4d26790 100644
--- a/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala
+++ b/sql/hive-thriftserver/src/test/scala/org/apache/spark/sql/hive/thriftserver/HiveCliSessionStateSuite.scala
@@ -27,30 +27,38 @@ import org.apache.spark.sql.hive.HiveUtils
 
 class HiveCliSessionStateSuite extends SparkFunSuite {
 
+  def withSessionClear(f: () => Unit): Unit = {
+    try f finally SessionState.detachSession()
+  }
+
   test("CliSessionState will be reused") {
-    val hiveConf = new HiveConf(classOf[SessionState])
-    HiveUtils.newTemporaryConfiguration(useInMemoryDerby = false).foreach {
-      case (key, value) => hiveConf.set(key, value)
+    withSessionClear { () =>
+      val hiveConf = new HiveConf(classOf[SessionState])
+      HiveUtils.newTemporaryConfiguration(useInMemoryDerby = false).foreach {
+        case (key, value) => hiveConf.set(key, value)
+      }
+      val sessionState: SessionState = new CliSessionState(hiveConf)
+      SessionState.start(sessionState)
+      val s1 = SessionState.get
+      val sparkConf = new SparkConf()
+      val hadoopConf = SparkHadoopUtil.get.newConfiguration(sparkConf)
+      val s2 = HiveUtils.newClientForMetadata(sparkConf, hadoopConf).getState
+      assert(s1 === s2)
+      assert(s2.isInstanceOf[CliSessionState])
     }
-    val sessionState: SessionState = new CliSessionState(hiveConf)
-    SessionState.start(sessionState)
-    val s1 = SessionState.get
-    val sparkConf = new SparkConf()
-    val hadoopConf = SparkHadoopUtil.get.newConfiguration(sparkConf)
-    val s2 = HiveUtils.newClientForMetadata(sparkConf, hadoopConf).getState
-    assert(s1 === s2)
-    assert(s2.isInstanceOf[CliSessionState])
   }
 
   test("SessionState will not be reused") {
-    val sparkConf = new SparkConf()
-    val hadoopConf = SparkHadoopUtil.get.newConfiguration(sparkConf)
-    HiveUtils.newTemporaryConfiguration(useInMemoryDerby = false).foreach {
-      case (key, value) => hadoopConf.set(key, value)
+    withSessionClear { () =>
+      val sparkConf = new SparkConf()
+      val hadoopConf = SparkHadoopUtil.get.newConfiguration(sparkConf)
+      HiveUtils.newTemporaryConfiguration(useInMemoryDerby = false).foreach {
+        case (key, value) => hadoopConf.set(key, value)
+      }
+      val hiveClient = HiveUtils.newClientForMetadata(sparkConf, hadoopConf)
+      val s1 = hiveClient.getState
+      val s2 = hiveClient.newSession().getState
+      assert(s1 !== s2)
     }
-    val hiveClient = HiveUtils.newClientForMetadata(sparkConf, hadoopConf)
-    val s1 = hiveClient.getState
-    val s2 = hiveClient.newSession().getState
-    assert(s1 !== s2)
   }
 }

From c50a32f0476d27ad7aed6e972ba9b6ffb5263aea Mon Sep 17 00:00:00 2001
From: Kent Yao <yaooqinn@hotmail.com>
Date: Wed, 9 Aug 2017 23:45:05 +0800
Subject: [PATCH 8/9] thanks for reviewing

---
 .../apache/spark/deploy/SparkHadoopUtil.scala |  3 +--
 .../org/apache/spark/deploy/SparkSubmit.scala |  2 +-
 .../org/apache/spark/sql/hive/HiveUtils.scala |  5 ++++
 .../sql/hive/client/HiveClientImpl.scala      | 24 ++++++++++++-------
 4 files changed, 23 insertions(+), 11 deletions(-)

diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala b/core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala
index d623cd9c4a5f..15832e6cf4ec 100644
--- a/core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/SparkHadoopUtil.scala
@@ -138,8 +138,7 @@ class SparkHadoopUtil extends Logging {
 
   def loginUserFromKeytab(principalName: String, keytabFilename: String): Unit = {
     if (!new File(keytabFilename).exists()) {
-      throw new SparkException(s"Keytab file: ${keytabFilename}" +
-        " specified in spark.yarn.keytab does not exist")
+      throw new SparkException(s"Keytab file: ${keytabFilename} does not exist")
     } else {
       logInfo("Attempting to login to Kerberos" +
         s" using principal: ${principalName} and keytab: ${keytabFilename}")
diff --git a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
index 2a5364ab8a3d..809aebbc5008 100644
--- a/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
+++ b/core/src/main/scala/org/apache/spark/deploy/SparkSubmit.scala
@@ -28,7 +28,7 @@ import javax.net.ssl._
 
 import scala.annotation.tailrec
 import scala.collection.mutable.{ArrayBuffer, HashMap, Map}
-import scala.util.{Failure, Properties, Success, Try}
+import scala.util.Properties
 
 import com.google.common.io.ByteStreams
 import org.apache.commons.io.FileUtils
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
index aacdac0b933c..8efad394d949 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/HiveUtils.scala
@@ -230,6 +230,11 @@ private[spark] object HiveUtils extends Logging {
     }.toMap
   }
 
+  /**
+   * Check current Thread's SessionState type
+   * @return true when SessionState.get returns an instance of CliSessionState,
+   *         false when it gets non-CliSessionState instance or null
+   */
   def isCliSessionState(): Boolean = {
     val state = SessionState.get
     var temp: Class[_] = if (state != null) state.getClass else null
diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index 1093c4104809..bdb864e59fc7 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -108,8 +108,7 @@ private[hive] class HiveClientImpl(
     if (clientLoader.isolationOn) {
       // Switch to the initClassLoader.
       Thread.currentThread().setContextClassLoader(initClassLoader)
-      // Set up kerberos credentials for UserGroupInformation.loginUser within
-      // current class loader
+      // Set up kerberos credentials for UserGroupInformation.loginUser within current class loader
       if (sparkConf.contains("spark.yarn.principal") && sparkConf.contains("spark.yarn.keytab")) {
         val principal = sparkConf.get("spark.yarn.principal")
         val keytab = sparkConf.get("spark.yarn.keytab")
@@ -121,7 +120,16 @@ private[hive] class HiveClientImpl(
         Thread.currentThread().setContextClassLoader(original)
       }
     } else {
-      Option(SessionState.get()).getOrElse(newState())
+      // Isolation off means we detect a CliSessionState instance in current thread. 1: Inside the
+      // spark project, we have already started a CliSessionState  in `SparkSQLCLIDriver`, which
+      // contains information like configurations from command lines. Later, we call
+      // `SparkSQLEnv.init()` there, which would new a hive client again into this part again. so we
+      // should keep those configurations and reuse the existing instance of `CliSessionState`. In
+      // this case, SessionState.get will always return a CliSessionState.
+      // In another case, a user app may start a CliSessionState outside spark project with built in
+      // hive jars, which will turn off isolation, after that, SessionSate.detachSession is called
+      // to remove the current state, hive client created later will init its state by newState()
+      Option(SessionState.get).getOrElse(newState())
     }
   }
 
@@ -151,11 +159,11 @@ private[hive] class HiveClientImpl(
     // 3: we set all entries in config to this hiveConf.
     (hadoopConf.iterator().asScala.map(kv => kv.getKey -> kv.getValue)
       ++ sparkConf.getAll.toMap ++ extraConfig).foreach { case (k, v) =>
-      if (k.toLowerCase(Locale.ROOT).contains("password")) {
-        logDebug(s"Applying Spark config to Hive Conf: $k=xxx")
-      } else {
-        logDebug(s"Applying Spark config to Hive Conf: $k=$v")
-      }
+      logDebug(
+        s"""
+           |Applying Hadoop/Hive/Spark and extra properties to Hive Conf:
+           |$k=${if (k.toLowerCase(Locale.ROOT).contains("password")) "xxx" else v}
+         """.stripMargin)
       hiveConf.set(k, v)
     }
     val state = new SessionState(hiveConf)

From c6ed2d7d0c5a8f3dc1e09f0a7d9991ede4a60a76 Mon Sep 17 00:00:00 2001
From: Kent Yao <yaooqinn@hotmail.com>
Date: Wed, 9 Aug 2017 23:55:17 +0800
Subject: [PATCH 9/9] typo

---
 .../sql/hive/client/HiveClientImpl.scala      | 19 ++++++++++---------
 1 file changed, 10 insertions(+), 9 deletions(-)

diff --git a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
index bdb864e59fc7..27f60cf3d391 100644
--- a/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
+++ b/sql/hive/src/main/scala/org/apache/spark/sql/hive/client/HiveClientImpl.scala
@@ -120,15 +120,16 @@ private[hive] class HiveClientImpl(
         Thread.currentThread().setContextClassLoader(original)
       }
     } else {
-      // Isolation off means we detect a CliSessionState instance in current thread. 1: Inside the
-      // spark project, we have already started a CliSessionState  in `SparkSQLCLIDriver`, which
-      // contains information like configurations from command lines. Later, we call
-      // `SparkSQLEnv.init()` there, which would new a hive client again into this part again. so we
-      // should keep those configurations and reuse the existing instance of `CliSessionState`. In
-      // this case, SessionState.get will always return a CliSessionState.
-      // In another case, a user app may start a CliSessionState outside spark project with built in
-      // hive jars, which will turn off isolation, after that, SessionSate.detachSession is called
-      // to remove the current state, hive client created later will init its state by newState()
+      // Isolation off means we detect a CliSessionState instance in current thread.
+      // 1: Inside the spark project, we have already started a CliSessionState in
+      // `SparkSQLCLIDriver`, which contains configurations from command lines. Later, we call
+      // `SparkSQLEnv.init()` there, which would new a hive client again. so we should keep those
+      // configurations and reuse the existing instance of `CliSessionState`. In this case,
+      // SessionState.get will always return a CliSessionState.
+      // 2: In another case, a user app may start a CliSessionState outside spark project with built
+      // in hive jars, which will turn off isolation, if SessionSate.detachSession is
+      // called to remove the current state after that, hive client created later will initialize
+      // its own state by newState()
       Option(SessionState.get).getOrElse(newState())
     }
   }