[KYUUBI #357] [Kyuubi-1.0] Update built-in Spark version to 3.0.2

![ulysses-you](https://badgen.net/badge/Hello/ulysses-you/green) [![Closes #357](https://badgen.net/badge/Preview/Closes%20%23357/blue)](https://github.com/yaooqinn/kyuubi/pull/357) ![12](https://badgen.net/badge/%2B/12/red) ![3](https://badgen.net/badge/-/3/green) ![2](https://badgen.net/badge/commits/2/yellow) ![Target Issue](https://badgen.net/badge/Missing/Target%20Issue/ff0000) ![Test Plan](https://badgen.net/badge/Missing/Test%20Plan/ff0000) [&#10088;?&#10089;](https://pullrequestbadge.com/?utm_medium=github&utm_source=yaooqinn&utm_campaign=badge_info)<!-- PR-BADGE: PLEASE DO NOT REMOVE THIS COMMENT -->

backport [#356](https://github.com/yaooqinn/kyuubi/pull/356) for branch-1.0.
<!--
Thanks for sending a pull request!

Here are some tips for you:
  1. If this is your first time, please read our contributor guidelines: https://kyuubi.readthedocs.io/en/latest/community/contributions.html
  2. If the PR is related to an issue in https://github.com/yaooqinn/kyuubi/issues, add '[KYUUBI #XXXX]' in your PR title, e.g., '[KYUUBI #XXXX] Your PR title ...'.
  3. If the PR is unfinished, add '[WIP]' in your PR title, e.g., '[WIP][KYUUBI #XXXX] Your PR title ...'.
-->

### _Why are the changes needed?_
<!--
Please clarify why the changes are needed. For instance,
  1. If you add a feature, you can talk about the use case of it.
  2. If you fix a bug, you can clarify why it is a bug.
-->
Spark 3.0.2 has been released now.

### _How was this patch tested?_
- [ ] Add some test cases that check the changes thoroughly including negative and positive cases if possible

- [ ] Add screenshots for manual tests if appropriate

- [ ] [Run test](https://kyuubi.readthedocs.io/en/latest/tools/testing.html#running-tests) locally before make a pull request

Closes #357 from ulysses-you/branch-1.0-spark-3.0.2.

b5d4442 [ulysses-you] sync
2c5ee8e [ulysses-you] 3.0.2

Authored-by: ulysses-you <ulyssesyou18@gmail.com>
Signed-off-by: Kent Yao <yao@apache.org>

bin/load-kyuubi-env.sh

@@ -66,7 +66,16 @@ fi
 SPARK_BUILTIN="${KYUUBI_HOME}/externals/spark-$SPARK_VERSION_BUILD-bin-hadoop${HADOOP_VERSION_BUILD:0:3}$HIVE_VERSION_SUFFIX"
 
 if [[ ! -d ${SPARK_BUILTIN} ]]; then
-  SPARK_BUILTIN="${KYUUBI_HOME}/externals/kyuubi-download/target/spark-3.0.1-bin-hadoop2.7"
+  MVN="${MVN:-"${KYUUBI_HOME}/build/mvn"}"
+  SPARK_VERSION_BUILD=$("$MVN" help:evaluate -Dexpression=spark.version 2>/dev/null\
+    | grep -v "INFO"\
+    | grep -v "WARNING"\
+    | tail -n 1)
+  HADOOP_VERSION_BUILD=$("$MVN" help:evaluate -Dexpression=hadoop.version 2>/dev/null\
+    | grep -v "INFO"\
+    | grep -v "WARNING"\
+    | tail -n 1)
+  SPARK_BUILTIN="${KYUUBI_HOME}/externals/kyuubi-download/target/spark-$SPARK_VERSION_BUILD-bin-hadoop${HADOOP_VERSION_BUILD:0:3}"
 fi
 
 export SPARK_HOME="${SPARK_HOME:-"${SPARK_BUILTIN}"}"

dev/kyuubi-codecov/pom.xml

@@ -22,7 +22,7 @@
     <parent>
         <artifactId>kyuubi</artifactId>
         <groupId>org.apache.kyuubi</groupId>
-        <version>1.0.0-SNAPSHOT</version>
+        <version>1.0.4-SNAPSHOT</version>
         <relativePath>../../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

dev/kyuubi-tpcds/pom.xml

@@ -22,7 +22,7 @@
     <parent>
         <artifactId>kyuubi</artifactId>
         <groupId>org.apache.kyuubi</groupId>
-        <version>1.0.0-SNAPSHOT</version>
+        <version>1.0.4-SNAPSHOT</version>
         <relativePath>../../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

docs/conf.py

@@ -54,7 +54,7 @@ limitations under the License.
 author = 'Kent Yao'
 
 # The full version, including alpha/beta/rc tags
-release = '1.0.0'
+release = '1.0.4'
 
 
 # -- General configuration ---------------------------------------------------

docs/tools/distribution.md

@@ -28,4 +28,4 @@ For instance,
 ./build/dist --name custom-name --tgz
 ```
 
-This results a Kyuubi distribution named `kyuubi-{version}-bin-custom-name-spark-3.0.1.tar.gz` for you. 
+This results a Kyuubi distribution named `kyuubi-{version}-bin-custom-name.tar.gz` for you.

externals/kyuubi-download/pom.xml

@@ -22,7 +22,7 @@
     <parent>
         <artifactId>kyuubi</artifactId>
         <groupId>org.apache.kyuubi</groupId>
-        <version>1.0.0-SNAPSHOT</version>
+        <version>1.0.4-SNAPSHOT</version>
         <relativePath>../../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

externals/kyuubi-spark-sql-engine/pom.xml

@@ -22,7 +22,7 @@
     <parent>
         <artifactId>kyuubi</artifactId>
         <groupId>org.apache.kyuubi</groupId>
-        <version>1.0.0-SNAPSHOT</version>
+        <version>1.0.4-SNAPSHOT</version>
         <relativePath>../../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngine.scala

@@ -40,6 +40,8 @@ private[spark] final class SparkSQLEngine(name: String, spark: SparkSession)
   private val discoveryService = new ServiceDiscovery(this)
 
   override def initialize(conf: KyuubiConf): Unit = {
+    val listener = new SparkSQLEngineListener(this)
+    spark.sparkContext.addSparkListener(listener)
     super.initialize(conf)
     if (ServiceDiscovery.supportServiceDiscovery(conf)) {
       addService(discoveryService)
@@ -49,7 +51,6 @@ private[spark] final class SparkSQLEngine(name: String, spark: SparkSession)
 
   override protected def stopServer(): Unit = {
     countDownLatch.countDown()
-    spark.stop()
   }
 }
 
@@ -112,9 +113,11 @@ object SparkSQLEngine extends Logging {
         error("Error start SparkSQLEngine", t)
         if (engine != null) {
           engine.stop()
-        } else if (spark != null) {
-          spark.stop()
         }
+    } finally {
+      if (spark != null) {
+        spark.stop()
+      }
     }
   }
 }

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngineListener.scala

@@ -0,0 +1,37 @@
+/*
+ * 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.kyuubi.engine.spark
+
+import org.apache.spark.scheduler.{SparkListener, SparkListenerApplicationEnd}
+
+import org.apache.kyuubi.Logging
+import org.apache.kyuubi.service.{Serverable, ServiceState}
+
+class SparkSQLEngineListener(server: Serverable) extends SparkListener with Logging {
+
+  override def onApplicationEnd(event: SparkListenerApplicationEnd): Unit = {
+    server.getServiceState match {
+      case ServiceState.STOPPED => debug("Received ApplicationEnd Message form Spark after the" +
+        " engine has stopped")
+      case state =>
+        info(s"Received ApplicationEnd Message from Spark at $state, stopping")
+        server.stop()
+    }
+  }
+
+}

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/operation/SparkOperation.scala

@@ -85,7 +85,9 @@ abstract class SparkOperation(spark: SparkSession, opType: OperationType, sessio
   }
 
   protected def onError(cancel: Boolean = false): PartialFunction[Throwable, Unit] = {
-    case e: Exception =>
+    // We should use Throwable instead of Exception since `java.lang.NoClassDefFoundError`
+    // could be thrown.
+    case e: Throwable =>
       if (cancel) spark.sparkContext.cancelJobGroup(statementId)
       state.synchronized {
         val errMsg = KyuubiSQLException.stringifyException(e)

externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/session/SparkSQLSessionManager.scala

@@ -71,7 +71,7 @@ class SparkSQLSessionManager private (name: String, spark: SparkSession)
     } catch {
       case e: Exception =>
         sessionImpl.close()
-        throw KyuubiSQLException(s"Error opening session $handle for $user: ${e.getMessage}", e)
+        throw KyuubiSQLException(e)
     }
   }
 

externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/SparkSQLEngineListenerSuite.scala

@@ -0,0 +1,45 @@
+/*
+ * 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.kyuubi.engine.spark
+
+import org.apache.spark.sql.SparkSession
+
+import org.apache.kyuubi.KyuubiFunSuite
+import org.apache.kyuubi.config.KyuubiConf
+import org.apache.kyuubi.service.ServiceState
+
+class SparkSQLEngineListenerSuite extends KyuubiFunSuite {
+
+  override def beforeAll(): Unit = {
+    SparkSession.clearActiveSession()
+    SparkSession.clearDefaultSession()
+    super.beforeAll()
+  }
+
+  test("application end") {
+    val spark = SparkSession
+      .builder().master("local").config("spark.ui.port", "0").getOrCreate()
+
+    val engine = new SparkSQLEngine(spark)
+    engine.initialize(KyuubiConf())
+    engine.start()
+    assert(engine.getServiceState === ServiceState.STARTED)
+    spark.stop()
+    assert(engine.getServiceState === ServiceState.STOPPED)
+  }
+}

externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/WithSparkSQLEngine.scala

@@ -39,6 +39,8 @@ trait WithSparkSQLEngine extends JDBCTests {
   System.setProperty("spark.sql.warehouse.dir", warehousePath.toString)
   System.setProperty("spark.sql.hive.metastore.sharedPrefixes", "org.apache.hive.jdbc")
 
+  SparkSession.clearActiveSession()
+  SparkSession.clearDefaultSession()
   protected val spark: SparkSession = SparkSQLEngine.createSpark()
 
   protected var engine: SparkSQLEngine = _
@@ -56,6 +58,7 @@ trait WithSparkSQLEngine extends JDBCTests {
     if (engine != null) {
       engine.stop()
     }
+    spark.stop()
     SessionState.detachSession()
     Hive.closeCurrent()
   }

kyuubi-assembly/pom.xml

@@ -24,7 +24,7 @@
     <parent>
         <artifactId>kyuubi</artifactId>
         <groupId>org.apache.kyuubi</groupId>
-        <version>1.0.0-SNAPSHOT</version>
+        <version>1.0.4-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
 

kyuubi-common/pom.xml

@@ -22,7 +22,7 @@
     <parent>
         <groupId>org.apache.kyuubi</groupId>
         <artifactId>kyuubi</artifactId>
-        <version>1.0.0-SNAPSHOT</version>
+        <version>1.0.4-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

kyuubi-common/src/main/scala/org/apache/kyuubi/KyuubiSQLException.scala

@@ -18,8 +18,10 @@
 package org.apache.kyuubi
 
 import java.io.{PrintWriter, StringWriter}
+import java.lang.reflect.{InvocationTargetException, UndeclaredThrowableException}
 import java.sql.SQLException
 
+import scala.annotation.tailrec
 import scala.collection.JavaConverters._
 
 import org.apache.hive.service.rpc.thrift.{TStatus, TStatusCode}
@@ -46,10 +48,11 @@ object KyuubiSQLException {
   private final val SEPARATOR: Char = ':'
 
   def apply(msg: String, throwable: Throwable): KyuubiSQLException = {
-    new KyuubiSQLException(msg, throwable)
+    new KyuubiSQLException(msg, findCause(throwable))
   }
   def apply(cause: Throwable): KyuubiSQLException = {
-    new KyuubiSQLException(cause.getMessage, cause)
+    val theCause = findCause(cause)
+    new KyuubiSQLException(theCause.getMessage, theCause)
   }
 
   def apply(msg: String): KyuubiSQLException = new KyuubiSQLException(msg, null)
@@ -157,4 +160,10 @@ object KyuubiSQLException {
     stm.toString
   }
 
+  @tailrec
+  def findCause(t: Throwable): Throwable = t match {
+    case e @ (_: UndeclaredThrowableException | _: InvocationTargetException)
+      if e.getCause != null => findCause(e.getCause)
+    case e => e
+  }
 }

kyuubi-common/src/main/scala/org/apache/kyuubi/service/FrontendService.scala

@@ -99,7 +99,8 @@ class FrontendService private (name: String, be: BackendService, oomHook: Runnab
         s" [$minThreads, $maxThreads] worker threads")
     } catch {
       case e: Throwable =>
-        throw new KyuubiException("Failed to initialize frontend service", e)
+        throw new KyuubiException(
+          s"Failed to initialize frontend service on $serverAddr:$portNum.", e)
     }
     super.initialize(conf)
   }

kyuubi-common/src/main/scala/org/apache/kyuubi/service/Serverable.scala

@@ -47,14 +47,19 @@ abstract class Serverable(name: String) extends CompositeService(name) {
 
   override def stop(): Unit = synchronized {
     try {
-      stopServer()
-    } catch {
-      case t: Throwable =>
-        warn(s"Error stopping spark ${t.getMessage}", t)
-    } finally {
       if (started.getAndSet(false)) {
         super.stop()
       }
+    } catch {
+      case t: Throwable =>
+        warn(s"Error stopping $name ${t.getMessage}", t)
+    } finally {
+      try {
+        stopServer()
+      } catch {
+        case t: Throwable =>
+          warn(s"Error stopping spark ${t.getMessage}", t)
+      }
     }
   }
 }

kyuubi-common/src/main/scala/org/apache/kyuubi/session/AbstractSession.scala

@@ -61,10 +61,11 @@ abstract class AbstractSession(
   private def release(userAccess: Boolean): Unit = {
     if (userAccess) {
       _lastAccessTime = System.currentTimeMillis
+      if (opHandleSet.isEmpty) {
+        _lastIdleTime = System.currentTimeMillis
+      }
     }
-    if (opHandleSet.isEmpty) {
-      _lastIdleTime = System.currentTimeMillis
-    } else {
+    if (!opHandleSet.isEmpty) {
       _lastIdleTime = 0
     }
   }
@@ -211,7 +212,7 @@ abstract class AbstractSession(
     }
   }
 
-  override def closeExpiredOperations: Unit = withAcquireRelease() {
+  override def closeExpiredOperations: Unit = withAcquireRelease(false) {
     val operations = sessionManager.operationManager
       .removeExpiredOperations(opHandleSet.asScala.toSeq)
     operations.foreach { op =>

kyuubi-common/src/test/scala/org/apache/kyuubi/KyuubiSQLExceptionSuite.scala

@@ -17,6 +17,8 @@
 
 package org.apache.kyuubi
 
+import java.lang.reflect.{InvocationTargetException, UndeclaredThrowableException}
+
 import org.apache.hive.service.rpc.thrift.TStatusCode
 
 class KyuubiSQLExceptionSuite extends KyuubiFunSuite {
@@ -50,4 +52,18 @@ class KyuubiSQLExceptionSuite extends KyuubiFunSuite {
     assert(e5.getMessage === msg0)
     assert(e5.getCause === e0)
   }
+
+  test("find the root cause") {
+    val theCause = new RuntimeException("this is just a dummy message but shall be seen")
+    val ite1 = new InvocationTargetException(theCause)
+    val ute1 = new UndeclaredThrowableException(ite1)
+    val ute2 = new UndeclaredThrowableException(ute1)
+    val ite2 = new InvocationTargetException(ute2)
+    val ke = KyuubiSQLException(ite2)
+    assert(ke.getMessage == theCause.getMessage)
+    assert(ke.getCause == theCause)
+
+    val cornerCase = new InvocationTargetException(null)
+    assert(KyuubiSQLException(cornerCase).getCause === cornerCase)
+  }
 }

kyuubi-common/src/test/scala/org/apache/kyuubi/service/ServerableSuite.scala

@@ -22,16 +22,7 @@ import org.apache.kyuubi.config.KyuubiConf
 
 class ServerableSuite extends KyuubiFunSuite {
 
-  test("Serverable") {
-    val serverable = new NoopServer()
-    serverable.stop()
-    assert(serverable.getStartTime === 0)
-    assert(serverable.getConf === null)
-    assert(serverable.getName === "noop")
-    intercept[IllegalStateException](serverable.connectionUrl)
-    assert(serverable.getServiceState === ServiceState.LATENT)
-    intercept[IllegalStateException](serverable.start())
-
+  ignore("Serverable") {
     val serverable1 = new NoopServer()
     val conf = KyuubiConf().set(KyuubiConf.FRONTEND_BIND_PORT, 0)
     serverable1.initialize(conf)
@@ -54,7 +45,7 @@ class ServerableSuite extends KyuubiFunSuite {
   test("invalid port") {
     val conf = KyuubiConf().set(KyuubiConf.FRONTEND_BIND_PORT, 100)
     val e = intercept[KyuubiException](new NoopServer().initialize(conf))
-    assert(e.getMessage === "Failed to initialize frontend service")
+    assert(e.getMessage.contains("Failed to initialize frontend service"))
     assert(e.getCause.getMessage === "Invalid Port number")
   }
 

kyuubi-ha/pom.xml

@@ -22,7 +22,7 @@
     <parent>
         <groupId>org.apache.kyuubi</groupId>
         <artifactId>kyuubi</artifactId>
-        <version>1.0.0-SNAPSHOT</version>
+        <version>1.0.4-SNAPSHOT</version>
         <relativePath>../pom.xml</relativePath>
     </parent>
     <modelVersion>4.0.0</modelVersion>

kyuubi-ha/src/main/scala/org/apache/kyuubi/ha/server/EmbeddedZkServer.scala

@@ -58,6 +58,12 @@ class EmbeddedZkServer private(name: String) extends AbstractService(name) with
 
   override def stop(): Unit = {
     if (server != null) {
+      // Just a tradeoff, otherwise we may get NPE if we call stop() immediately.
+      // (e.g. the unit test EmbeddedZkServerSuite)
+      // TestingZooKeeperMain has a bug that the CountDownLatch released before cnxnFactory inited.
+      // More details could see TestingZooKeeperMain.runFromConfig.
+      while (getStartTime > 0 && System.currentTimeMillis() - getStartTime < 5000) {
+      }
       server.close()
       server = null
     }

kyuubi-main/pom.xml

@@ -22,7 +22,7 @@
     <parent>
         <artifactId>kyuubi</artifactId>
         <groupId>org.apache.kyuubi</groupId>
-        <version>1.0.0-SNAPSHOT</version>
+        <version>1.0.4-SNAPSHOT</version>
     </parent>
     <modelVersion>4.0.0</modelVersion>
 
@@ -65,6 +65,16 @@
             <artifactId>hive-jdbc</artifactId>
             <scope>test</scope>
         </dependency>
+
+        <dependency>
+            <groupId>org.apache.hadoop</groupId>
+            <artifactId>hadoop-minikdc</artifactId>
+        </dependency>
+
+        <dependency>
+            <groupId>org.apache.directory.server</groupId>
+            <artifactId>apacheds-service</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

kyuubi-main/src/main/scala/org/apache/kyuubi/engine/ProcBuilder.scala

@@ -56,6 +56,8 @@ trait ProcBuilder {
   }
 
   @volatile private var error: Throwable = UNCAUGHT_ERROR
+  // Visible for test
+  private[kyuubi] var logCaptureThread: Thread = null
 
   final def start: Process = synchronized {
     val procLog = Paths.get(workingDir.toAbsolutePath.toString,
@@ -86,15 +88,23 @@ trait ProcBuilder {
         }
       } catch {
         case _: IOException =>
+        case _: InterruptedException =>
       } finally {
         reader.close()
       }
     }
 
-    PROC_BUILD_LOGGER.newThread(redirect).start()
+    logCaptureThread = PROC_BUILD_LOGGER.newThread(redirect)
+    logCaptureThread.start()
     proc
   }
 
+  def close(): Unit = {
+    if (logCaptureThread != null) {
+      logCaptureThread.interrupt()
+    }
+  }
+
   def getError: Throwable = synchronized {
     if (error == UNCAUGHT_ERROR) {
       Thread.sleep(3000)

kyuubi-main/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala

@@ -17,10 +17,13 @@
 
 package org.apache.kyuubi.engine.spark
 
+import java.io.IOException
 import java.nio.file.{Files, Path, Paths}
 
 import scala.collection.mutable.ArrayBuffer
 
+import org.apache.hadoop.security.UserGroupInformation
+
 import org.apache.kyuubi._
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf.ENGINE_SPARK_MAIN_RESOURCE
@@ -108,8 +111,11 @@ class SparkProcessBuilder(
       buffer += CONF
       buffer += s"$k=$v"
     }
-    buffer += PROXY_USER
-    buffer += proxyUser
+    // iff the keytab is specified, PROXY_USER is not supported
+    if (!useKeytab()) {
+      buffer += PROXY_USER
+      buffer += proxyUser
+    }
 
     mainResource.foreach { r => buffer += r }
 
@@ -119,11 +125,32 @@ class SparkProcessBuilder(
   override def toString: String = commands.mkString(" ")
 
   override protected def module: String = "kyuubi-spark-sql-engine"
+
+  private def useKeytab(): Boolean = {
+    val principal = conf.get(PRINCIPAL)
+    val keytab = conf.get(KEYTAB)
+    if (principal.isEmpty || keytab.isEmpty) {
+      false
+    } else {
+      try {
+        val ugi = UserGroupInformation
+          .loginUserFromKeytabAndReturnUGI(principal.get, keytab.get)
+        ugi.getShortUserName == proxyUser
+      } catch {
+        case e: IOException =>
+          error(s"Failed to login for ${principal.get}", e)
+          false
+      }
+    }
+  }
 }
 
 object SparkProcessBuilder {
+  final val APP_KEY = "spark.app.name"
+
   private final val CONF = "--conf"
   private final val CLASS = "--class"
   private final val PROXY_USER = "--proxy-user"
-  final val APP_KEY = "spark.app.name"
+  private final val PRINCIPAL = "spark.kerberos.principal"
+  private final val KEYTAB = "spark.kerberos.keytab"
 }

kyuubi-main/src/main/scala/org/apache/kyuubi/server/KyuubiServer.scala

@@ -36,7 +36,6 @@ object KyuubiServer extends Logging {
     if (!ServiceDiscovery.supportServiceDiscovery(conf)) {
       zkServer.initialize(conf)
       zkServer.start()
-      sys.addShutdownHook(zkServer.stop())
       conf.set(HA_ZK_QUORUM, zkServer.getConnectString)
       conf.set(HA_ZK_ACL_ENABLED, false)
     }
@@ -82,7 +81,7 @@ class KyuubiServer(name: String) extends Serverable(name) {
   override private[kyuubi] val backendService: AbstractBackendService = new KyuubiBackendService()
   private val discoveryService = new ServiceDiscovery(this)
 
-  override def initialize(conf: KyuubiConf): Unit = {
+  override def initialize(conf: KyuubiConf): Unit = synchronized {
     val kinit = new KinitAuxiliaryService()
     addService(kinit)
     super.initialize(conf)

kyuubi-main/src/main/scala/org/apache/kyuubi/session/KyuubiSessionImpl.scala

@@ -100,38 +100,47 @@ class KyuubiSessionImpl(
     super.open()
     // Init zookeeper client here to capture errors
     zkClient
-    getServerHost match {
-      case Some((host, port)) => openSession(host, port)
-      case None =>
-        sessionConf.set(SparkProcessBuilder.APP_KEY, boundAppName.toString)
-        sessionConf.set(HA_ZK_NAMESPACE, appZkNamespace)
-        val builder = new SparkProcessBuilder(appUser, sessionConf.toSparkPrefixedConf)
-        val process = builder.start
-        info(s"Launching SQL engine: $builder")
-        var sh = getServerHost
-        val started = System.currentTimeMillis()
-        var exitValue: Option[Int] = None
-        while (sh.isEmpty) {
-          if (exitValue.isEmpty && process.waitFor(1, TimeUnit.SECONDS)) {
-            exitValue = Some(process.exitValue())
-            if (exitValue.get != 0) {
-              throw builder.getError
-            }
-          }
-          if (started + timeout <= System.currentTimeMillis()) {
-            process.destroyForcibly()
-            throw KyuubiSQLException(s"Timed out($timeout ms) to launched Spark with $builder",
-              builder.getError)
-          }
-          sh = getServerHost
-        }
-        val Some((host, port)) = sh
-        openSession(host, port)
-    }
     try {
-      zkClient.close()
-    } catch {
-      case e: IOException => error("Failed to release the zkClient after session established", e)
+      getServerHost match {
+        case Some((host, port)) => openSession(host, port)
+        case None =>
+          sessionConf.set(SparkProcessBuilder.APP_KEY, boundAppName.toString)
+          sessionConf.set(HA_ZK_NAMESPACE, appZkNamespace)
+          val builder = new SparkProcessBuilder(appUser, sessionConf.toSparkPrefixedConf)
+          try {
+            val process = builder.start
+            info(s"Launching SQL engine: $builder")
+            var sh = getServerHost
+            val started = System.currentTimeMillis()
+            var exitValue: Option[Int] = None
+            while (sh.isEmpty) {
+              if (exitValue.isEmpty && process.waitFor(1, TimeUnit.SECONDS)) {
+                exitValue = Some(process.exitValue())
+                if (exitValue.get != 0) {
+                  throw builder.getError
+                }
+              }
+              if (started + timeout <= System.currentTimeMillis()) {
+                process.destroyForcibly()
+                throw KyuubiSQLException(s"Timed out($timeout ms) to launched Spark with $builder",
+                  builder.getError)
+              }
+              sh = getServerHost
+            }
+            val Some((host, port)) = sh
+            openSession(host, port)
+          } finally {
+            // we must close the process builder whether session open is success or failure since
+            // we have a log capture thread in process builder.
+            builder.close()
+          }
+      }
+    } finally {
+      try {
+        zkClient.close()
+      } catch {
+        case e: IOException => error("Failed to release the zkClient after session established", e)
+      }
     }
   }
 

kyuubi-main/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala

@@ -20,11 +20,12 @@ package org.apache.kyuubi.engine.spark
 import java.nio.file.{Files, Paths}
 import java.util.concurrent.TimeUnit
 
-import org.apache.kyuubi.{KyuubiFunSuite, KyuubiSQLException}
+import org.apache.kyuubi.{KerberizedTestHelper, KyuubiSQLException, Utils}
 import org.apache.kyuubi.config.KyuubiConf
 import org.apache.kyuubi.config.KyuubiConf._
+import org.apache.kyuubi.service.ServiceUtils
 
-class SparkProcessBuilderSuite extends KyuubiFunSuite {
+class SparkProcessBuilderSuite extends KerberizedTestHelper {
   private val conf = KyuubiConf()
     .set(EMBEDDED_ZK_PORT, 5555)
     .set(EMBEDDED_ZK_TEMP_DIR, "spark_process_test")
@@ -71,4 +72,47 @@ class SparkProcessBuilderSuite extends KyuubiFunSuite {
       error1.getMessage.contains("Caused by: org.apache.hadoop.hive.ql.metadata.HiveException:"))
   }
 
+  test("proxy user or keytab") {
+    val b1 = new SparkProcessBuilder("kentyao", conf)
+    assert(b1.toString.contains("--proxy-user kentyao"))
+
+    val conf1 = conf ++ Map("spark.kerberos.principal" -> testPrincipal)
+    val b2 = new SparkProcessBuilder("kentyao", conf1)
+    assert(b2.toString.contains("--proxy-user kentyao"))
+
+    val conf2 = conf ++ Map("spark.kerberos.keytab" -> testKeytab)
+    val b3 = new SparkProcessBuilder("kentyao", conf2)
+    assert(b3.toString.contains("--proxy-user kentyao"))
+
+    tryWithSecurityEnabled {
+      val conf3 = conf ++ Map("spark.kerberos.principal" -> testPrincipal,
+        "spark.kerberos.keytab" -> "testKeytab")
+      val b4 = new SparkProcessBuilder(Utils.currentUser, conf3)
+      assert(b4.toString.contains(s"--proxy-user ${Utils.currentUser}"))
+
+      val conf4 = conf ++ Map("spark.kerberos.principal" -> testPrincipal,
+        "spark.kerberos.keytab" -> testKeytab)
+      val b5 = new SparkProcessBuilder("kentyao", conf4)
+      assert(b5.toString.contains("--proxy-user kentyao"))
+
+      val b6 = new SparkProcessBuilder(ServiceUtils.getShortName(testPrincipal), conf4)
+      assert(!b6.toString.contains("--proxy-user kentyao"))
+    }
+  }
+
+  test("log capture should release after close") {
+    val process = new FakeSparkProcessBuilder
+    try {
+      val subProcess = process.start
+      assert(!process.logCaptureThread.isInterrupted)
+      subProcess.waitFor(3, TimeUnit.SECONDS)
+    } finally {
+      process.close()
+    }
+    assert(process.logCaptureThread.isInterrupted)
+  }
+}
+
+class FakeSparkProcessBuilder extends SparkProcessBuilder("fake", Map.empty) {
+  override protected def commands: Array[String] = Array("ls")
 }

kyuubi-main/src/test/scala/org/apache/kyuubi/server/KyuubiServerSuite.scala

@@ -60,7 +60,7 @@ class KyuubiServerSuite extends KyuubiFunSuite {
   ignore("invalid port") {
     val conf = KyuubiConf().set(KyuubiConf.FRONTEND_BIND_PORT, 100)
     val e = intercept[KyuubiException](new KyuubiServer().initialize(conf))
-    assert(e.getMessage === "Failed to initialize frontend service")
+    assert(e.getMessage.contains("Failed to initialize frontend service"))
     assert(e.getCause.getMessage === "Invalid Port number")
   }
 }

pom.xml

@@ -23,7 +23,7 @@
     <groupId>org.apache.kyuubi</groupId>
     <artifactId>kyuubi</artifactId>
     <name>Kyuubi Project Parent</name>
-    <version>1.0.0-SNAPSHOT</version>
+    <version>1.0.4-SNAPSHOT</version>
     <modules>
         <module>kyuubi-common</module>
         <module>kyuubi-ha</module>
@@ -65,7 +65,7 @@
         <scalatest.version>3.0.3</scalatest.version>
         <scala.binary.version>2.12</scala.binary.version>
         <maven.version>3.6.3</maven.version>
-        <spark.version>3.0.1</spark.version>
+        <spark.version>3.0.2</spark.version>
         <log4j.version>1.7.30</log4j.version>
     </properties>