LuciferYang commented on code in PR #55917:
URL: https://github.com/apache/spark/pull/55917#discussion_r3638634551
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -219,6 +219,12 @@ private[connect] class ExecuteThreadRunner(executeHolder:
ExecuteHolder) extends
s"Spark Connect - ${Utils.abbreviate(debugString, 128)}")
session.sparkContext.setInterruptOnCancel(true)
+ val sessionUser = executeHolder.sessionHolder.userId
+ if (sessionUser != null && sessionUser.nonEmpty) {
+ session.sparkContext.setLocalProperty(
Review Comment:
If you move the `set` into `withSession` per the other comment, please pair
it with a try/finally that restores the previous value. `sc.localProperties` is
an `InheritableThreadLocal`, and analyze/config run on reused gRPC handler
threads, so a set-without-restore lets one user's `spark.connect.session.user`
linger into the next request on that thread — Ranger could then authorize under
the stale identity. The current code happens to dodge this only because execute
runs on a throwaway `ExecutionThread` (the comment at lines 49-51 also flags
that thread-pooling would need thread-local handling), but a shared entry point
requires explicit cleanup.
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -219,6 +219,12 @@ private[connect] class ExecuteThreadRunner(executeHolder:
ExecuteHolder) extends
s"Spark Connect - ${Utils.abbreviate(debugString, 128)}")
session.sparkContext.setInterruptOnCancel(true)
+ val sessionUser = executeHolder.sessionHolder.userId
Review Comment:
This property is only set in `ExecuteThreadRunner`, but the analyze path
(`SparkConnectAnalyzeHandler`) also goes through `sessionHolder.withSession`
and runs plan analysis plus `persist`/`unpersist`, yet it never gets
`spark.connect.session.user`. Since the goal is RBAC via Ranger, analyze-style
operations need the user identity too — wiring only the execute path leaves the
feature half-covered. `userId` is already a field of `SessionHolder`, and
`withSession` is the shared entry point for execute/analyze/config, so would it
be cleaner to set the property inside `SessionHolder.withSession`? That covers
every path at once and avoids having to remember an extra line each time a new
path is added. Note this needs a matching restore of the previous value once
moved.
Amd this property is an external contract that the Ranger RBAC side depends
on, so it's more worth protecting than debug-only info like callSite. Consider
a tiny test: after an execute, assert
`sc.getLocalProperty(SPARK_CONNECT_SESSION_USER_KEY)` equals the expected user.
It's cheap and guards against a later refactor silently dropping it. I realize
the local properties in this module (callSite, etc.) aren't individually
tested, so this isn't a merge blocker; it's just that this key is an external
integration point, which is a bit different in nature.
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/service/SparkConnectService.scala:
##########
@@ -314,6 +314,7 @@ class SparkConnectService(debug: Boolean) extends
AsyncService with BindableServ
*/
object SparkConnectService extends Logging {
+ private[connect] val SPARK_CONNECT_SESSION_USER_KEY: String =
"spark.connect.session.user"
Review Comment:
This key lives under the `spark.connect.session.*` prefix, which is
currently the namespace for the static confs in `Connect.scala` (planCache,
planCompression, manager.*, all `buildStaticConf`/`buildConf`), yet it's
actually a runtime local property. I get that the name can't change (the PR
says Kyuubi's Ranger plugin will fall back to reading
`spark.connect.session.user`, so it's an agreed contract), so I'd suggest a
scaladoc line on this constant stating "this is a SparkContext local property,
not a SQLConf; Ranger/Kyuubi depends on this key", so nobody later mistakes it
for something settable via `spark.conf.set`. It currently sits right next to
the heavily-commented keepalive constant but has no comment of its own.
##########
sql/connect/server/src/main/scala/org/apache/spark/sql/connect/execution/ExecuteThreadRunner.scala:
##########
@@ -219,6 +219,12 @@ private[connect] class ExecuteThreadRunner(executeHolder:
ExecuteHolder) extends
s"Spark Connect - ${Utils.abbreviate(debugString, 128)}")
session.sparkContext.setInterruptOnCancel(true)
+ val sessionUser = executeHolder.sessionHolder.userId
+ if (sessionUser != null && sessionUser.nonEmpty) {
Review Comment:
+1
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
