[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.

Author: sarutak

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.

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

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