diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ProcBuilder.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ProcBuilder.scala index d0e6bae6b..f1e49d687 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ProcBuilder.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/ProcBuilder.scala @@ -359,4 +359,6 @@ object ProcBuilder extends Logging { private val PROC_BUILD_LOGGER = new NamedThreadFactory("process-logger-capture", daemon = true) private val UNCAUGHT_ERROR = new RuntimeException("Uncaught error") + + private[engine] val KYUUBI_ENGINE_LOG_PATH_KEY = "kyuubi.engine.engineLog.path" } diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilder.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilder.scala index 0d20068de..7f1be93b5 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilder.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkBatchProcessBuilder.scala @@ -36,7 +36,7 @@ class SparkBatchProcessBuilder( extends SparkProcessBuilder(proxyUser, conf, batchId, extraEngineLog) { import SparkProcessBuilder._ - override protected val commands: Array[String] = { + override protected lazy val commands: Array[String] = { val buffer = new ArrayBuffer[String]() buffer += executable Option(mainClass).foreach { cla => @@ -51,7 +51,7 @@ class SparkBatchProcessBuilder( // tag batch application KyuubiApplicationManager.tagApplication(batchId, "spark", clusterManager(), batchKyuubiConf) - (batchKyuubiConf.getAll ++ sparkAppNameConf()).foreach { case (k, v) => + (batchKyuubiConf.getAll ++ sparkAppNameConf() ++ engineLogPathConf()).foreach { case (k, v) => buffer += CONF buffer += s"${convertConfigKey(k)}=$v" } diff --git a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala index 5bfe506da..5998d5d4e 100644 --- a/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala +++ b/kyuubi-server/src/main/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilder.scala @@ -30,6 +30,7 @@ import org.apache.kyuubi._ import org.apache.kyuubi.config.KyuubiConf import org.apache.kyuubi.engine.{ApplicationManagerInfo, KyuubiApplicationManager, ProcBuilder} import org.apache.kyuubi.engine.KubernetesApplicationOperation.{KUBERNETES_SERVICE_HOST, KUBERNETES_SERVICE_PORT} +import org.apache.kyuubi.engine.ProcBuilder.KYUUBI_ENGINE_LOG_PATH_KEY import org.apache.kyuubi.ha.HighAvailabilityConf import org.apache.kyuubi.ha.client.AuthTypes import org.apache.kyuubi.operation.log.OperationLog @@ -99,7 +100,7 @@ class SparkProcessBuilder( } } - override protected val commands: Array[String] = { + override protected lazy val commands: Array[String] = { // complete `spark.master` if absent on kubernetes completeMasterUrl(conf) @@ -116,8 +117,8 @@ class SparkProcessBuilder( == AuthTypes.KERBEROS) { allConf = allConf ++ zkAuthKeytabFileConf(allConf) } - - allConf.foreach { case (k, v) => + // pass spark engine log path to spark conf + (allConf ++ engineLogPathConf).foreach { case (k, v) => buffer += CONF buffer += s"${convertConfigKey(k)}=$v" } @@ -244,6 +245,10 @@ class SparkProcessBuilder( } } } + + private[spark] def engineLogPathConf(): Map[String, String] = { + Map(KYUUBI_ENGINE_LOG_PATH_KEY -> engineLog.getAbsolutePath) + } } object SparkProcessBuilder { diff --git a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala index ad8324c85..cd549c0f3 100644 --- a/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala +++ b/kyuubi-server/src/test/scala/org/apache/kyuubi/engine/spark/SparkProcessBuilderSuite.scala @@ -29,6 +29,8 @@ import org.scalatestplus.mockito.MockitoSugar import org.apache.kyuubi.{KerberizedTestHelper, KyuubiSQLException, Utils} import org.apache.kyuubi.config.KyuubiConf import org.apache.kyuubi.config.KyuubiConf.{ENGINE_LOG_TIMEOUT, ENGINE_SPARK_MAIN_RESOURCE} +import org.apache.kyuubi.engine.ProcBuilder.KYUUBI_ENGINE_LOG_PATH_KEY +import org.apache.kyuubi.engine.spark.SparkProcessBuilder.CONF import org.apache.kyuubi.ha.HighAvailabilityConf import org.apache.kyuubi.ha.client.AuthTypes import org.apache.kyuubi.service.ServiceUtils @@ -296,9 +298,16 @@ class SparkProcessBuilderSuite extends KerberizedTestHelper with MockitoSugar { assert(!c3.contains(s"spark.kubernetes.driverEnv.SPARK_USER_NAME=$proxyName")) assert(!c3.contains(s"spark.executorEnv.SPARK_USER_NAME=$proxyName")) } + + test("[KYUUBI #5009] Test pass spark engine log path to spark conf") { + val b1 = new SparkProcessBuilder("kyuubi", conf) + assert( + b1.toString.contains( + s"$CONF spark.$KYUUBI_ENGINE_LOG_PATH_KEY=${b1.engineLog.getAbsolutePath}")) + } } class FakeSparkProcessBuilder(config: KyuubiConf) extends SparkProcessBuilder("fake", config) { - override protected val commands: Array[String] = Array("ls") + override protected lazy val commands: Array[String] = Array("ls") }