dongjoon-hyun commented on code in PR #58737:
URL: https://github.com/apache/spark/pull/58737#discussion_r3993543635
##########
core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala:
##########
@@ -137,6 +143,48 @@ private[spark] class LocalSchedulerBackend(
SparkHadoopUtil.get.addDelegationTokens(tokens, conf)
}
+ /**
+ * Start the UserCredentialManager if OIDC credential propagation is
enabled, mirroring
+ * CoarseGrainedSchedulerBackend. Runs independently of
Kerberos/HadoopDelegationTokenManager.
+ *
+ * In local mode the driver and the single executor share this JVM and the
same
+ * `SparkEnv.get.userCredentials`, so the propagation callback simply
updates that reference
+ * (there is no remote executor to message); the in-JVM Executor picks up
credentials from the
+ * same store via TaskDescription. Driver-side filesystem access uses the
provider wiring that
+ * the selection phase (UserCredentialManager.applyProviderProperties)
already applied to the
+ * driver's Hadoop Configuration.
+ */
+ private def setupUserCredentialManager(): Unit = {
+ // Capture this backend's SparkEnv once, rather than looking up the global
SparkEnv.get on
+ // every callback. stop() only waits a bounded time for the renewal
thread, so a renewal that
+ // outlives this SparkContext must not write into a *different*
SparkContext's credential
+ // store (e.g. a new context created in the same JVM by a notebook, test,
or Spark Connect
+ // session). Binding to this env ensures a late renewal updates only this
application's store,
+ // which is harmless once this env is stopped.
(CoarseGrainedSchedulerBackend avoids the issue
+ // differently, by routing updates through its own already-stopped
driverEndpoint.)
+ val env = SparkEnv.get
+ // Reuse the loader from SparkContext's selection phase (Some when OIDC is
enabled, None
+ // otherwise). Passing the Option straight through keeps SparkContext as
the single owner of
+ // the loader: create() enforces that an enabled configuration has a
loader rather than
+ // silently allocating one here that no one would close.
+ userCredentialManager = UserCredentialManager.create(conf, { (version,
credentials) =>
Review Comment:
`conf` here is `sc.getConf`, i.e. a clone
(`SparkContext.createTaskScheduler` passes `sc.getConf` to
`LocalSchedulerBackend`), while `CoarseGrainedSchedulerBackend` binds the
manager to `scheduler.sc.conf`. `UserCredentialManager.start()` runs
`resolveCredentials(ctx, applyProperties = true)`, whose fallback writes
provider-declared `spark.*` keys into `sparkConf`. In local mode those writes
land in the clone and never reach `sc.conf`, so the fallback silently does
nothing here while it works in cluster mode. Shall we pass `scheduler.sc.conf`
for parity?
##########
core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala:
##########
@@ -137,6 +143,48 @@ private[spark] class LocalSchedulerBackend(
SparkHadoopUtil.get.addDelegationTokens(tokens, conf)
}
+ /**
+ * Start the UserCredentialManager if OIDC credential propagation is
enabled, mirroring
+ * CoarseGrainedSchedulerBackend. Runs independently of
Kerberos/HadoopDelegationTokenManager.
+ *
+ * In local mode the driver and the single executor share this JVM and the
same
+ * `SparkEnv.get.userCredentials`, so the propagation callback simply
updates that reference
+ * (there is no remote executor to message); the in-JVM Executor picks up
credentials from the
+ * same store via TaskDescription. Driver-side filesystem access uses the
provider wiring that
+ * the selection phase (UserCredentialManager.applyProviderProperties)
already applied to the
+ * driver's Hadoop Configuration.
+ */
+ private def setupUserCredentialManager(): Unit = {
+ // Capture this backend's SparkEnv once, rather than looking up the global
SparkEnv.get on
+ // every callback. stop() only waits a bounded time for the renewal
thread, so a renewal that
+ // outlives this SparkContext must not write into a *different*
SparkContext's credential
+ // store (e.g. a new context created in the same JVM by a notebook, test,
or Spark Connect
+ // session). Binding to this env ensures a late renewal updates only this
application's store,
+ // which is harmless once this env is stopped.
(CoarseGrainedSchedulerBackend avoids the issue
+ // differently, by routing updates through its own already-stopped
driverEndpoint.)
+ val env = SparkEnv.get
Review Comment:
Since the goal is to bind to this backend's own env, `scheduler.sc.env`
expresses that directly instead of depending on what the global `SparkEnv.get`
points at when `start()` runs. Also, the callback only needs the store:
capturing `val store = scheduler.sc.env.userCredentials` rather than the whole
`SparkEnv` avoids pinning the stopped context's
BlockManager/MemoryManager/RpcEnv if a renewal outlives the 10s
`awaitTermination` in `stop()`.
Minor: CGSB stops the manager before `StopDriver`, so a late renewal there
still reaches a live `DriverEndpoint` (which reads the global `SparkEnv.get`).
The comment's claim that CGSB avoids the issue that way is not quite accurate.
##########
core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala:
##########
@@ -742,13 +743,80 @@ class UserCredentialManagerSuite extends SparkFunSuite {
assert(conf.get("spark.fake.credentials.enabled") === "true")
}
+ test("selection then resolution reuse one loader and initialize the provider
exactly once") {
+ // Ordering invariant behind the driver-side fix: the selection phase
+ // (applyProviderProperties) selects the provider WITHOUT init() and
applies its declared
+ // properties, then the resolution phase (start(), via
UserCredentialManager.create) reuses
+ // the SAME loader so the provider is initialized exactly once. This
mirrors what
+ // SparkContext (selection) and the scheduler backend (resolution) do at
runtime -- including
+ // LocalSchedulerBackend now that local mode runs a resolution phase.
+ val conf = createSparkConf()
+ conf.set("spark.security.oidc.provider.fake",
+ "org.apache.spark.security.FakeCredentialProvider")
+
+ // Selection phase: applies properties and returns the loader to reuse.
+ val loaderOpt = UserCredentialManager.applyProviderProperties(conf)
+ assert(loaderOpt.isDefined)
+ val loader = loaderOpt.get
+ assert(conf.get("spark.hadoop.fs.fake.credentials.provider") ===
+ "org.apache.spark.security.FakeExecutorCredentialProvider")
+
+ // Selection must NOT have initialized the provider
(selectProviderForProperties skips init).
+ val confMap = conf.getAll
+ .filter { case (k, _) => k.startsWith("spark.security.oidc.") }
+ .toMap.asJava
+ // Observe the SAME provider instance the loader caches, without
initializing it, and assert
+ // the selection phase left it uninitialized.
+ val provider = loader.selectProviderForProperties("fake", confMap).get()
+ .asInstanceOf[FakeCredentialProvider]
+ assert(provider.getInitCount === 0,
+ "the selection phase must not initialize the provider")
+ // The first providerFor() call (resolution path) performs the single
init().
+ val resolved = loader.providerFor("fake", confMap).get()
Review Comment:
This `providerFor` call initializes the provider before `manager.start()`,
so the `getInitCount === 1` assertion after `start()` only checks that
`CredentialProviderLoader` is idempotent, not that `create()`/`start()` reused
the selection-phase loader. `initCount` is per instance: if `create()`
regressed to allocating a fresh loader, `start()` would initialize a different
`FakeCredentialProvider` and this test would still pass. Dropping this call
(keeping `selectProviderForProperties` + `=== 0`) and asserting `=== 1` only
after `start()` makes the test prove the invariant in its name. Also consider
`loader.closeAll()` in `finally` so the initialized provider is closed.
##########
core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala:
##########
@@ -137,6 +143,48 @@ private[spark] class LocalSchedulerBackend(
SparkHadoopUtil.get.addDelegationTokens(tokens, conf)
}
+ /**
+ * Start the UserCredentialManager if OIDC credential propagation is
enabled, mirroring
+ * CoarseGrainedSchedulerBackend. Runs independently of
Kerberos/HadoopDelegationTokenManager.
+ *
+ * In local mode the driver and the single executor share this JVM and the
same
+ * `SparkEnv.get.userCredentials`, so the propagation callback simply
updates that reference
+ * (there is no remote executor to message); the in-JVM Executor picks up
credentials from the
+ * same store via TaskDescription. Driver-side filesystem access uses the
provider wiring that
+ * the selection phase (UserCredentialManager.applyProviderProperties)
already applied to the
+ * driver's Hadoop Configuration.
+ */
+ private def setupUserCredentialManager(): Unit = {
Review Comment:
This is a near-verbatim copy of
`CoarseGrainedSchedulerBackend.setupUserCredentialManager`/`stopUserCredentialManager`,
and the two already differ (captured env vs. global `SparkEnv.get`). Since the
sibling `HadoopDelegationTokenManager` lifecycle lives once in
`SupportsDelegationToken` for both backends, how about hoisting
`userCredentialManager`, `setupUserCredentialManager()` and
`stopUserCredentialManager()` there with a single `protected def
propagateUserCredentials(version: Long, credentials: Array[Byte]): Unit` hook?
CGSB would implement it via `driverEndpoint.send(UpdateUserCredentials(...))`
and this backend via the direct store write.
##########
core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala:
##########
@@ -197,8 +246,20 @@ private[spark] class LocalSchedulerBackend(
}
private def stop(finalState: SparkAppHandle.State): Unit = {
- localEndpoint.ask(StopExecutor)
- stopTokenManager()
+ // Ensure both managers are always stopped, even if stopping the executor
endpoint throws.
+ // The UserCredentialManager renewal thread must be shut down before
SparkContext.stop()
+ // closes the shared CredentialProviderLoader, otherwise a renewal task
could race against an
+ // already-closed loader. Each step is isolated so that a failure in one
does not skip the
+ // others (mirrors CoarseGrainedSchedulerBackend.stop, which stops the
managers in a finally).
+ Utils.tryLogNonFatalError {
+ localEndpoint.ask(StopExecutor)
Review Comment:
`ask` is non-blocking and turns failures into a failed `Future`, so the only
synchronous exception this wrapper can catch is the NPE when `localEndpoint` is
still null, i.e. a launcher stop request arriving before `start()`
(`launcherBackend.connect()` runs in the constructor and `onStopRequest` fires
on its own thread). A null guard like CGSB's `if (driverEndpoint != null)` plus
`try { ... } finally { stopTokenManager(); stopUserCredentialManager() }` would
state the invariant instead of logging an NPE.
Relatedly, if `stop(KILLED)` lands before `manager.start()`,
`stopUserCredentialManager()` is a no-op and `start()` still brings up the
renewal thread on a killed app until `SparkContext.stop()`. A `stopped` flag
checked in `start()` would close that. The shape is pre-existing for the token
manager, but it now involves a thread doing network I/O.
##########
core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala:
##########
@@ -137,6 +143,48 @@ private[spark] class LocalSchedulerBackend(
SparkHadoopUtil.get.addDelegationTokens(tokens, conf)
}
+ /**
+ * Start the UserCredentialManager if OIDC credential propagation is
enabled, mirroring
+ * CoarseGrainedSchedulerBackend. Runs independently of
Kerberos/HadoopDelegationTokenManager.
+ *
+ * In local mode the driver and the single executor share this JVM and the
same
+ * `SparkEnv.get.userCredentials`, so the propagation callback simply
updates that reference
+ * (there is no remote executor to message); the in-JVM Executor picks up
credentials from the
+ * same store via TaskDescription. Driver-side filesystem access uses the
provider wiring that
+ * the selection phase (UserCredentialManager.applyProviderProperties)
already applied to the
+ * driver's Hadoop Configuration.
+ */
+ private def setupUserCredentialManager(): Unit = {
+ // Capture this backend's SparkEnv once, rather than looking up the global
SparkEnv.get on
+ // every callback. stop() only waits a bounded time for the renewal
thread, so a renewal that
+ // outlives this SparkContext must not write into a *different*
SparkContext's credential
+ // store (e.g. a new context created in the same JVM by a notebook, test,
or Spark Connect
+ // session). Binding to this env ensures a late renewal updates only this
application's store,
+ // which is harmless once this env is stopped.
(CoarseGrainedSchedulerBackend avoids the issue
+ // differently, by routing updates through its own already-stopped
driverEndpoint.)
+ val env = SparkEnv.get
+ // Reuse the loader from SparkContext's selection phase (Some when OIDC is
enabled, None
+ // otherwise). Passing the Option straight through keeps SparkContext as
the single owner of
+ // the loader: create() enforces that an enabled configuration has a
loader rather than
+ // silently allocating one here that no one would close.
+ userCredentialManager = UserCredentialManager.create(conf, { (version,
credentials) =>
+ // No remote executors in local mode; update the shared credential store
directly so that
+ // subsequently dispatched tasks (and driver-side access) observe the
new credentials.
+ VersionedCredentials.updateIfNewer(env.userCredentials, version,
credentials)
+ }, scheduler.sc.userCredentialProviderLoader)
+ userCredentialManager.foreach { manager =>
+ val (version, initialCredentials) = manager.start()
+ // Store initial credentials synchronously so they are available for
TaskDescription
+ // (task dispatch) immediately. The onCredentialsUpdate callback above
also runs the same
+ // updateIfNewer, so this is idempotent.
+ VersionedCredentials.updateIfNewer(env.userCredentials, version,
initialCredentials)
Review Comment:
`UserCredentialManager.start()` invokes `onCredentialsUpdate` synchronously
before returning, and the callback above is itself `updateIfNewer` on the same
store, so this second `updateIfNewer` is always a no-op here. The rationale in
CGSB applies only because its callback goes through an async
`driverEndpoint.send`. `userCredentialManager.foreach(_.start())` would be
enough, with a note that the callback performs the synchronous store.
##########
core/src/test/scala/org/apache/spark/scheduler/local/LocalSchedulerBackendSuite.scala:
##########
@@ -0,0 +1,128 @@
+/*
+ * 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.scheduler.local
+
+import java.io.File
+import java.nio.charset.StandardCharsets
+import java.nio.file.Files
+import java.time.Instant
+import java.util.Base64
+
+import org.apache.spark.{LocalSparkContext, SparkConf, SparkContext, SparkEnv,
SparkFunSuite}
+import org.apache.spark.deploy.security.UserCredentialManager
+import org.apache.spark.internal.config._
+
+/**
+ * Tests that [[LocalSchedulerBackend]] starts a `UserCredentialManager` in
local mode when OIDC
+ * credential propagation is enabled, for parity with
`HadoopDelegationTokenManager` (which
+ * `LocalSchedulerBackend` already runs via `createTokenManager()`). This is
the SPARK-59296
+ * follow-up: before it, OIDC was a no-op in local mode.
+ */
+class LocalSchedulerBackendSuite extends SparkFunSuite with LocalSparkContext {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ // A real (unsigned) JWT with the claims FileTokenIngestor requires: sub +
iss (+ exp).
+ Files.write(tokenFile.toPath, makeJwt().getBytes(StandardCharsets.UTF_8))
+ }
+
+ override def afterEach(): Unit = {
+ try {
+ if (tokenFile != null) tokenFile.delete()
+ } finally {
+ super.afterEach()
+ }
+ }
+
+ /** Build a minimal unsigned JWT (header.payload) that FileTokenIngestor can
parse. */
+ private def makeJwt(): String = {
Review Comment:
The same unsigned-JWT builder is added inline in
`UserCredentialManagerSuite` too, and `FileTokenIngestorSuite` already builds
these with jjwt
(`Jwts.builder().subject(..).issuer(..).expiration(..).compact()`), which is on
core's test classpath. One shared helper (or `Jwts.builder()` inline) would
avoid a third hand-rolled copy. Likewise `withTempDir` from `SparkFunSuite`
would replace the `createTempFile`/`deleteOnExit`/`afterEach` plumbing. For the
new `UserCredentialManagerSuite` test, the 4-arg constructor with
`createIngestor(createUserContext())` needs no token file at all.
--
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]
