[SPARK-59296][CORE][FOLLOWUP] Run OIDC UserCredentialManager in local mode for parity with HadoopDelegationTokenManager

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)

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.
@@ -482,11 +483,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 +505,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 +540,18 @@ 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 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.<scheme>`) or a
@@ -554,17 +561,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
     }
 

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())
+  }
 }

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 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.
-    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))
   }
 
   /**