[SPARK-59296][CORE][FOLLOWUP] Run OIDC UserCredentialManager in local mode for parity with HadoopDelegationTokenManager - #58737
Conversation
| isLocal: Boolean): Option[CredentialProviderLoader] = { | ||
| if (!sparkConf.get(SECURITY_OIDC_ENABLED) || isLocal) { | ||
| sparkConf: SparkConf): Option[CredentialProviderLoader] = { | ||
| if (!sparkConf.get(SECURITY_OIDC_ENABLED)) { |
There was a problem hiding this comment.
With the isLocal guard removed, local mode now wires the provider into the driver's Hadoop Configuration here, but credentials are acquired only later, in LocalSchedulerBackend.start() (via _taskScheduler.start()). Driver-side FS access that happens before that point in SparkContext initialization sees an empty SparkEnv.userCredentials:
setCheckpointDir(spark.checkpoint.dir) andaddFile(spark.files/spark.archives) throw, soSparkContextconstruction fails.addJar(spark.jars) swallows the error, so the jar is silently dropped and tasks later fail withClassNotFoundException.
For example, local[*] + spark.security.oidc.enabled=true + the AWS provider + spark.checkpoint.dir=s3a://bucket/ckpt started fine before this PR (default credential chain), but fails after it.
This is the same limitation that already exists in cluster mode, but this PR extends it to local mode, and the new scaladoc ("points at credentials that are actually populated") is not accurate for this window. Could you document this, or make sure the resolution phase runs before these accesses in local mode?
There was a problem hiding this comment.
I've documented it rather than resolving earlier in local mode, since moving resolution up would
diverge local from cluster and break the selection-vs-resolution separation (resolution does I/O and
must stay late).
| 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(SparkEnv.get.userCredentials, version, credentials) |
There was a problem hiding this comment.
This callback looks up the global SparkEnv.get every time it fires, instead of the env this backend belongs to.
UserCredentialManager.stop() waits only 10 seconds for the renewal thread. If a renewal is stuck longer than that (e.g., a slow STS call that ignores interrupts) and a new SparkContext is created in the same JVM (notebooks, tests, Spark Connect session restart), the late renewal writes the old application's credentials (version N >= 2) into the new application's store. Since the new application's version counter restarts at 1, updateIfNewer then rejects its own renewals until its version exceeds N, so it keeps using the previous application's (possibly different principal's) credentials.
CoarseGrainedSchedulerBackend doesn't have this issue because it routes the update through its own (already stopped) driverEndpoint. Capturing the env once in setupUserCredentialManager() (e.g., val env = SparkEnv.get) and using it in both the callback and the initial store would avoid this.
There was a problem hiding this comment.
Thanks, setupUserCredentialManager() now captures val env = SparkEnv.get once and uses it in both the callback and the initial store, so a late renewal can only touch this (stopped) application's store, never a new context's.
dongjoon-hyun
left a comment
dongjoon-hyun
left a comment
There was a problem hiding this comment.
Thank you for the follow-up, @sarutak. A few comments inline. Two more that do not attach to a changed line:
- The
UserCredentialManagerclass scaladoc (line 54) still saysIntended to be started from CoarseGrainedSchedulerBackend.start(). It should mentionLocalSchedulerBackendnow as well. - The renamed test
applyProviderProperties applies properties regardless of local modeis now substantively identical toapplyProviderProperties applies additionalSparkProperties from selected providers(same conf, same asserts). We could drop one of them.
| // 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) => |
There was a problem hiding this 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?
| // 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 |
There was a problem hiding this 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.
| // 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) |
There was a problem hiding this 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.
| // 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) |
There was a problem hiding this 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.
| * the selection phase (UserCredentialManager.applyProviderProperties) already applied to the | ||
| * driver's Hadoop Configuration. | ||
| */ | ||
| private def setupUserCredentialManager(): Unit = { |
There was a problem hiding this 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.
| 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() |
There was a problem hiding this 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.
| } | ||
|
|
||
| /** Build a minimal unsigned JWT (header.payload) that FileTokenIngestor can parse. */ | ||
| private def makeJwt(): String = { |
There was a problem hiding this 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.
…ntialManager lifecycle into SupportsDelegationToken and tighten local-mode wiring Addresses the second round of review on PR apache#58737: - Hoist the UserCredentialManager lifecycle (field + setup/stopUserCredentialManager) into SupportsDelegationToken, next to the sibling HadoopDelegationTokenManager, so both CoarseGrainedSchedulerBackend and LocalSchedulerBackend share one implementation. The two backends now differ only via a `propagateUserCredentials(version, credentials)` hook: CGSB updates the driver store then broadcasts via its DriverEndpoint; LocalSchedulerBackend (no remote executors) updates the shared store directly. The trait gains an abstract `scheduler` accessor (both backends already hold `scheduler`). - Bind the manager to `scheduler.sc.conf` (the live SparkConf) rather than the constructor `conf`. LocalSchedulerBackend receives `sc.getConf`, a clone; the resolution-time fallback in UserCredentialManager.start() (which applies provider-declared spark.* keys for schemes the selection phase could not) wrote into that clone and never reached sc.conf, so it silently did nothing in local mode while working in cluster mode. Using scheduler.sc.conf restores parity. - Use `scheduler.sc.env.userCredentials` (this backend's own env store) instead of the global SparkEnv.get, so a renewal that outlives this SparkContext cannot write into a different SparkContext's store created later in the same JVM. - Drop the now-redundant separate initial synchronous store: start() invokes the callback synchronously, and for both backends the callback updates the store, so foreach(_.start()) is enough (CGSB's callback still updates the store synchronously before the async broadcast, keeping the no-null-window guarantee). - LocalSchedulerBackend.stop(): guard localEndpoint (ask() only throws synchronously when localEndpoint is still null, i.e. a launcher stop before start()) and stop the managers in a finally, matching CGSB. Add a `stopped` flag checked in start() so a stop(KILLED) arriving before start() does not bring up the renewal thread (network I/O) on an already-killed app. - Tests/docs: strengthen the selection/resolution ordering test to assert init count 0 after selection and 1 only after start() (proving loader reuse, not just loader idempotency), using the test-only 4-arg constructor and a mock ingestor (no token file) with loader.closeAll() in finally; build JWTs with jjwt and use withTempDir in LocalSchedulerBackendSuite; drop a duplicate applyProviderProperties test; and note LocalSchedulerBackend in the UserCredentialManager class scaladoc.
… mode for parity with HadoopDelegationTokenManager ### What changes were proposed in this pull request? Follow-up to SPARK-59296. Make OIDC credential propagation active in local mode, for parity with `HadoopDelegationTokenManager` (which already runs in `LocalSchedulerBackend` via `createTokenManager()`). - `LocalSchedulerBackend` now starts a `UserCredentialManager` when `spark.security.oidc.enabled=true`, mirroring `CoarseGrainedSchedulerBackend`. It is started in `start()` (after the token manager) and stopped in `stop()`. Both managers are stopped independently (each wrapped in `Utils.tryLogNonFatalError`) so a failure in one does not skip the other, and the renewal thread is shut down before `SparkContext.stop()` closes the shared `CredentialProviderLoader`. In local mode the driver and the single in-JVM executor share `SparkEnv.get.userCredentials`, so the propagation callback updates that store directly (there is no remote executor to message); tasks pick the credentials up via `TaskDescription`. - The provider selection phase (`UserCredentialManager.applyProviderProperties`) now runs in local mode too. SPARK-59296 skipped it in local mode because no resolution phase followed there; now that `LocalSchedulerBackend` runs a resolution phase, the wiring it applies to the driver's Hadoop `Configuration` points at credentials that are actually populated. The `isLocal` parameter of `applyProviderProperties` (whose only purpose was that skip) is removed; the `SparkContext` call site and the surrounding scaladoc/comments are updated accordingly. ### Why are the changes needed? Before this change, `spark.security.oidc.enabled=true` had no effect in local mode: no `UserCredentialManager` was started and the selection phase was skipped. `HadoopDelegationTokenManager`, by contrast, runs in `LocalSchedulerBackend`, so Kerberos-based credential acquisition already works in local mode. This closes that parity gap so a local-mode driver acquires and uses OIDC-derived credentials for its own storage access (and renews them). ### Does this PR introduce _any_ user-facing change? Yes (behavior only; no public API added or removed, and the feature is unreleased). With OIDC enabled in local mode, the driver now acquires OIDC credentials and applies provider-declared properties, instead of the feature being a no-op. As in cluster mode, initial acquisition is fail-fast: if OIDC is enabled but the identity token file is missing or malformed, `SparkContext` startup fails rather than running with no credentials. ### How was this patch tested? - New `LocalSchedulerBackendSuite` (real `SparkContext` in `local[1]`): with OIDC enabled, the backend is a `LocalSchedulerBackend`, selection wires provider-declared properties into the driver's Hadoop `Configuration` (and non-Hadoop `spark.*`), a loader is retained on `SparkContext`, and resolution populates `SparkEnv.get.userCredentials`; with OIDC disabled, all are no-ops. - New `UserCredentialManagerSuite` test proving the selection -> resolution ordering invariant: selection selects without `init()`, and reusing the same loader for resolution initializes the provider exactly once. The "no-op in local mode" test is replaced by one asserting selection now applies properties regardless of local mode. - `UserCredentialManagerSuite`, `OidcCredentialIntegrationSuite`, and `LocalSchedulerBackendSuite` pass (49 tests). `SparkContextSuite` passes (84 tests). `dev/lint-scala` and `dev/lint-java` pass. ### Was this patch authored or co-authored using generative AI tooling? Yes.
…allback to this SparkEnv and document the driver-side early-startup window
Addresses review feedback on the local-mode parity change:
- LocalSchedulerBackend.setupUserCredentialManager now captures this backend's SparkEnv once and
uses it in both the propagation callback and the initial credential store, instead of 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 could otherwise write this application's
credentials into a different SparkContext's store created later in the same JVM (notebook, test,
or Spark Connect session), where the restarted version counter would then reject the new
application's own renewals. Binding to this env confines a late renewal to this (stopped)
application's store.
- Correct the applyProviderProperties scaladoc: running the selection phase in local mode extends
the same driver-side early-startup window that already exists in cluster mode. Provider wiring is
applied during SparkContext construction but credentials are not resolved until scheduler-backend
start, so driver-side access to a wired scheme during construction (spark.jars / spark.files /
spark.archives / spark.checkpoint.dir on e.g. s3a://) runs before credentials exist. The earlier
wording ("points at credentials that are actually populated") was inaccurate for this window. The
OIDC security docs (separate docs PR) document this for local mode alongside cluster mode.
This keeps credential resolution late (parity with HadoopDelegationTokenManager and with cluster
mode); moving resolution earlier in local mode only would diverge local from cluster and break that
design intent, so the window is documented rather than closed.
…ntialManager lifecycle into SupportsDelegationToken and tighten local-mode wiring Addresses the second round of review on PR apache#58737: - Hoist the UserCredentialManager lifecycle (field + setup/stopUserCredentialManager) into SupportsDelegationToken, next to the sibling HadoopDelegationTokenManager, so both CoarseGrainedSchedulerBackend and LocalSchedulerBackend share one implementation. The two backends now differ only via a `propagateUserCredentials(version, credentials)` hook: CGSB updates the driver store then broadcasts via its DriverEndpoint; LocalSchedulerBackend (no remote executors) updates the shared store directly. The trait gains an abstract `scheduler` accessor (both backends already hold `scheduler`). - Bind the manager to `scheduler.sc.conf` (the live SparkConf) rather than the constructor `conf`. LocalSchedulerBackend receives `sc.getConf`, a clone; the resolution-time fallback in UserCredentialManager.start() (which applies provider-declared spark.* keys for schemes the selection phase could not) wrote into that clone and never reached sc.conf, so it silently did nothing in local mode while working in cluster mode. Using scheduler.sc.conf restores parity. - Use `scheduler.sc.env.userCredentials` (this backend's own env store) instead of the global SparkEnv.get, so a renewal that outlives this SparkContext cannot write into a different SparkContext's store created later in the same JVM. - Drop the now-redundant separate initial synchronous store: start() invokes the callback synchronously, and for both backends the callback updates the store, so foreach(_.start()) is enough (CGSB's callback still updates the store synchronously before the async broadcast, keeping the no-null-window guarantee). - LocalSchedulerBackend.stop(): guard localEndpoint (ask() only throws synchronously when localEndpoint is still null, i.e. a launcher stop before start()) and stop the managers in a finally, matching CGSB. Add a `stopped` flag checked in start() so a stop(KILLED) arriving before start() does not bring up the renewal thread (network I/O) on an already-killed app. - Tests/docs: strengthen the selection/resolution ordering test to assert init count 0 after selection and 1 only after start() (proving loader reuse, not just loader idempotency), using the test-only 4-arg constructor and a mock ingestor (no token file) with loader.closeAll() in finally; build JWTs with jjwt and use withTempDir in LocalSchedulerBackendSuite; drop a duplicate applyProviderProperties test; and note LocalSchedulerBackend in the UserCredentialManager class scaladoc.
1b1ced8 to
935b440
Compare
2 participants
What changes were proposed in this pull request?
Follow-up to SPARK-59296. Make OIDC credential propagation active in local mode, for parity with
HadoopDelegationTokenManager(which already runs inLocalSchedulerBackendviacreateTokenManager()).LocalSchedulerBackendnow starts aUserCredentialManagerwhenspark.security.oidc.enabled=true, mirroringCoarseGrainedSchedulerBackend. It is started instart()(after the token manager) and stopped instop(). Both the token manager and the user-credential manager are stopped independently (each wrapped inUtils.tryLogNonFatalError) so a failure in one does not skip the other, and the renewal thread is shut down beforeSparkContext.stop()closes the sharedCredentialProviderLoader. In local mode the driver and the single in-JVM executor shareSparkEnv.get.userCredentials, so the propagation callback updates that store directly (there is no remote executor to message); tasks pick the credentials up viaTaskDescription.UserCredentialManager.applyProviderProperties) now runs in local mode too. SPARK-59296 skipped it in local mode because no resolution phase followed there; now thatLocalSchedulerBackendruns a resolution phase, the wiring it applies to the driver's HadoopConfigurationpoints at credentials that are actually populated. TheisLocalparameter ofapplyProviderProperties(whose only purpose was that skip) is removed; theSparkContextcall site and the surrounding scaladoc/comments are updated accordingly.Why are the changes needed?
Before this change,
spark.security.oidc.enabled=truehad no effect in local mode: noUserCredentialManagerwas started, so OIDC credentials were never acquired, and the selection phase was skipped so no provider wiring was applied.HadoopDelegationTokenManager, by contrast, runs inLocalSchedulerBackend, so Kerberos-based credential acquisition already works in local mode. This closes that parity gap, so a local-mode driver acquires and uses OIDC-derivedcredentials for its own storage access (and renews them).
Does this PR introduce any user-facing change?
Yes (behavior only; no public API added or removed, and the feature is unreleased). With OIDC enabled in local mode, the driver now acquires OIDC credentials and applies provider-declared properties, instead of the feature being a no-op. As in cluster mode, initial credential acquisition is fail-fast: if OIDC is enabled but the identity token file is missing or malformed,
SparkContextstartup fails rather than running with no credentials.How was this patch tested?
LocalSchedulerBackendSuite(realSparkContextinlocal[1]): with OIDC enabled, the scheduler backend is aLocalSchedulerBackend, the selection phase wires the provider-declared properties into the driver's HadoopConfiguration(and non-Hadoopspark.*too), a loader is retained onSparkContext, and the resolution phase populatesSparkEnv.get.userCredentials; with OIDC disabled, all of these are no-ops.UserCredentialManagerSuitetest proving the selection -> resolution ordering invariant: the selection phase does not initialize the provider (getInitCountis 0 afterselectProviderForProperties), and reusing the same loader for resolution initializes the provider exactly once (getInitCountbecomes 1, and the loader returns the same cachedinstance). The "no-op in local mode" test is replaced by one asserting selection now applies properties regardless of local mode.
UserCredentialManagerSuite,OidcCredentialIntegrationSuite, andLocalSchedulerBackendSuitepass (49 tests).SparkContextSuitepasses (84 tests) as a regression check on the modifiedSparkContextinit path.dev/lint-scalaanddev/lint-javapass.Was this patch authored or co-authored using generative AI tooling?
Kiro CLI / Claude