From 93cb8c377ce02603e10376b5b20fdffa80148421 Mon Sep 17 00:00:00 2001 From: Kousuke Saruta Date: Fri, 11 Sep 2026 02:00:45 +0000 Subject: [PATCH 1/3] [SPARK-59296][CORE][FOLLOWUP] Run OIDC UserCredentialManager in local 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. --- .../scala/org/apache/spark/SparkContext.scala | 28 ++-- .../security/UserCredentialManager.scala | 37 +++-- .../CoarseGrainedSchedulerBackend.scala | 8 +- .../local/LocalSchedulerBackend.scala | 62 ++++++++- .../security/UserCredentialManagerSuite.scala | 100 ++++++++++++-- .../local/LocalSchedulerBackendSuite.scala | 128 ++++++++++++++++++ 6 files changed, 307 insertions(+), 56 deletions(-) create mode 100644 core/src/test/scala/org/apache/spark/scheduler/local/LocalSchedulerBackendSuite.scala diff --git a/core/src/main/scala/org/apache/spark/SparkContext.scala b/core/src/main/scala/org/apache/spark/SparkContext.scala index 5c8c30e2807c7..e1a1de5b865df 100644 --- a/core/src/main/scala/org/apache/spark/SparkContext.scala +++ b/core/src/main/scala/org/apache/spark/SparkContext.scala @@ -228,7 +228,8 @@ class SparkContext(config: SparkConf) extends Logging { // The CredentialProviderLoader created by the OIDC selection phase (applyProviderProperties), // retained so the later credential resolution phase (UserCredentialManager, started by the // scheduler backend) reuses the same loader. None when OIDC credential propagation is - // disabled or in local mode (the selection phase is skipped and allocates no loader). + // disabled (the selection phase is skipped and allocates no loader). The selection phase runs + // in local mode as well, since LocalSchedulerBackend now starts a resolution phase. // SparkContext is the single owner of this loader and is responsible for closing it in stop(). private var _userCredentialProviderLoader: Option[CredentialProviderLoader] = None private var _executorMemory: Int = _ @@ -348,7 +349,7 @@ class SparkContext(config: SparkConf) extends Logging { // The CredentialProviderLoader from the OIDC selection phase, reused by the credential // resolution phase so providers are initialized exactly once. `None` when OIDC credential - // propagation is disabled, in local mode, or before initialization. Internal. + // propagation is disabled or before initialization. Internal. private[spark] def userCredentialProviderLoader: Option[CredentialProviderLoader] = _userCredentialProviderLoader @@ -448,17 +449,18 @@ class SparkContext(config: SparkConf) extends Logging { // This should be set as early as possible. SparkContext.enableMagicCommitterIfNeeded(_conf) - // OIDC credential propagation: provider SELECTION phase. When enabled (and not in local - // mode), discover the credential provider(s) for the configured scheme(s) and apply their - // declared Spark properties (e.g. the S3A credentials provider class) into _conf, so that - // the driver's Hadoop Configuration built later -- and other config-derived components -- - // pick them up. This is done here, at the "as early as possible" slot, so the applied keys - // are visible to the spark.logConf dump and to any Hadoop Configuration built during - // createSparkEnv (e.g. by SecurityManager). It performs no credential resolution and no - // network I/O (providers are selected without init()); actual acquisition happens later in - // the scheduler backend (UserCredentialManager). Any returned loader is retained so the - // resolution phase reuses it, and SparkContext closes it in stop(). - _userCredentialProviderLoader = UserCredentialManager.applyProviderProperties(_conf, isLocal) + // OIDC credential propagation: provider SELECTION phase. When enabled, discover the + // credential provider(s) for the configured scheme(s) and apply their declared Spark + // properties (e.g. the S3A credentials provider class) into _conf, so that the driver's + // Hadoop Configuration built later -- and other config-derived components -- pick them up. + // This is done here, at the "as early as possible" slot, so the applied keys are visible to + // the spark.logConf dump and to any Hadoop Configuration built during createSparkEnv (e.g. + // by SecurityManager). It performs no credential resolution and no network I/O (providers + // are selected without init()); actual acquisition happens later in the scheduler backend + // (UserCredentialManager) -- including in local mode, where LocalSchedulerBackend now starts + // a resolution phase. Any returned loader is retained so the resolution phase reuses it, and + // SparkContext closes it in stop(). + _userCredentialProviderLoader = UserCredentialManager.applyProviderProperties(_conf) SparkContext.supplementJavaModuleOptions(_conf) SparkContext.supplementJavaIPv6Options(_conf) diff --git a/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala b/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala index 77ab05ac21d26..3cee7c46b1712 100644 --- a/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala +++ b/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala @@ -482,11 +482,12 @@ private[spark] object UserCredentialManager extends Logging { * @param onCredentialsUpdate Callback to propagate credentials to executors * @param loader The [[CredentialProviderLoader]] from the selection phase * ([[applyProviderProperties]]), passed as an `Option`. When OIDC is enabled it - * must be `Some` (the selection phase, which runs earlier and is skipped only - * when OIDC is disabled or in local mode, produced it); reusing that same - * instance ensures providers are discovered and initialized exactly once, so the - * resolution phase reuses the already-selected providers. It is an error for the - * loader to be `None` while OIDC is enabled and a resolution phase is expected. + * must be `Some` (the selection phase, which runs earlier whenever OIDC is + * enabled -- in local mode as well -- and is skipped only when OIDC is disabled, + * produced it); reusing that same instance ensures providers are discovered and + * initialized exactly once, so the resolution phase reuses the already-selected + * providers. It is an error for the loader to be `None` while OIDC is enabled and + * a resolution phase is expected. * @return Some(manager) if enabled, None otherwise */ def create( @@ -503,7 +504,7 @@ private[spark] object UserCredentialManager extends Logging { "OIDC credential propagation is enabled but no CredentialProviderLoader was produced " + "by the selection phase. This indicates the selection phase " + "(UserCredentialManager.applyProviderProperties) did not run before the resolution " + - "phase, which should not happen outside local mode.") + "phase, which should not happen.") } val tokenFile = sparkConf.get(SECURITY_OIDC_IDENTITY_TOKEN_FILE).getOrElse { throw new IllegalArgumentException( @@ -538,13 +539,12 @@ private[spark] object UserCredentialManager extends Logging { * renewal) happen later in [[start]] on the scheduler backend. This separation of provider * SELECTION from credential RESOLUTION is intentional. * - * The phase is skipped entirely when `isLocal` is true: `LocalSchedulerBackend` does not start - * a [[UserCredentialManager]], so no resolution phase follows and no credentials are ever - * populated. Wiring a provider class into the driver's Hadoop `Configuration` in that case - * would make driver-side access fail (the provider would find no credentials) instead of - * falling back to the default chain. (Running credential resolution in local mode -- for - * parity with `HadoopDelegationTokenManager`, which does run in `LocalSchedulerBackend` -- is - * left to a follow-up.) + * This phase runs in local mode as well: `LocalSchedulerBackend` starts a + * [[UserCredentialManager]] (for parity with `HadoopDelegationTokenManager`, which also runs + * in `LocalSchedulerBackend`), so a resolution phase follows and the wiring applied here + * points at credentials that are actually populated. (Before SPARK-59296's follow-up, local + * mode had no resolution phase, so this selection phase was skipped there to avoid wiring a + * provider whose credentials would never be resolved.) * * Scheme selection is limited to schemes for which a provider is UNAMBIGUOUSLY selected: * either an explicitly-configured scheme (`spark.security.oidc.provider.`) or a @@ -554,17 +554,14 @@ private[spark] object UserCredentialManager extends Logging { * raises a clear error prompting explicit configuration. * * @param sparkConf The Spark configuration to apply properties into. Not modified when OIDC - * credential propagation is disabled or when `isLocal` is true. - * @param isLocal Whether the application runs in local mode (no scheduler backend that starts - * a resolution phase). + * credential propagation is disabled. * @return `Some(loader)` with the [[CredentialProviderLoader]] used, to be passed to * [[create]] so the resolution phase reuses the same loader; `None` when OIDC is - * disabled or when `isLocal` is true (no loader is allocated in those cases). + * disabled (no loader is allocated in that case). */ def applyProviderProperties( - sparkConf: SparkConf, - isLocal: Boolean): Option[CredentialProviderLoader] = { - if (!sparkConf.get(SECURITY_OIDC_ENABLED) || isLocal) { + sparkConf: SparkConf): Option[CredentialProviderLoader] = { + if (!sparkConf.get(SECURITY_OIDC_ENABLED)) { return None } diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala index d56da893f86fc..e03ea10f03b12 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala @@ -1263,10 +1263,10 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp * Called from start(), independently of Kerberos/HadoopDelegationTokenManager. */ private def setupUserCredentialManager(): Unit = { - // Reuse the loader from SparkContext's selection phase (Some when OIDC is enabled and not - // in local mode; 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. + // 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) => // Send to DriverEndpoint to ensure thread-safe access to executorDataMap. // This mirrors HadoopDelegationTokenManager's pattern of sending diff --git a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala index b6a90565aaff4..3ef41cdcd8c45 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala @@ -21,10 +21,10 @@ import java.io.File import java.net.URL import java.nio.ByteBuffer -import org.apache.spark.{SparkConf, SparkContext, SparkEnv, TaskState} +import org.apache.spark.{SparkConf, SparkContext, SparkEnv, TaskState, VersionedCredentials} import org.apache.spark.TaskState.TaskState import org.apache.spark.deploy.SparkHadoopUtil -import org.apache.spark.deploy.security.HadoopDelegationTokenManager +import org.apache.spark.deploy.security.{HadoopDelegationTokenManager, UserCredentialManager} import org.apache.spark.executor.{Executor, ExecutorBackend} import org.apache.spark.internal.{config, Logging, LogKeys} import org.apache.spark.launcher.{LauncherBackend, SparkAppHandle} @@ -120,6 +120,12 @@ private[spark] class LocalSchedulerBackend( private var localEndpoint: RpcEndpointRef = null private val userClassPath = getUserClasspath(conf) private val listenerBus = scheduler.sc.listenerBus + + // UserCredentialManager for OIDC credential propagation (if enabled). Started in start() and + // stopped in stop(), mirroring CoarseGrainedSchedulerBackend so that OIDC credentials are + // acquired and renewed in local mode too, for parity with HadoopDelegationTokenManager (which + // this backend already runs via createTokenManager()). + private var userCredentialManager: Option[UserCredentialManager] = None private val launcherBackend = new LauncherBackend() { override def conf: SparkConf = LocalSchedulerBackend.this.conf override def onStopRequest(): Unit = stop(SparkAppHandle.State.KILLED) @@ -137,6 +143,41 @@ 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 = { + // 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(SparkEnv.get.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( + SparkEnv.get.userCredentials, version, initialCredentials) + } + } + + private def stopUserCredentialManager(): Unit = { + userCredentialManager.foreach(_.stop()) + } + /** * Returns a list of URLs representing the user classpath. * @@ -156,6 +197,7 @@ private[spark] class LocalSchedulerBackend( // call this after localEndpoint is assigned setupTokenManager() + setupUserCredentialManager() listenerBus.post(SparkListenerExecutorAdded( System.currentTimeMillis, @@ -197,8 +239,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) + } + Utils.tryLogNonFatalError { + stopTokenManager() + } + Utils.tryLogNonFatalError { + stopUserCredentialManager() + } try { launcherBackend.setState(finalState) } finally { diff --git a/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala b/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala index a0caf562fa1ad..3224642512d83 100644 --- a/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala @@ -24,6 +24,7 @@ import java.util.concurrent.{CountDownLatch, TimeUnit} import java.util.concurrent.atomic.{AtomicInteger, AtomicReference} import scala.concurrent.duration._ +import scala.jdk.CollectionConverters._ import org.scalatest.concurrent.Eventually.{eventually, timeout} @@ -731,8 +732,8 @@ class UserCredentialManagerSuite extends SparkFunSuite { conf.set("spark.security.oidc.provider.fake", "org.apache.spark.security.FakeCredentialProvider") - val loader = UserCredentialManager.applyProviderProperties(conf, isLocal = false) - assert(loader.isDefined, "a loader should be returned when OIDC is enabled and not local") + val loader = UserCredentialManager.applyProviderProperties(conf) + assert(loader.isDefined, "a loader should be returned when OIDC is enabled") // spark.hadoop.* property is applied ... assert(conf.get("spark.hadoop.fs.fake.credentials.provider") === @@ -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() + .asInstanceOf[FakeCredentialProvider] + assert(resolved eq provider, "loader must reuse the same cached provider instance") + assert(provider.getInitCount === 1, + "providerFor should initialize the provider exactly once") + + // start() goes through the real FileTokenIngestor (via create()), so a parseable token + // file must exist. Write a minimal unsigned JWT with the required sub/iss/exp claims. + val tokenFile = java.io.File.createTempFile("oidc-token-", ".jwt") + tokenFile.deleteOnExit() + val enc = java.util.Base64.getUrlEncoder.withoutPadding() + val header = enc.encodeToString( + """{"alg":"none","typ":"JWT"}""".getBytes(java.nio.charset.StandardCharsets.UTF_8)) + val exp = Instant.now().plusSeconds(300).getEpochSecond + val payload = enc.encodeToString( + s"""{"sub":"test-user","iss":"https://issuer.example.com","exp":$exp}""" + .getBytes(java.nio.charset.StandardCharsets.UTF_8)) + conf.set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath) + java.nio.file.Files.writeString(tokenFile.toPath, s"$header.$payload") + + // Resolution phase reuses the same loader (as UserCredentialManager.create does with the + // loader from the selection phase). start() calls providerFor internally, which must NOT + // re-init the already-initialized provider instance. + val manager = UserCredentialManager.create( + conf, (_: Long, _: Array[Byte]) => (), Some(loader)).get + try { + val (version, bytes) = manager.start() + assert(version === 1L) + val creds = UserCredentialManager.deserializeUserCredentials(bytes) + assert(creds.forScheme("fake").isPresent) + assert(provider.getInitCount === 1, + "reusing the selection-phase loader must not re-initialize the provider") + } finally { + manager.stop() + tokenFile.delete() + } + } + test("applyProviderProperties auto-selects a single-candidate scheme with no explicit config") { // Zero-config path: no spark.security.oidc.provider. is set, so the loader falls // back to discoverAllSchemes(). "fake" has exactly one candidate (FakeCredentialProvider), // so it must be auto-selected and its declared properties applied. val conf = createSparkConf() - val loader = UserCredentialManager.applyProviderProperties(conf, isLocal = false) + val loader = UserCredentialManager.applyProviderProperties(conf) assert(loader.isDefined) assert(conf.get("spark.hadoop.fs.fake.credentials.provider") === "org.apache.spark.security.FakeExecutorCredentialProvider") @@ -758,23 +826,25 @@ class UserCredentialManagerSuite extends SparkFunSuite { test("applyProviderProperties is a no-op and returns None when OIDC is disabled") { val conf = new SparkConf(false) .set(SECURITY_OIDC_ENABLED, false) - val loader = UserCredentialManager.applyProviderProperties(conf, isLocal = false) + val loader = UserCredentialManager.applyProviderProperties(conf) assert(loader.isEmpty, "no loader should be allocated when OIDC is disabled") assert(!conf.contains("spark.hadoop.fs.fake.credentials.provider")) assert(!conf.contains("spark.fake.credentials.enabled")) } - test("applyProviderProperties is a no-op and returns None in local mode") { - // Even with OIDC enabled, local mode has no scheduler backend that starts the resolution - // phase, so wiring a provider would leave driver-side access unable to resolve credentials. - // The selection phase must be skipped and no loader allocated. + test("applyProviderProperties applies properties regardless of local mode") { + // SPARK-59296 follow-up: local mode now starts a resolution phase (LocalSchedulerBackend + // starts a UserCredentialManager), so the selection phase is no longer skipped in local + // mode. There is no longer an isLocal parameter; selection applies properties whenever OIDC + // is enabled, exactly as it does for cluster mode. val conf = createSparkConf() conf.set("spark.security.oidc.provider.fake", "org.apache.spark.security.FakeCredentialProvider") - val loader = UserCredentialManager.applyProviderProperties(conf, isLocal = true) - assert(loader.isEmpty, "no loader should be allocated in local mode") - assert(!conf.contains("spark.hadoop.fs.fake.credentials.provider")) - assert(!conf.contains("spark.fake.credentials.enabled")) + val loader = UserCredentialManager.applyProviderProperties(conf) + assert(loader.isDefined, "a loader should be allocated whenever OIDC is enabled") + assert(conf.get("spark.hadoop.fs.fake.credentials.provider") === + "org.apache.spark.security.FakeExecutorCredentialProvider") + assert(conf.get("spark.fake.credentials.enabled") === "true") } test("applyProviderProperties does not overwrite user-set properties") { @@ -784,7 +854,7 @@ class UserCredentialManagerSuite extends SparkFunSuite { // User explicitly sets the property beforehand. conf.set("spark.hadoop.fs.fake.credentials.provider", "user.Custom") - UserCredentialManager.applyProviderProperties(conf, isLocal = false) + UserCredentialManager.applyProviderProperties(conf) // User-set value must NOT be overwritten; the unset one is still applied. assert(conf.get("spark.hadoop.fs.fake.credentials.provider") === "user.Custom") @@ -798,7 +868,7 @@ class UserCredentialManagerSuite extends SparkFunSuite { val conf = createSparkConf() // Do not set spark.security.oidc.provider.shared -> ambiguous for "shared". // No exception should escape. - UserCredentialManager.applyProviderProperties(conf, isLocal = false) + UserCredentialManager.applyProviderProperties(conf) // Nothing asserted about "shared"; the point is that the call returned normally. } @@ -813,7 +883,7 @@ class UserCredentialManagerSuite extends SparkFunSuite { try { // Must not fail even though AnotherFakeCredentialProvider throws; FakeCredentialProvider's // properties must still be applied (per-provider exception isolation). - UserCredentialManager.applyProviderProperties(conf, isLocal = false) + UserCredentialManager.applyProviderProperties(conf) assert(conf.get("spark.hadoop.fs.fake.credentials.provider") === "org.apache.spark.security.FakeExecutorCredentialProvider") } finally { diff --git a/core/src/test/scala/org/apache/spark/scheduler/local/LocalSchedulerBackendSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/local/LocalSchedulerBackendSuite.scala new file mode 100644 index 0000000000000..a89982dfb0f99 --- /dev/null +++ b/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 = { + val enc = Base64.getUrlEncoder.withoutPadding() + val header = enc.encodeToString( + """{"alg":"none","typ":"JWT"}""".getBytes(StandardCharsets.UTF_8)) + val exp = Instant.now().plusSeconds(300).getEpochSecond + val payload = enc.encodeToString( + s"""{"sub":"test-user","iss":"https://issuer.example.com","exp":$exp}""" + .getBytes(StandardCharsets.UTF_8)) + s"$header.$payload" + } + + private def oidcConf(enabled: Boolean): SparkConf = { + val conf = new SparkConf() + .setMaster("local[1]") + .setAppName("LocalSchedulerBackendSuite") + .set(SECURITY_OIDC_ENABLED, enabled) + // Short intervals so any renewal that fires during the test is cheap. + .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L) + .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L) + if (enabled) { + conf + .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath) + .set("spark.security.oidc.provider.fake", + "org.apache.spark.security.FakeCredentialProvider") + } + conf + } + + test("LocalSchedulerBackend runs OIDC selection and resolution in local mode") { + sc = new SparkContext(oidcConf(enabled = true)) + + // The scheduler backend in local mode is a LocalSchedulerBackend. + assert(sc.schedulerBackend.isInstanceOf[LocalSchedulerBackend], + "local[1] should use LocalSchedulerBackend") + + // Selection phase ran on the driver (SparkContext) even in local mode: the provider's + // declared spark.hadoop.* property reached the driver's Hadoop Configuration (prefix + // stripped), and its non-Hadoop spark.* property reached SparkConf. + assert(sc.hadoopConfiguration.get("fs.fake.credentials.provider") === + "org.apache.spark.security.FakeExecutorCredentialProvider", + "selection phase should wire the provider into the driver's Hadoop Configuration") + assert(sc.getConf.get("spark.fake.credentials.enabled") === "true", + "selection phase should apply non-Hadoop provider properties too") + + // A loader was retained for reuse by the resolution phase. + assert(sc.userCredentialProviderLoader.isDefined, + "SparkContext should retain the selection-phase loader when OIDC is enabled") + + // Resolution phase ran in LocalSchedulerBackend: credentials were acquired and stored in + // the shared SparkEnv credential store (the same one tasks read from in local mode). + val stored = SparkEnv.get.userCredentials.get() + assert(stored != null, + "LocalSchedulerBackend should acquire and store OIDC credentials in local mode") + assert(stored.version >= 1L, "stored credentials should carry a version >= 1") + val creds = UserCredentialManager.deserializeUserCredentials(stored.bytes) + assert(creds.forScheme("fake").isPresent, + "stored credentials should contain the 'fake' scheme resolved by FakeCredentialProvider") + } + + test("LocalSchedulerBackend is a no-op for OIDC when disabled") { + sc = new SparkContext(oidcConf(enabled = false)) + + assert(sc.schedulerBackend.isInstanceOf[LocalSchedulerBackend]) + assert(sc.userCredentialProviderLoader.isEmpty, + "no loader should be allocated when OIDC is disabled") + assert(SparkEnv.get.userCredentials.get() == null, + "no OIDC credentials should be stored when OIDC is disabled") + // The provider's declared property must not have been applied. + assert(sc.hadoopConfiguration.get("fs.fake.credentials.provider") == null, + "no provider wiring should be applied when OIDC is disabled") + } +} From a3e8d552fbd13e4cc9d7e911fbb3ea5c19038f2e Mon Sep 17 00:00:00 2001 From: Kousuke Saruta Date: Fri, 11 Sep 2026 18:41:19 +0000 Subject: [PATCH 2/3] [SPARK-59296][CORE][FOLLOWUP] Address review: bind local credential callback 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. --- .../deploy/security/UserCredentialManager.scala | 14 ++++++++++---- .../scheduler/local/LocalSchedulerBackend.scala | 13 ++++++++++--- 2 files changed, 20 insertions(+), 7 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala b/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala index 3cee7c46b1712..7309a92cc23ab 100644 --- a/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala +++ b/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala @@ -541,10 +541,16 @@ private[spark] object UserCredentialManager extends Logging { * * This phase runs in local mode as well: `LocalSchedulerBackend` starts a * [[UserCredentialManager]] (for parity with `HadoopDelegationTokenManager`, which also runs - * in `LocalSchedulerBackend`), so a resolution phase follows and the wiring applied here - * points at credentials that are actually populated. (Before SPARK-59296's follow-up, local - * mode had no resolution phase, so this selection phase was skipped there to avoid wiring a - * provider whose credentials would never be resolved.) + * in `LocalSchedulerBackend`), so a resolution phase follows and populates the credentials the + * wiring points at. As in cluster mode, the resolution phase runs later than this selection + * phase (at scheduler-backend start), so there is a driver-side early-startup window: any + * driver-side access to a wired scheme during `SparkContext` construction -- for example + * fetching `spark.jars` / `spark.files` / `spark.archives`, or a `spark.checkpoint.dir` on such + * a scheme -- happens before the credentials exist and therefore cannot use them. This is the + * same driver-side window that exists in cluster mode; prefer `local://` for such resources. + * (Before SPARK-59296's follow-up, local mode had no resolution phase; the wiring applied here + * would never have been backed by resolved credentials, so this phase returned early in local + * mode.) * * Scheme selection is limited to schemes for which a provider is UNAMBIGUOUSLY selected: * either an explicitly-configured scheme (`spark.security.oidc.provider.`) or a diff --git a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala index 3ef41cdcd8c45..543ad55df40af 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala @@ -155,6 +155,14 @@ private[spark] class LocalSchedulerBackend( * 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 @@ -162,15 +170,14 @@ private[spark] class LocalSchedulerBackend( 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) + 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( - SparkEnv.get.userCredentials, version, initialCredentials) + VersionedCredentials.updateIfNewer(env.userCredentials, version, initialCredentials) } } From a9873cf4f3ebf41a08e50df18dcc784e05d69b3c Mon Sep 17 00:00:00 2001 From: Kousuke Saruta Date: Sat, 12 Sep 2026 05:23:57 +0000 Subject: [PATCH 3/3] [SPARK-59296][CORE][FOLLOWUP] Address round-2 review: hoist UserCredentialManager lifecycle into SupportsDelegationToken and tighten local-mode wiring Addresses the second round of review on PR #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. --- .../security/UserCredentialManager.scala | 5 +- .../scheduler/SupportsDelegationToken.scala | 47 +++++++- .../CoarseGrainedSchedulerBackend.scala | 48 +++----- .../local/LocalSchedulerBackend.scala | 92 +++++++--------- .../security/UserCredentialManagerSuite.scala | 68 +++--------- .../local/LocalSchedulerBackendSuite.scala | 104 ++++++++---------- 6 files changed, 166 insertions(+), 198 deletions(-) diff --git a/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala b/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala index 7309a92cc23ab..9d6a72a48884f 100644 --- a/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala +++ b/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala @@ -51,8 +51,9 @@ import org.apache.spark.util.{ThreadUtils, Utils} * safetyMargin` * 5. Retries with exponential backoff on failure * - * Intended to be started from `CoarseGrainedSchedulerBackend.start()` when - * `spark.security.oidc.enabled=true`, independently of + * Intended to be started from a scheduler backend -- `CoarseGrainedSchedulerBackend.start()` or + * `LocalSchedulerBackend.start()` (both via `SupportsDelegationToken.setupUserCredentialManager()`) + * -- when `spark.security.oidc.enabled=true`, independently of * `UserGroupInformation.isSecurityEnabled()`. * * Lifecycle: call `start()` exactly once, then `stop()` to shut down. diff --git a/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala b/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala index 6edcc7806e01c..27c00840afbe3 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala @@ -20,7 +20,7 @@ package org.apache.spark.scheduler import org.apache.hadoop.security.UserGroupInformation import org.apache.spark.deploy.SparkHadoopUtil -import org.apache.spark.deploy.security.HadoopDelegationTokenManager +import org.apache.spark.deploy.security.{HadoopDelegationTokenManager, UserCredentialManager} /** * A mix-in trait for SchedulerBackend that supports delegation tokens. @@ -30,6 +30,20 @@ private[spark] trait SupportsDelegationToken { // The token manager used to create security tokens. protected var delegationTokenManager: Option[HadoopDelegationTokenManager] = None + // The OIDC user-credential manager, a sibling of the Kerberos delegation token manager. + // Its lifecycle lives here (like delegationTokenManager) so that both + // CoarseGrainedSchedulerBackend and LocalSchedulerBackend share a single implementation; the + // two backends differ only in how they propagate credentials, expressed via + // propagateUserCredentials(). + protected var userCredentialManager: Option[UserCredentialManager] = None + + /** + * The task scheduler this backend belongs to. Implemented by the mixing-in backend (both + * CoarseGrainedSchedulerBackend and LocalSchedulerBackend already hold a `scheduler` field). + * Used to reach `scheduler.sc.conf` / `scheduler.sc.env` / the OIDC credential provider loader. + */ + protected def scheduler: TaskSchedulerImpl + /** * Create the delegation token manager to be used for the application. This method is called * once during the start of the scheduler backend (so after the object has already been @@ -42,6 +56,15 @@ private[spark] trait SupportsDelegationToken { */ protected def updateDelegationTokens(tokens: Array[Byte]): Unit + /** + * Propagate a freshly acquired set of OIDC user credentials. Called on the driver by the + * [[UserCredentialManager]] (initially from `start()`, then on each renewal). Implemented per + * backend: `CoarseGrainedSchedulerBackend` updates the driver store and broadcasts to executors + * via its `DriverEndpoint`; `LocalSchedulerBackend` (no remote executors) updates the shared + * credential store directly. + */ + protected def propagateUserCredentials(version: Long, credentials: Array[Byte]): Unit + /** * Whether the token manager should be started. The default implementation returns true when * Hadoop security is enabled. Backends that support direct credential providers override this @@ -75,4 +98,26 @@ private[spark] trait SupportsDelegationToken { protected def stopTokenManager(): Unit = { delegationTokenManager.foreach(_.stop()) } + + /** + * Start the [[UserCredentialManager]] if OIDC credential propagation is enabled. Called once + * during scheduler-backend start, independently of the Kerberos delegation token manager. + * + * Binds the manager to `scheduler.sc.conf` (the live SparkConf, not a clone) so the + * resolution-time fallback in `UserCredentialManager.start()` -- which applies provider-declared + * `spark.*` properties for any scheme the selection phase could not -- reaches the same conf on + * both backends. Reuses the [[org.apache.spark.security.CredentialProviderLoader]] produced by + * SparkContext's selection phase so providers are initialized exactly once; SparkContext remains + * the loader's single owner. `start()` invokes `propagateUserCredentials` synchronously for the + * initial credentials, so no separate initial store is needed here. + */ + protected def setupUserCredentialManager(): Unit = { + userCredentialManager = UserCredentialManager.create( + scheduler.sc.conf, propagateUserCredentials, scheduler.sc.userCredentialProviderLoader) + userCredentialManager.foreach(_.start()) + } + + protected def stopUserCredentialManager(): Unit = { + userCredentialManager.foreach(_.stop()) + } } diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala index e03ea10f03b12..7bd3e49a09c8c 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala @@ -28,7 +28,6 @@ import com.google.common.cache.CacheBuilder import org.apache.spark.{ExecutorAllocationClient, SparkEnv, TaskState, VersionedCredentials} import org.apache.spark.deploy.SparkHadoopUtil -import org.apache.spark.deploy.security.UserCredentialManager import org.apache.spark.errors.SparkCoreErrors import org.apache.spark.executor.ExecutorLogUrlHandler import org.apache.spark.internal.{config, Logging} @@ -54,7 +53,7 @@ import org.apache.spark.util.ArrayImplicits._ * Spark's standalone deploy mode (spark.deploy.*). */ private[spark] -class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: RpcEnv) +class CoarseGrainedSchedulerBackend(protected val scheduler: TaskSchedulerImpl, val rpcEnv: RpcEnv) extends ExecutorAllocationClient with SchedulerBackend with SupportsDelegationToken with Logging { @@ -144,9 +143,6 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp // Current set of delegation tokens to send to executors. private val delegationTokens = new AtomicReference[Array[Byte]]() - // UserCredentialManager for OIDC credential propagation (if enabled). - private var userCredentialManager: Option[UserCredentialManager] = None - private val reviveThread = ThreadUtils.newDaemonSingleThreadScheduledExecutor("driver-revive-thread") @@ -1252,42 +1248,26 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp * Called from the DriverEndpoint receive loop (thread-safe access to executorDataMap). */ private def updateUserCredentials(version: Long, credentials: Array[Byte]): Unit = { - VersionedCredentials.updateIfNewer(SparkEnv.get.userCredentials, version, credentials) + VersionedCredentials.updateIfNewer(scheduler.sc.env.userCredentials, version, credentials) executorDataMap.values.foreach { ed => ed.executorEndpoint.send(UpdateUserCredentials(version, credentials)) } } /** - * Start the UserCredentialManager if OIDC credential propagation is enabled. - * Called from start(), independently of Kerberos/HadoopDelegationTokenManager. + * Propagate OIDC user credentials to executors. Called on the driver by the + * [[org.apache.spark.deploy.security.UserCredentialManager]] (initially and on each renewal). + * + * Updates the driver's own credential store synchronously so the credentials are available for + * `SparkAppConfig` (late-registering executors) and `TaskDescription` (task dispatch) with no + * null window, then broadcasts to registered executors via the `DriverEndpoint` (mirroring + * `HadoopDelegationTokenManager`'s `UpdateDelegationTokens` path) to ensure thread-safe access + * to `executorDataMap`. */ - private def setupUserCredentialManager(): Unit = { - // 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) => - // Send to DriverEndpoint to ensure thread-safe access to executorDataMap. - // This mirrors HadoopDelegationTokenManager's pattern of sending - // UpdateDelegationTokens via schedulerRef. - driverEndpoint.send(UpdateUserCredentials(version, credentials)) - }, scheduler.sc.userCredentialProviderLoader) - userCredentialManager.foreach { manager => - val (version, initialCredentials) = manager.start() - // Store initial credentials synchronously so they are available for SparkAppConfig - // (late-registering executors) and TaskDescription (task dispatch) immediately. - // Note: the onCredentialsUpdate callback above also triggers an async - // UpdateUserCredentials message that will redundantly call updateIfNewer. - // The synchronous set here ensures no null window before the async message - // is processed by DriverEndpoint. - VersionedCredentials.updateIfNewer( - SparkEnv.get.userCredentials, version, initialCredentials) - } - } - - private def stopUserCredentialManager(): Unit = { - userCredentialManager.foreach(_.stop()) + override protected def propagateUserCredentials( + version: Long, credentials: Array[Byte]): Unit = { + VersionedCredentials.updateIfNewer(scheduler.sc.env.userCredentials, version, credentials) + driverEndpoint.send(UpdateUserCredentials(version, credentials)) } /** diff --git a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala index 543ad55df40af..6d9bd5f5e6c78 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala @@ -24,7 +24,7 @@ import java.nio.ByteBuffer import org.apache.spark.{SparkConf, SparkContext, SparkEnv, TaskState, VersionedCredentials} import org.apache.spark.TaskState.TaskState import org.apache.spark.deploy.SparkHadoopUtil -import org.apache.spark.deploy.security.{HadoopDelegationTokenManager, UserCredentialManager} +import org.apache.spark.deploy.security.HadoopDelegationTokenManager import org.apache.spark.executor.{Executor, ExecutorBackend} import org.apache.spark.internal.{config, Logging, LogKeys} import org.apache.spark.launcher.{LauncherBackend, SparkAppHandle} @@ -112,20 +112,19 @@ private[spark] class LocalEndpoint( */ private[spark] class LocalSchedulerBackend( conf: SparkConf, - scheduler: TaskSchedulerImpl, + protected val scheduler: TaskSchedulerImpl, val totalCores: Int) extends SchedulerBackend with ExecutorBackend with SupportsDelegationToken with Logging { private val appId = conf.get("spark.test.appId", "local-" + System.currentTimeMillis) private var localEndpoint: RpcEndpointRef = null + // Set true by stop() so that a stop request arriving before start() (e.g. a launcher KILLED + // request; launcherBackend.connect() runs in the constructor and onStopRequest fires on its own + // thread) prevents start() from bringing up the executor endpoint and, in particular, the + // UserCredentialManager renewal thread (which performs network I/O) on an already-killed app. + @volatile private var stopped = false private val userClassPath = getUserClasspath(conf) private val listenerBus = scheduler.sc.listenerBus - - // UserCredentialManager for OIDC credential propagation (if enabled). Started in start() and - // stopped in stop(), mirroring CoarseGrainedSchedulerBackend so that OIDC credentials are - // acquired and renewed in local mode too, for parity with HadoopDelegationTokenManager (which - // this backend already runs via createTokenManager()). - private var userCredentialManager: Option[UserCredentialManager] = None private val launcherBackend = new LauncherBackend() { override def conf: SparkConf = LocalSchedulerBackend.this.conf override def onStopRequest(): Unit = stop(SparkAppHandle.State.KILLED) @@ -144,45 +143,22 @@ private[spark] class LocalSchedulerBackend( } /** - * Start the UserCredentialManager if OIDC credential propagation is enabled, mirroring - * CoarseGrainedSchedulerBackend. Runs independently of Kerberos/HadoopDelegationTokenManager. + * Propagate OIDC user credentials in local mode. There are no remote executors here: the driver + * and the single in-JVM executor share `scheduler.sc.env.userCredentials`, so this updates that + * store directly and subsequently dispatched tasks (and driver-side access) observe the new + * credentials. `UserCredentialManager.start()` calls this synchronously for the initial + * credentials, so they are in the store before any task is offered. * - * 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. + * The store is taken from `scheduler.sc.env` (this backend's own env), not the global + * `SparkEnv.get`: `UserCredentialManager.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 store (e.g. a new context created in the same JVM by a notebook, test, or + * Spark Connect session). Binding to this env's store confines a late renewal to this + * application's (by then unused) store. */ - 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) - } - } - - private def stopUserCredentialManager(): Unit = { - userCredentialManager.foreach(_.stop()) + override protected def propagateUserCredentials( + version: Long, credentials: Array[Byte]): Unit = { + VersionedCredentials.updateIfNewer(scheduler.sc.env.userCredentials, version, credentials) } /** @@ -198,6 +174,12 @@ private[spark] class LocalSchedulerBackend( launcherBackend.connect() override def start(): Unit = { + // If a stop request already arrived (e.g. launcher KILLED before start()), do not bring up the + // executor endpoint or the token/credential managers on an app that is already stopping. + if (stopped) { + logInfo("Not starting LocalSchedulerBackend because it was already stopped") + return + } val rpcEnv = SparkEnv.get.rpcEnv val executorEndpoint = new LocalEndpoint(rpcEnv, userClassPath, scheduler, this, totalCores) localEndpoint = rpcEnv.setupEndpoint("LocalSchedulerBackendEndpoint", executorEndpoint) @@ -246,13 +228,19 @@ private[spark] class LocalSchedulerBackend( } private def stop(finalState: SparkAppHandle.State): Unit = { - // 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) + // Mark stopped so a start() that has not run yet becomes a no-op (a KILLED request can race + // ahead of start()). The executor endpoint only needs stopping if start() already created it; + // guard on localEndpoint being non-null, mirroring CoarseGrainedSchedulerBackend's + // `if (driverEndpoint != null)`. The token and user-credential managers are always stopped + // (each isolated with tryLogNonFatalError so a failure in one does not skip the other or the + // launcher state update below); 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. + stopped = true + if (localEndpoint != null) { + Utils.tryLogNonFatalError { + localEndpoint.ask(StopExecutor) + } } Utils.tryLogNonFatalError { stopTokenManager() diff --git a/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala b/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala index 3224642512d83..e8ff807e7173c 100644 --- a/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala +++ b/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala @@ -746,10 +746,10 @@ class UserCredentialManagerSuite extends SparkFunSuite { 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. + // properties, then the resolution phase (start()) 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") @@ -761,52 +761,35 @@ class UserCredentialManagerSuite extends SparkFunSuite { assert(conf.get("spark.hadoop.fs.fake.credentials.provider") === "org.apache.spark.security.FakeExecutorCredentialProvider") - // Selection must NOT have initialized the provider (selectProviderForProperties skips init). + // Observe the SAME provider instance the loader caches, WITHOUT initializing it, and assert + // the selection phase left it uninitialized. Do not call providerFor() here: that would + // initialize the provider before start(), making the post-start assertion below only prove + // loader idempotency rather than that start() reused this selection-phase loader. 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() - .asInstanceOf[FakeCredentialProvider] - assert(resolved eq provider, "loader must reuse the same cached provider instance") - assert(provider.getInitCount === 1, - "providerFor should initialize the provider exactly once") - - // start() goes through the real FileTokenIngestor (via create()), so a parseable token - // file must exist. Write a minimal unsigned JWT with the required sub/iss/exp claims. - val tokenFile = java.io.File.createTempFile("oidc-token-", ".jwt") - tokenFile.deleteOnExit() - val enc = java.util.Base64.getUrlEncoder.withoutPadding() - val header = enc.encodeToString( - """{"alg":"none","typ":"JWT"}""".getBytes(java.nio.charset.StandardCharsets.UTF_8)) - val exp = Instant.now().plusSeconds(300).getEpochSecond - val payload = enc.encodeToString( - s"""{"sub":"test-user","iss":"https://issuer.example.com","exp":$exp}""" - .getBytes(java.nio.charset.StandardCharsets.UTF_8)) - conf.set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath) - java.nio.file.Files.writeString(tokenFile.toPath, s"$header.$payload") - - // Resolution phase reuses the same loader (as UserCredentialManager.create does with the - // loader from the selection phase). start() calls providerFor internally, which must NOT - // re-init the already-initialized provider instance. - val manager = UserCredentialManager.create( - conf, (_: Long, _: Array[Byte]) => (), Some(loader)).get + + // Resolution phase reuses the SAME loader (passed explicitly, as UserCredentialManager.create + // does with the loader from the selection phase). A mock ingestor avoids needing a token file. + // start() calls providerFor internally, which performs the single init() on this same cached + // instance -- so getInitCount goes 0 -> 1. If start() had regressed to a fresh loader, it would + // initialize a DIFFERENT FakeCredentialProvider and this instance's count would stay 0. + val manager = new UserCredentialManager( + conf, createIngestor(createUserContext()), (_: Long, _: Array[Byte]) => (), loader) try { val (version, bytes) = manager.start() assert(version === 1L) val creds = UserCredentialManager.deserializeUserCredentials(bytes) assert(creds.forScheme("fake").isPresent) assert(provider.getInitCount === 1, - "reusing the selection-phase loader must not re-initialize the provider") + "resolution must initialize the selection-phase provider exactly once") } finally { manager.stop() - tokenFile.delete() + loader.closeAll() } } @@ -832,21 +815,6 @@ class UserCredentialManagerSuite extends SparkFunSuite { assert(!conf.contains("spark.fake.credentials.enabled")) } - test("applyProviderProperties applies properties regardless of local mode") { - // SPARK-59296 follow-up: local mode now starts a resolution phase (LocalSchedulerBackend - // starts a UserCredentialManager), so the selection phase is no longer skipped in local - // mode. There is no longer an isLocal parameter; selection applies properties whenever OIDC - // is enabled, exactly as it does for cluster mode. - val conf = createSparkConf() - conf.set("spark.security.oidc.provider.fake", - "org.apache.spark.security.FakeCredentialProvider") - val loader = UserCredentialManager.applyProviderProperties(conf) - assert(loader.isDefined, "a loader should be allocated whenever OIDC is enabled") - assert(conf.get("spark.hadoop.fs.fake.credentials.provider") === - "org.apache.spark.security.FakeExecutorCredentialProvider") - assert(conf.get("spark.fake.credentials.enabled") === "true") - } - test("applyProviderProperties does not overwrite user-set properties") { val conf = createSparkConf() conf.set("spark.security.oidc.provider.fake", diff --git a/core/src/test/scala/org/apache/spark/scheduler/local/LocalSchedulerBackendSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/local/LocalSchedulerBackendSuite.scala index a89982dfb0f99..ab5a27b90a930 100644 --- a/core/src/test/scala/org/apache/spark/scheduler/local/LocalSchedulerBackendSuite.scala +++ b/core/src/test/scala/org/apache/spark/scheduler/local/LocalSchedulerBackendSuite.scala @@ -21,7 +21,9 @@ import java.io.File import java.nio.charset.StandardCharsets import java.nio.file.Files import java.time.Instant -import java.util.Base64 +import java.util.Date + +import io.jsonwebtoken.Jwts import org.apache.spark.{LocalSparkContext, SparkConf, SparkContext, SparkEnv, SparkFunSuite} import org.apache.spark.deploy.security.UserCredentialManager @@ -35,37 +37,19 @@ import org.apache.spark.internal.config._ */ 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 = { - val enc = Base64.getUrlEncoder.withoutPadding() - val header = enc.encodeToString( - """{"alg":"none","typ":"JWT"}""".getBytes(StandardCharsets.UTF_8)) - val exp = Instant.now().plusSeconds(300).getEpochSecond - val payload = enc.encodeToString( - s"""{"sub":"test-user","iss":"https://issuer.example.com","exp":$exp}""" - .getBytes(StandardCharsets.UTF_8)) - s"$header.$payload" + /** Write an unsigned JWT (with the sub/iss/exp claims FileTokenIngestor requires) to `dir`. */ + private def writeTokenFile(dir: File): File = { + val jwt = Jwts.builder() + .subject("test-user") + .issuer("https://issuer.example.com") + .expiration(Date.from(Instant.now().plusSeconds(300))) + .compact() + val f = new File(dir, "oidc-token.jwt") + Files.write(f.toPath, jwt.getBytes(StandardCharsets.UTF_8)) + f } - private def oidcConf(enabled: Boolean): SparkConf = { + private def oidcConf(enabled: Boolean, tokenFile: Option[File] = None): SparkConf = { val conf = new SparkConf() .setMaster("local[1]") .setAppName("LocalSchedulerBackendSuite") @@ -75,7 +59,7 @@ class LocalSchedulerBackendSuite extends SparkFunSuite with LocalSparkContext { .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L) if (enabled) { conf - .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath) + .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.get.getAbsolutePath) .set("spark.security.oidc.provider.fake", "org.apache.spark.security.FakeCredentialProvider") } @@ -83,34 +67,36 @@ class LocalSchedulerBackendSuite extends SparkFunSuite with LocalSparkContext { } test("LocalSchedulerBackend runs OIDC selection and resolution in local mode") { - sc = new SparkContext(oidcConf(enabled = true)) - - // The scheduler backend in local mode is a LocalSchedulerBackend. - assert(sc.schedulerBackend.isInstanceOf[LocalSchedulerBackend], - "local[1] should use LocalSchedulerBackend") - - // Selection phase ran on the driver (SparkContext) even in local mode: the provider's - // declared spark.hadoop.* property reached the driver's Hadoop Configuration (prefix - // stripped), and its non-Hadoop spark.* property reached SparkConf. - assert(sc.hadoopConfiguration.get("fs.fake.credentials.provider") === - "org.apache.spark.security.FakeExecutorCredentialProvider", - "selection phase should wire the provider into the driver's Hadoop Configuration") - assert(sc.getConf.get("spark.fake.credentials.enabled") === "true", - "selection phase should apply non-Hadoop provider properties too") - - // A loader was retained for reuse by the resolution phase. - assert(sc.userCredentialProviderLoader.isDefined, - "SparkContext should retain the selection-phase loader when OIDC is enabled") - - // Resolution phase ran in LocalSchedulerBackend: credentials were acquired and stored in - // the shared SparkEnv credential store (the same one tasks read from in local mode). - val stored = SparkEnv.get.userCredentials.get() - assert(stored != null, - "LocalSchedulerBackend should acquire and store OIDC credentials in local mode") - assert(stored.version >= 1L, "stored credentials should carry a version >= 1") - val creds = UserCredentialManager.deserializeUserCredentials(stored.bytes) - assert(creds.forScheme("fake").isPresent, - "stored credentials should contain the 'fake' scheme resolved by FakeCredentialProvider") + withTempDir { dir => + sc = new SparkContext(oidcConf(enabled = true, tokenFile = Some(writeTokenFile(dir)))) + + // The scheduler backend in local mode is a LocalSchedulerBackend. + assert(sc.schedulerBackend.isInstanceOf[LocalSchedulerBackend], + "local[1] should use LocalSchedulerBackend") + + // Selection phase ran on the driver (SparkContext) even in local mode: the provider's + // declared spark.hadoop.* property reached the driver's Hadoop Configuration (prefix + // stripped), and its non-Hadoop spark.* property reached SparkConf. + assert(sc.hadoopConfiguration.get("fs.fake.credentials.provider") === + "org.apache.spark.security.FakeExecutorCredentialProvider", + "selection phase should wire the provider into the driver's Hadoop Configuration") + assert(sc.getConf.get("spark.fake.credentials.enabled") === "true", + "selection phase should apply non-Hadoop provider properties too") + + // A loader was retained for reuse by the resolution phase. + assert(sc.userCredentialProviderLoader.isDefined, + "SparkContext should retain the selection-phase loader when OIDC is enabled") + + // Resolution phase ran in LocalSchedulerBackend: credentials were acquired and stored in + // the shared SparkEnv credential store (the same one tasks read from in local mode). + val stored = SparkEnv.get.userCredentials.get() + assert(stored != null, + "LocalSchedulerBackend should acquire and store OIDC credentials in local mode") + assert(stored.version >= 1L, "stored credentials should carry a version >= 1") + val creds = UserCredentialManager.deserializeUserCredentials(stored.bytes) + assert(creds.forScheme("fake").isPresent, + "stored credentials should contain the 'fake' scheme resolved by FakeCredentialProvider") + } } test("LocalSchedulerBackend is a no-op for OIDC when disabled") {