From f7ec0a3412df21fbe971c7e40aaa666c32aad47a Mon Sep 17 00:00:00 2001 From: Aleksandr Efimov Date: Fri, 28 Aug 2026 10:40:06 +0300 Subject: [PATCH] [KYUUBI #7511][SPARK] Start Spark Connect in the Spark SQL engine The Spark SQL engine can serve Spark Connect clients next to its Thrift frontend. Connect is started by configuration only - the plugin goes into spark.plugins and the bound endpoint is read back reflectively - because io.grpc is relocated in this module and a compile-time reference to SparkConnectService does not compile here. That also keeps the default spark-3.5 build free of Connect classes. Which engines serve Connect is decided in SparkProcessBuilder, where the values are the server's own. kyuubi.engine.spark.connect.enabled is immutable, but the engine rebuilds its KyuubiConf from every spark.kyuubi.* entry it is handed, so the prefixed copy of that key is dropped from the engine's command line. An engine that would not run as the session user - any share level but USER, or doAs disabled - starts without Connect rather than failing, because a plan submitted over Spark Connect runs as the user the engine runs as, not as the client's user_id. With Connect on, the engine still refuses to start when the runtime cannot authenticate its clients - Spark 3.5 serves Connect with no authentication at all - or when no token is configured: an endpoint that authenticates nobody is worse than no endpoint. The Spark 4.0 requirement is a capability probe rather than a version comparison. --- docs/configuration/settings.md | 1 + docs/deployment/spark/index.rst | 1 + docs/deployment/spark/spark_connect.md | 78 +++++++++ .../kyuubi/engine/spark/SparkSQLEngine.scala | 11 ++ .../spark/connect/SparkConnectLauncher.scala | 163 ++++++++++++++++++ .../spark/connect/FakeConnectClasses.scala | 35 ++++ .../connect/SparkConnectLauncherSuite.scala | 112 ++++++++++++ .../org/apache/kyuubi/config/KyuubiConf.scala | 12 ++ .../engine/spark/SparkProcessBuilder.scala | 28 ++- .../spark/SparkProcessBuilderSuite.scala | 37 ++++ 10 files changed, 477 insertions(+), 1 deletion(-) create mode 100644 docs/deployment/spark/spark_connect.md create mode 100644 externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/connect/SparkConnectLauncher.scala create mode 100644 externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/connect/FakeConnectClasses.scala create mode 100644 externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/connect/SparkConnectLauncherSuite.scala diff --git a/docs/configuration/settings.md b/docs/configuration/settings.md index 2562f0ef94c..7d48895297f 100644 --- a/docs/configuration/settings.md +++ b/docs/configuration/settings.md @@ -191,6 +191,7 @@ You can configure the Kyuubi properties in `$KYUUBI_HOME/conf/kyuubi-defaults.co | kyuubi.engine.share.level.sub.domain | <undefined> | (deprecated) - Using kyuubi.engine.share.level.subdomain instead | string | 1.2.0 | | kyuubi.engine.share.level.subdomain | <undefined> | Allow end-users to create a subdomain for the share level of an engine. A subdomain is a case-insensitive string values that must be a valid zookeeper subpath. For example, for the `USER` share level, an end-user can share a certain engine within a subdomain, not for all of its clients. End-users are free to create multiple engines in the `USER` share level. When disable engine pool, use 'default' if absent. | string | 1.4.0 | | kyuubi.engine.single.spark.session | false | When set to true, this engine is running in a single session mode. All the JDBC/ODBC connections share the temporary views, function registries, SQL configuration and the current database. | boolean | 1.3.0 | +| kyuubi.engine.spark.connect.enabled | false | When true, the Spark SQL engine starts a Spark Connect service next to its Thrift frontend, so that Spark Connect clients can use the same engine. It requires a Spark distribution whose Spark Connect supports authentication, which is Spark 4.0 and later, a pre-shared `spark.connect.authenticate.token`, and the USER share level; the engine fails to start otherwise. | boolean | 1.13.0 | | kyuubi.engine.spark.event.loggers | SPARK | A comma-separated list of engine loggers, where engine/session/operation etc events go. | seq | 1.7.0 | | kyuubi.engine.spark.initialize.sql | SHOW DATABASES | The initialize sql for Spark engine. It fallback to `kyuubi.engine.initialize.sql`. | seq | 1.8.1 | | kyuubi.engine.spark.operation.incremental.collect | false | When true, the result will be sequentially calculated and returned to the Spark driver. Note that, kyuubi.operation.result.max.rows will be ignored on incremental collect mode. It fallback to `kyuubi.operation.incremental.collect` | boolean | 1.10.0 | diff --git a/docs/deployment/spark/index.rst b/docs/deployment/spark/index.rst index 1beb48d6406..1ecfe1cbf7c 100644 --- a/docs/deployment/spark/index.rst +++ b/docs/deployment/spark/index.rst @@ -30,3 +30,4 @@ Even if you don't use Kyuubi, as a simple Spark user, I'm sure you'll find the n dynamic_allocation aqe large_query_results + spark_connect diff --git a/docs/deployment/spark/spark_connect.md b/docs/deployment/spark/spark_connect.md new file mode 100644 index 00000000000..e6dc284f240 --- /dev/null +++ b/docs/deployment/spark/spark_connect.md @@ -0,0 +1,78 @@ + + +# Spark Connect in the Spark SQL Engine + +The Spark SQL engine can start a Spark Connect service next to its Thrift frontend, so that +Spark Connect clients reach a Kyuubi-managed engine: the same `SparkContext`, the same session +extensions and so the same authorization rules a JDBC client gets. Spark gives each Connect session +its own `SparkSession`, so temporary views, registered functions and SQL configs are not shared with +the JDBC sessions of that engine. + +The feature is off by default and is enabled per engine: + +```properties +kyuubi.engine.spark.connect.enabled=true +spark.connect.authenticate.token= +``` + +`kyuubi.engine.spark.connect.enabled` is immutable and the server decides it per engine, so a +session cannot turn Spark Connect on by itself — neither under that key nor under the +`spark.kyuubi.` prefixed copy that the engine would otherwise read. + +## Requirements + +| Requirement | Why | +|----------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| +| Spark 4.0 or later, with the `spark-connect` jar in the distribution | `spark.connect.authenticate.token` was added in Spark 4.0. Spark 3.5 serves Spark Connect without authentication, so the engine refuses to start rather than open an endpoint that authenticates no one | +| `spark.connect.authenticate.token`, or the `SPARK_CONNECT_AUTHENTICATE_TOKEN` environment variable | Every client that reaches the engine is authenticated by this pre-shared token | +| `kyuubi.engine.share.level=USER` and `kyuubi.engine.doAs.enabled=true` | A plan submitted over Spark Connect runs as the user the engine runs as, not as the `user_id` the client sends. Only these two together make that the session user; an engine started any other way serves Thrift only and says so in its launch log | + +A runtime that cannot authenticate Spark Connect clients, or a missing token, fails the engine with +a message naming what is missing: the deployment asked for Connect and cannot have it. The share +level is the session's own choice, so an engine that would not run as the session user starts +without Connect instead of failing. + +## What the engine configures + +Unless the deployment sets them, the engine adds: + +- `spark.plugins` gains `org.apache.spark.sql.connect.SparkConnectPlugin`, which is how Spark + starts Connect inside the driver; +- `spark.connect.grpc.binding.port=0`, so that engines of different users on the same host do not + compete for a fixed port. The port that was bound is logged as `sc://:`. + +Configs the deployment sets are kept, so a fixed port or extra plugins keep working. + +## Limitations + +- The Spark Connect endpoint is not advertised in engine discovery yet, so clients need the + address from the engine log. +- Engine idle timeout and graceful shutdown still count Thrift sessions only. An engine whose only + client is a Spark Connect session can be terminated as idle. +- The token is passed to the engine like any other Spark config, so it is visible in the driver's + command line. +- On a Spark distribution without [SPARK-58658](https://issues.apache.org/jira/browse/SPARK-58658), + an authenticated Spark Connect client reads the engine's configuration back through the Config + RPC, and the secrets the server hands the engine are in it — + `spark.kyuubi.ha.zookeeper.auth.digest` among them. The engine names them in + `spark.redaction.regex`, but only the patched handler consults that pattern. The fix is merged + on every Spark branch and released in none: 4.0.4, 4.1.3 and 4.2.0 all predate it, so it arrives + in the next patch release of each line. +- Share levels other than `USER`, and engines started with `kyuubi.engine.doAs.enabled=false`, do + not serve Spark Connect. + diff --git a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngine.scala b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngine.scala index 485a955a755..5f3f66471c5 100644 --- a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngine.scala +++ b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/SparkSQLEngine.scala @@ -38,6 +38,7 @@ import org.apache.kyuubi.config.KyuubiConf._ import org.apache.kyuubi.config.KyuubiReservedKeys.{KYUUBI_ENGINE_SUBMIT_TIME_KEY, KYUUBI_ENGINE_URL} import org.apache.kyuubi.engine.ShareLevel import org.apache.kyuubi.engine.spark.SparkSQLEngine.{countDownLatch, currentEngine} +import org.apache.kyuubi.engine.spark.connect.SparkConnectLauncher import org.apache.kyuubi.engine.spark.events.{EngineEvent, EngineEventsStore, SparkEventHandlerRegister} import org.apache.kyuubi.engine.spark.session.{SparkSessionImpl, SparkSQLSessionManager} import org.apache.kyuubi.events.EventBus @@ -326,6 +327,10 @@ object SparkSQLEngine extends Logging { } def createSpark(): SparkSession = { + if (SparkConnectLauncher.isEnabled(kyuubiConf)) { + SparkConnectLauncher.setup(kyuubiConf, _sparkConf) + } + val engineCredentials = kyuubiConf.getOption(KyuubiReservedKeys.KYUUBI_ENGINE_CREDENTIALS_KEY) kyuubiConf.unset(KyuubiReservedKeys.KYUUBI_ENGINE_CREDENTIALS_KEY) _sparkConf.set(s"spark.${KyuubiReservedKeys.KYUUBI_ENGINE_CREDENTIALS_KEY}", "") @@ -407,6 +412,12 @@ object SparkSQLEngine extends Logging { startInitTimeoutChecker(submitTime, initTimeout) spark = createSpark() sparkSessionCreated.set(true) + if (SparkConnectLauncher.isEnabled(kyuubiConf)) { + SparkConnectLauncher.boundEndpoint(spark.sparkContext.getConf).foreach { + case (host, port) => + info(s"Spark Connect service is listening on sc://$host:$port") + } + } try { startEngine(spark) // blocking main thread diff --git a/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/connect/SparkConnectLauncher.scala b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/connect/SparkConnectLauncher.scala new file mode 100644 index 00000000000..da451b181c9 --- /dev/null +++ b/externals/kyuubi-spark-sql-engine/src/main/scala/org/apache/kyuubi/engine/spark/connect/SparkConnectLauncher.scala @@ -0,0 +1,163 @@ +/* + * 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.connect + +import scala.util.Try +import scala.util.control.NonFatal + +import org.apache.spark.SparkConf + +import org.apache.kyuubi.{KyuubiException, Logging} +import org.apache.kyuubi.config.KyuubiConf +import org.apache.kyuubi.config.KyuubiConf.{ENGINE_SHARE_LEVEL, ENGINE_SPARK_CONNECT_ENABLED} +import org.apache.kyuubi.engine.ShareLevel +import org.apache.kyuubi.util.reflect.{DynFields, DynMethods, ReflectUtils} + +/** + * Starts Spark Connect inside the Spark SQL engine, next to the Thrift frontend, by writing the + * Spark configs that make the driver load the Connect plugin. + * + * Spark Connect classes are never referenced at compile time: `io.grpc` is relocated to + * `org.apache.kyuubi.shade.io.grpc` in this module, and the Scala signatures of + * `SparkConnectService` carry `io.grpc` types, so a direct reference does not compile. That also + * keeps the engine building against the default `spark-3.5` profile, where the feature stays off. + */ +object SparkConnectLauncher extends Logging { + + final val PLUGINS_KEY = "spark.plugins" + final val BINDING_PORT_KEY = "spark.connect.grpc.binding.port" + final val AUTHENTICATE_TOKEN_KEY = "spark.connect.authenticate.token" + final val AUTHENTICATE_TOKEN_ENV = "SPARK_CONNECT_AUTHENTICATE_TOKEN" + final val DRIVER_HOST_KEY = "spark.driver.host" + + final private val CONNECT_PLUGIN_CLASS = "org.apache.spark.sql.connect.SparkConnectPlugin" + final private val CONNECT_CONFIG_CLASS = "org.apache.spark.sql.connect.config.Connect$" + final private val CONNECT_SERVICE_CLASS = + "org.apache.spark.sql.connect.service.SparkConnectService$" + + def isEnabled(kyuubiConf: KyuubiConf): Boolean = kyuubiConf.get(ENGINE_SPARK_CONNECT_ENABLED) + + /** + * Only the USER share level is supported for now: a Spark Connect client authenticates with a + * token that is shared by everyone reaching the engine, and the plan runs as the engine's own + * user, so a shared engine would execute one user's plans under another user's identity. + * + * `SparkProcessBuilder` decides whether an engine serves Connect at all, where the share level + * and doAs are the server's own values; this is the engine refusing a combination that should + * never have been launched. + */ + def checkShareLevel(kyuubiConf: KyuubiConf): Unit = { + val shareLevel = kyuubiConf.get(ENGINE_SHARE_LEVEL) + if (shareLevel != ShareLevel.USER.toString) { + throw new KyuubiException( + s"${ENGINE_SPARK_CONNECT_ENABLED.key} requires ${ENGINE_SHARE_LEVEL.key}=" + + s"${ShareLevel.USER}, but it is $shareLevel. Other share levels are not supported yet.") + } + } + + /** + * Spark 3.5 ships Spark Connect without any authentication - `spark.connect.authenticate.token` + * arrived in 4.0 - so enabling it there would open an unauthenticated endpoint on the engine. + * The capability is probed instead of comparing versions. + */ + def checkRuntimeSupport( + pluginClass: String = CONNECT_PLUGIN_CLASS, + configClass: String = CONNECT_CONFIG_CLASS): Unit = { + if (!ReflectUtils.isClassLoadable(pluginClass)) { + throw new KyuubiException( + s"${ENGINE_SPARK_CONNECT_ENABLED.key} is set, but $pluginClass is not on the" + + " engine classpath. Spark Connect ships in the spark-connect jar of a Spark 4.0 or" + + " later distribution.") + } + if (!authenticationSupported(configClass)) { + throw new KyuubiException( + s"${ENGINE_SPARK_CONNECT_ENABLED.key} is set, but this Spark runtime has a Spark Connect" + + s" without authentication support - $AUTHENTICATE_TOKEN_KEY was introduced in Spark" + + " 4.0. Enabling it here would expose an unauthenticated endpoint.") + } + } + + /** + * The engine refuses to open the endpoint without a pre-shared token. Generating one at engine + * launch is a separate change; today the deployment supplies it. + * + * The config wins over the environment even when it is empty, because that is the order + * `Connect.getAuthenticateToken` reads them in: an empty config would install an interceptor + * with an empty key rather than fall back to the environment. + */ + def checkAuthenticateToken(sparkConf: SparkConf): Unit = { + val token = sparkConf.getOption(AUTHENTICATE_TOKEN_KEY) + .orElse(sys.env.get(AUTHENTICATE_TOKEN_ENV)) + if (!token.exists(_.nonEmpty)) { + throw new KyuubiException( + s"${ENGINE_SPARK_CONNECT_ENABLED.key} is set, but no Spark Connect token is configured." + + s" Set $AUTHENTICATE_TOKEN_KEY or the $AUTHENTICATE_TOKEN_ENV environment variable so" + + " that clients are authenticated.") + } + } + + /** + * The Spark configs to add so that the driver starts Connect. Configs the deployment set are + * kept as they are. + */ + def sparkConfOverrides(sparkConf: SparkConf): Seq[(String, String)] = { + val plugins = sparkConf.get(PLUGINS_KEY, "").split(",").map(_.trim).filter(_.nonEmpty) + val pluginOverride = + if (plugins.contains(CONNECT_PLUGIN_CLASS)) Nil + else Seq(PLUGINS_KEY -> (plugins :+ CONNECT_PLUGIN_CLASS).mkString(",")) + // Kyuubi runs one engine per user on a host, so bind an ephemeral port unless the deployment + // picked one; the port that was actually bound is read back afterwards. + val portOverride = + if (sparkConf.contains(BINDING_PORT_KEY)) Nil else Seq(BINDING_PORT_KEY -> "0") + pluginOverride ++ portOverride + } + + def setup(kyuubiConf: KyuubiConf, sparkConf: SparkConf): Unit = { + checkShareLevel(kyuubiConf) + checkRuntimeSupport() + checkAuthenticateToken(sparkConf) + sparkConfOverrides(sparkConf).foreach { case (key, value) => sparkConf.set(key, value) } + } + + /** + * The endpoint a Spark Connect client reaches, once the plugin has started the service. + * + * The host comes from `spark.driver.host` rather than `SparkConnectService.hostAddress`: the + * latter is the local canonical host name, which on Kubernetes is the pod name and does not + * resolve outside the pod. + */ + def boundEndpoint( + sparkConf: SparkConf, + serviceClass: String = CONNECT_SERVICE_CLASS): Option[(String, Int)] = { + try { + val service = DynFields.builder() + .impl(serviceClass, "MODULE$") + .buildChecked[Object]() + .get(null) + val port = ReflectUtils.invokeAs[Int](service, "localPort") + sparkConf.getOption(DRIVER_HOST_KEY).map(_ -> port) + } catch { + case NonFatal(e) => + warn("Failed to read the Spark Connect endpoint back from the driver", e) + None + } + } + + private[connect] def authenticationSupported(configClass: String): Boolean = + Try(DynMethods.builder("getAuthenticateToken").impl(configClass).buildChecked()).isSuccess +} diff --git a/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/connect/FakeConnectClasses.scala b/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/connect/FakeConnectClasses.scala new file mode 100644 index 00000000000..8e8864eec15 --- /dev/null +++ b/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/connect/FakeConnectClasses.scala @@ -0,0 +1,35 @@ +/* + * 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.connect + +/** + * Stand-ins for the Spark Connect classes the engine reaches by reflection, so that the reflection + * itself is exercised. The engine module does not depend on `spark-connect`, and adding the real + * jar to the test classpath is a dependency problem of its own. + */ +object FakeConnectPlugin + +/** Shaped like `org.apache.spark.sql.connect.config.Connect` on a Spark that authenticates. */ +object FakeConnectConfig { + def getAuthenticateToken: Option[String] = Some("a-token") +} + +/** Shaped like `org.apache.spark.sql.connect.service.SparkConnectService` with a bound port. */ +object FakeConnectService { + def localPort: Int = 15002 +} diff --git a/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/connect/SparkConnectLauncherSuite.scala b/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/connect/SparkConnectLauncherSuite.scala new file mode 100644 index 00000000000..05998f38fa1 --- /dev/null +++ b/externals/kyuubi-spark-sql-engine/src/test/scala/org/apache/kyuubi/engine/spark/connect/SparkConnectLauncherSuite.scala @@ -0,0 +1,112 @@ +/* + * 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.connect + +import org.apache.spark.SparkConf + +import org.apache.kyuubi.{KyuubiException, KyuubiFunSuite} +import org.apache.kyuubi.config.KyuubiConf +import org.apache.kyuubi.config.KyuubiConf.{ENGINE_SHARE_LEVEL, ENGINE_SPARK_CONNECT_ENABLED} +import org.apache.kyuubi.engine.ShareLevel +import org.apache.kyuubi.engine.spark.connect.SparkConnectLauncher._ + +class SparkConnectLauncherSuite extends KyuubiFunSuite { + + private val connectPlugin = "org.apache.spark.sql.connect.SparkConnectPlugin" + private val fakePluginClass = FakeConnectPlugin.getClass.getName + private val fakeConfigClass = FakeConnectConfig.getClass.getName + private val fakeServiceClass = FakeConnectService.getClass.getName + + test("disabled unless the engine is configured for it") { + assert(!isEnabled(KyuubiConf())) + assert(isEnabled(KyuubiConf().set(ENGINE_SPARK_CONNECT_ENABLED, true))) + } + + test("add the Connect plugin and bind an ephemeral port") { + val overrides = sparkConfOverrides(new SparkConf(false)).toMap + assert(overrides(PLUGINS_KEY) === connectPlugin) + assert(overrides(BINDING_PORT_KEY) === "0") + } + + test("keep the plugins and the port the deployment configured") { + val sparkConf = new SparkConf(false) + .set(PLUGINS_KEY, "org.apache.spark.custom.Plugin") + .set(BINDING_PORT_KEY, "15002") + val overrides = sparkConfOverrides(sparkConf).toMap + assert(overrides(PLUGINS_KEY) === s"org.apache.spark.custom.Plugin,$connectPlugin") + assert(!overrides.contains(BINDING_PORT_KEY)) + } + + test("do not add the Connect plugin twice") { + val sparkConf = new SparkConf(false).set(PLUGINS_KEY, connectPlugin) + assert(!sparkConfOverrides(sparkConf).toMap.contains(PLUGINS_KEY)) + } + + test("only the USER share level is supported") { + checkShareLevel(KyuubiConf().set(ENGINE_SHARE_LEVEL, ShareLevel.USER.toString)) + Seq(ShareLevel.CONNECTION, ShareLevel.GROUP, ShareLevel.SERVER).foreach { shareLevel => + val e = intercept[KyuubiException] { + checkShareLevel(KyuubiConf().set(ENGINE_SHARE_LEVEL, shareLevel.toString)) + } + assert(e.getMessage.contains(s"requires ${ENGINE_SHARE_LEVEL.key}=${ShareLevel.USER}")) + } + } + + test("refuse to open the endpoint without a token") { + val e = intercept[KyuubiException] { + checkAuthenticateToken(new SparkConf(false)) + } + assert(e.getMessage.contains(AUTHENTICATE_TOKEN_KEY)) + checkAuthenticateToken(new SparkConf(false).set(AUTHENTICATE_TOKEN_KEY, "a-token")) + } + + test("an empty token is no token, the way Spark reads it") { + // Connect.getAuthenticateToken takes the config first and only falls back to the environment + // when it is absent, so an empty config value authenticates nobody + intercept[KyuubiException] { + checkAuthenticateToken(new SparkConf(false).set(AUTHENTICATE_TOKEN_KEY, "")) + } + } + + test("probe the runtime for Spark Connect authentication support") { + assert(authenticationSupported(fakeConfigClass)) + assert(!authenticationSupported(fakeServiceClass)) + assert(!authenticationSupported("org.apache.kyuubi.engine.spark.connect.NoSuchClass$")) + } + + test("accept a runtime whose Spark Connect authenticates, refuse one that does not") { + checkRuntimeSupport(fakePluginClass, fakeConfigClass) + val e = intercept[KyuubiException](checkRuntimeSupport(fakePluginClass, fakeServiceClass)) + assert(e.getMessage.contains("without authentication support")) + } + + test("read the bound endpoint back from the Connect service") { + val sparkConf = new SparkConf(false).set(DRIVER_HOST_KEY, "10.0.0.1") + assert(boundEndpoint(sparkConf, fakeServiceClass) === Some("10.0.0.1" -> 15002)) + // nothing to advertise when the driver host is unknown + assert(boundEndpoint(new SparkConf(false), fakeServiceClass).isEmpty) + // nor when the service is not there at all + assert(boundEndpoint(sparkConf).isEmpty) + } + + test("refuse to start when the Spark distribution has no Spark Connect") { + // the engine module does not depend on spark-connect, so this is the 'no Connect' runtime + val e = intercept[KyuubiException](checkRuntimeSupport()) + assert(e.getMessage.contains(connectPlugin)) + } +} diff --git a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala index 9fff9ee32f4..2ed228120bf 100644 --- a/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala +++ b/kyuubi-common/src/main/scala/org/apache/kyuubi/config/KyuubiConf.scala @@ -4151,6 +4151,18 @@ object KyuubiConf { .toSequence() .createWithDefault(Seq("spark.driver.memory", "spark.executor.memory")) + val ENGINE_SPARK_CONNECT_ENABLED: ConfigEntry[Boolean] = + buildConf("kyuubi.engine.spark.connect.enabled") + .doc("When true, the Spark SQL engine starts a Spark Connect service next to its Thrift" + + " frontend, so that Spark Connect clients can use the same engine. It requires a Spark" + + " distribution whose Spark Connect supports authentication, which is Spark 4.0 and" + + " later, a pre-shared `spark.connect.authenticate.token`, and the USER share level;" + + " the engine fails to start otherwise.") + .version("1.13.0") + .immutable + .booleanConf + .createWithDefault(false) + val ENGINE_SPARK_INITIALIZE_SQL: ConfigEntry[Seq[String]] = buildConf("kyuubi.engine.spark.initialize.sql") .doc("The initialize sql for Spark engine. It fallback to `kyuubi.engine.initialize.sql`.") 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 560e659d917..c1372e12ce5 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 @@ -34,7 +34,7 @@ import org.apache.hadoop.security.UserGroupInformation import org.apache.kyuubi._ import org.apache.kyuubi.config.KyuubiConf import org.apache.kyuubi.config.KyuubiConf._ -import org.apache.kyuubi.engine.{ApplicationManagerInfo, EngineType, KyuubiApplicationManager, ProcBuilder} +import org.apache.kyuubi.engine.{ApplicationManagerInfo, EngineType, KyuubiApplicationManager, ProcBuilder, ShareLevel} 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 @@ -146,6 +146,13 @@ class SparkProcessBuilder( if (AuthTypes.withName(conf.get(HA_ZK_ENGINE_AUTH_TYPE)) == AuthTypes.KERBEROS) { allConf = allConf ++ zkAuthKeytabFileConf(allConf) } + // The engine rebuilds its own KyuubiConf from every `spark.kyuubi.*` entry it is handed, so + // this prefixed copy would turn Spark Connect on inside the engine without the server ever + // agreeing - `kyuubi.engine.spark.connect.enabled` is immutable under its own name only. + allConf = allConf - convertConfigKey(ENGINE_SPARK_CONNECT_ENABLED.key) + if (!sparkConnectEnabled) { + allConf = allConf - ENGINE_SPARK_CONNECT_ENABLED.key + } // pass spark engine log path to spark conf (allConf ++ engineLogPathConf ++ @@ -163,6 +170,25 @@ class SparkProcessBuilder( buffer } + /** + * Whether this engine serves Spark Connect. The server decides, because a Spark Connect client + * is authorized as the user the engine runs as - its `user_id` never reaches execution - and + * only a `USER` engine with doAs runs as the session user. Other engines start without Connect + * rather than failing, since the share level is the session's own choice. + */ + private[kyuubi] lazy val sparkConnectEnabled: Boolean = { + conf.get(ENGINE_SPARK_CONNECT_ENABLED) && { + val shareLevel = conf.get(ENGINE_SHARE_LEVEL) + val runsAsSessionUser = shareLevel == ShareLevel.USER.toString && doAsEnabled + if (!runsAsSessionUser) { + warn(s"Not starting Spark Connect for this engine: it serves the session user only with" + + s" ${ENGINE_SHARE_LEVEL.key}=${ShareLevel.USER} and ${ENGINE_DO_AS_ENABLED.key}=true," + + s" which are $shareLevel and $doAsEnabled here") + } + runsAsSessionUser + } + } + override protected def module: String = "kyuubi-spark-sql-engine" protected def setupKerberos(buffer: mutable.Buffer[String]): Unit = { 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 be24c53c20c..6216d886791 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 @@ -31,6 +31,7 @@ import org.apache.kyuubi._ import org.apache.kyuubi.config.KyuubiConf import org.apache.kyuubi.config.KyuubiConf._ import org.apache.kyuubi.engine.ProcBuilder.KYUUBI_ENGINE_LOG_PATH_KEY +import org.apache.kyuubi.engine.ShareLevel import org.apache.kyuubi.engine.spark.SparkProcessBuilder._ import org.apache.kyuubi.ha.HighAvailabilityConf import org.apache.kyuubi.ha.client.AuthTypes @@ -59,6 +60,42 @@ class SparkProcessBuilderSuite extends KerberizedTestHelper with MockitoSugar { process.destroyForcibly() } + test("the server decides that an engine serves Spark Connect") { + val builder = new SparkProcessBuilder( + "kentyao", + true, + conf.set(ENGINE_SPARK_CONNECT_ENABLED, true) + .set(ENGINE_SHARE_LEVEL, ShareLevel.USER.toString)) + assert(builder.sparkConnectEnabled) + assert(builder.commands.exists(_ == s"spark.${ENGINE_SPARK_CONNECT_ENABLED.key}=true")) + } + + test("a session cannot turn Spark Connect on through the spark.kyuubi prefix") { + // the engine rebuilds its KyuubiConf from `spark.kyuubi.*`, where the key is not immutable + val builder = new SparkProcessBuilder( + "kentyao", + true, + conf.set(s"spark.${ENGINE_SPARK_CONNECT_ENABLED.key}", "true")) + assert(!builder.sparkConnectEnabled) + assert(!builder.commands.exists(_.contains(ENGINE_SPARK_CONNECT_ENABLED.key))) + } + + test("an engine that does not run as the session user serves Thrift only") { + val sharedEngine = new SparkProcessBuilder( + "kentyao", + true, + conf.set(ENGINE_SPARK_CONNECT_ENABLED, true) + .set(ENGINE_SHARE_LEVEL, ShareLevel.CONNECTION.toString)) + assert(!sharedEngine.sparkConnectEnabled) + assert(!sharedEngine.commands.exists(_.contains(ENGINE_SPARK_CONNECT_ENABLED.key))) + + val withoutDoAs = new SparkProcessBuilder( + "kentyao", + false, + conf.set(ENGINE_SPARK_CONNECT_ENABLED, true)) + assert(!withoutDoAs.sparkConnectEnabled) + } + test("capture error from spark process builder") { val processBuilder = new SparkProcessBuilder("kentyao", true, conf.set("spark.ui.port", "abc")) processBuilder.start