[SPARK-57894][SPARK-57895][CORE][K8S] Implement UpdateUserCredentials RPC, backend integration, and initial credential delivery

core/src/main/scala/org/apache/spark/SparkEnv.scala

@@ -19,6 +19,7 @@ package org.apache.spark
 
 import java.io.File
 import java.util.concurrent.{CountDownLatch, TimeUnit}
+import java.util.concurrent.atomic.AtomicReference
 
 import scala.collection.concurrent
 import scala.collection.mutable
@@ -268,6 +269,15 @@ class SparkEnv (
 
   private[spark] var executorBackend: Option[ExecutorBackend] = None
 
+  /**
+   * Credential store for OIDC-based user credentials on executors.
+   * Updated via `UpdateUserCredentials` RPC and `TaskDescription` credential delivery.
+   * Read by connector-specific credential providers (e.g., SparkOidcAwsCredentialsProvider).
+   * Contains serialized `UserCredentials` (no raw identity token).
+   */
+  private[spark] val userCredentials: AtomicReference[Array[Byte]] =
+    new AtomicReference[Array[Byte]]()
+
   private[spark] def stop(): Unit = {
 
     if (!isStopped) {

core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala

@@ -223,6 +223,11 @@ private[spark] class CoarseGrainedExecutorBackend(
       logInfo(log"Received tokens of ${MDC(LogKeys.NUM_BYTES, tokenBytes.length)} bytes")
       SparkHadoopUtil.get.addDelegationTokens(tokenBytes, env.conf)
 
+    case UpdateUserCredentials(credentials) =>
+      logInfo(log"Received user credentials of " +
+        log"${MDC(LogKeys.NUM_BYTES, credentials.length)} bytes")
+      env.userCredentials.set(credentials)
+
     case DecommissionExecutor =>
       decommissionSelf()
   }
@@ -505,6 +510,12 @@ private[spark] object CoarseGrainedExecutorBackend extends Logging {
       }
       val env = SparkEnv.createExecutorEnv(driverConf, arguments.executorId, arguments.bindAddress,
         arguments.hostname, arguments.cores, cfg.ioEncryptionKey, isLocal = false)
+
+      // Apply initial user credentials to the executor credential store.
+      cfg.userCredentials.foreach { credentials =>
+        env.userCredentials.set(credentials)
+      }
+
       // Set the application attemptId in the BlockStoreClient if available.
       val appAttemptId = env.conf.get(APP_ATTEMPT_ID)
       appAttemptId.foreach(attemptId =>

core/src/main/scala/org/apache/spark/executor/Executor.scala

@@ -847,6 +847,15 @@ private[spark] class Executor(
         // requires access to properties contained within (e.g. for access control).
         Executor.taskDeserializationProps.set(taskDescription.properties)
 
+        // Apply user credentials from TaskDescription to the executor credential store.
+        // This ensures credentials are available before any task code runs, avoiding
+        // the race between RPC broadcast and task dispatch.
+        // All tasks in a Spark application belong to the same user, so last-writer-wins
+        // is acceptable when multiple task threads update concurrently.
+        taskDescription.userCredentials.foreach { creds =>
+          env.userCredentials.set(creds)
+        }
+
         updateDependencies(
           taskDescription.artifacts.files,
           taskDescription.artifacts.jars,

core/src/main/scala/org/apache/spark/scheduler/TaskDescription.scala

@@ -61,6 +61,7 @@ private[spark] class TaskDescription(
     // Eg, Map("gpu" -> Map("0" -> ResourceAmountUtils.toInternalResource(0.7))):
     // assign 0.7 of the gpu address "0" to this task
     val resources: immutable.Map[String, immutable.Map[String, Long]],
+    val userCredentials: Option[Array[Byte]],
     val serializedTask: ByteBuffer) {
 
   assert(cpus > 0, "CPUs per task should be > 0")
@@ -121,6 +122,16 @@ private[spark] object TaskDescription {
     // Write resources.
     serializeResources(taskDescription.resources, dataOut)
 
+    // Write user credentials (OIDC).
+    taskDescription.userCredentials match {
+      case Some(creds) =>
+        dataOut.writeBoolean(true)
+        dataOut.writeInt(creds.length)
+        dataOut.write(creds)
+      case None =>
+        dataOut.writeBoolean(false)
+    }
+
     // Write the task. The task is already serialized, so write it directly to the byte buffer.
     Utils.writeByteBuffer(taskDescription.serializedTask, bytesOut)
 
@@ -228,10 +239,20 @@ private[spark] object TaskDescription {
     // Read resources.
     val resources = deserializeResources(dataIn)
 
+    // Read user credentials (OIDC).
+    val userCredentials = if (dataIn.readBoolean()) {
+      val length = dataIn.readInt()
+      val creds = new Array[Byte](length)
+      dataIn.readFully(creds)
+      Some(creds)
+    } else {
+      None
+    }
+
     // Create a sub-buffer for the serialized task into its own buffer (to be deserialized later).
     val serializedTask = byteBuffer.slice()
 
     new TaskDescription(taskId, attemptNumber, executorId, name, index, partitionId, artifacts,
-      properties, cpus, resources, serializedTask)
+      properties, cpus, resources, userCredentials, serializedTask)
   }
 }

core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala

@@ -604,6 +604,7 @@ private[spark] class TaskSetManager(
       task.localProperties,
       taskCpus,
       taskResourceAssignments,
+      Option(SparkEnv.get.userCredentials.get()),
       serializedTask)
   }
 

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala

@@ -35,6 +35,7 @@ private[spark] object CoarseGrainedClusterMessages {
       sparkProperties: Seq[(String, String)],
       ioEncryptionKey: Option[Array[Byte]],
       hadoopDelegationCreds: Option[Array[Byte]],
+      userCredentials: Option[Array[Byte]],
       resourceProfile: ResourceProfile,
       logLevel: Option[String])
     extends CoarseGrainedClusterMessage
@@ -60,6 +61,9 @@ private[spark] object CoarseGrainedClusterMessages {
   case class UpdateDelegationTokens(tokens: Array[Byte])
     extends CoarseGrainedClusterMessage
 
+  case class UpdateUserCredentials(credentials: Array[Byte])
+    extends CoarseGrainedClusterMessage
+
   // Executors to driver
   case class RegisterExecutor(
       executorId: String,

core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala

@@ -28,6 +28,7 @@ import com.google.common.cache.CacheBuilder
 
 import org.apache.spark.{ExecutorAllocationClient, SparkEnv, TaskState}
 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}
@@ -132,6 +133,13 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
   // Current set of delegation tokens to send to executors.
   private val delegationTokens = new AtomicReference[Array[Byte]]()
 
+  // Current set of OIDC user credentials to send to executors.
+  // Updated by UserCredentialManager and sent via UpdateUserCredentials RPC.
+  private val userCredentials = 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")
 
@@ -220,6 +228,9 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
       case UpdateDelegationTokens(newDelegationTokens) =>
         updateDelegationTokens(newDelegationTokens)
 
+      case UpdateUserCredentials(newCredentials) =>
+        updateUserCredentials(newCredentials)
+
       case RemoveExecutor(executorId, reason) =>
         // We will remove the executor's state and cannot restore it. However, the connection
         // between the driver and the executor may be still alive so that the executor won't exit
@@ -358,6 +369,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
           sparkProperties,
           SparkEnv.get.securityManager.getIOEncryptionKey(),
           Option(delegationTokens.get()),
+          Option(userCredentials.get()),
           rp,
           currentLogLevel)
         context.reply(reply)
@@ -618,6 +630,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
 
   override def start(): Unit = {
     setupTokenManager()
+    setupUserCredentialManager()
     if (conf.get(DIRECT_CREDENTIAL_PROVIDERS_ENABLED) && delegationTokenManager.isEmpty) {
       logWarning("spark.security.directCredentialProviders.enabled is set but " +
         "this cluster manager does not support credential distribution. " +
@@ -648,6 +661,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
     cleanupService.foreach(_.shutdownNow())
     stopExecutors()
     stopTokenManager()
+    stopUserCredentialManager()
     try {
       if (driverEndpoint != null) {
         driverEndpoint.askSync[Boolean](StopDriver)
@@ -1047,6 +1061,46 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
 
   protected def currentDelegationTokens: Array[Byte] = delegationTokens.get()
 
+  /**
+   * Update user credentials and broadcast to all registered executors.
+   * Called from the DriverEndpoint receive loop (thread-safe access to executorDataMap).
+   */
+  private def updateUserCredentials(credentials: Array[Byte]): Unit = {
+    userCredentials.set(credentials)
+    // Also update driver's SparkEnv so TaskDescription can include current credentials.
+    SparkEnv.get.userCredentials.set(credentials)
+    executorDataMap.values.foreach { ed =>
+      ed.executorEndpoint.send(UpdateUserCredentials(credentials))
+    }
+  }
+
+  protected def currentUserCredentials: Array[Byte] = userCredentials.get()
+
+  /**
+   * Start the UserCredentialManager if OIDC credential propagation is enabled.
+   * Called from start(), independently of Kerberos/HadoopDelegationTokenManager.
+   */
+  private def setupUserCredentialManager(): Unit = {
+    userCredentialManager = UserCredentialManager.create(conf, { credentials =>
+      // Send to DriverEndpoint to ensure thread-safe access to executorDataMap.
+      // This mirrors HadoopDelegationTokenManager's pattern of sending
+      // UpdateDelegationTokens via schedulerRef.
+      driverEndpoint.send(UpdateUserCredentials(credentials))
+    })
+    userCredentialManager.foreach { manager =>
+      val initialCredentials = manager.start()
+      // Store initial credentials for SparkAppConfig (late-registering executors)
+      // and TaskDescription (task dispatch). Both stores must be populated synchronously
+      // to avoid a window where tasks could be dispatched without credentials.
+      userCredentials.set(initialCredentials)
+      SparkEnv.get.userCredentials.set(initialCredentials)
+    }
+  }
+
+  private def stopUserCredentialManager(): Unit = {
+    userCredentialManager.foreach(_.stop())
+  }
+
   /**
    * Checks whether the executor is excluded due to failure(s). This is called when the executor
    * tries to register with the scheduler, and will deny registration if this method returns true.

core/src/test/scala/org/apache/spark/executor/CoarseGrainedExecutorBackendSuite.scala

@@ -312,7 +312,7 @@ class CoarseGrainedExecutorBackendSuite extends SparkFunSuite
       // We don't really verify the data, just pass it around.
       val data = ByteBuffer.wrap(Array[Byte](1, 2, 3, 4))
       val taskDescription = new TaskDescription(taskId, 2, "1", "TASK 1000000", 19,
-        1, JobArtifactSet.emptyJobArtifactSet, new Properties, 1, resourcesAmounts, data)
+        1, JobArtifactSet.emptyJobArtifactSet, new Properties, 1, resourcesAmounts, None, data)
       val serializedTaskDescription = TaskDescription.encode(taskDescription)
       backend.rpcEnv.setupEndpoint("Executor 1", backend)
       backend.executor = mock[Executor](CALLS_REAL_METHODS)
@@ -437,7 +437,7 @@ class CoarseGrainedExecutorBackendSuite extends SparkFunSuite
       // Fake tasks with different taskIds.
       val taskDescriptions = (1 to numTasks).map {
         taskId => new TaskDescription(taskId, 2, "1", s"TASK $taskId", 19,
-          1, JobArtifactSet.emptyJobArtifactSet, new Properties, 1, resourcesAmounts, data)
+          1, JobArtifactSet.emptyJobArtifactSet, new Properties, 1, resourcesAmounts, None, data)
       }
       assert(taskDescriptions.length == numTasks)
 
@@ -531,7 +531,7 @@ class CoarseGrainedExecutorBackendSuite extends SparkFunSuite
       // Fake tasks with different taskIds.
       val taskDescriptions = (1 to numTasks).map {
         taskId => new TaskDescription(taskId, 2, "1", s"TASK $taskId", 19,
-          1, JobArtifactSet.emptyJobArtifactSet, new Properties, 1, resourcesAmounts, data)
+          1, JobArtifactSet.emptyJobArtifactSet, new Properties, 1, resourcesAmounts, None, data)
       }
       assert(taskDescriptions.length == numTasks)
 

core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala

@@ -870,6 +870,7 @@ class ExecutorSuite extends SparkFunSuite
       properties = new Properties,
       cpus = 1,
       resources = Map.empty,
+      None,
       serializedTask)
   }
 

core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala

@@ -348,7 +348,7 @@ class CoarseGrainedSchedulerBackendSuite extends SparkFunSuite with LocalSparkCo
     val taskCpus = 1
     val taskDescs: Seq[Seq[TaskDescription]] = Seq(Seq(new TaskDescription(1, 0, "1",
       "t1", 0, 1, JobArtifactSet.emptyJobArtifactSet, new Properties(),
-      taskCpus, taskResources, bytebuffer)))
+      taskCpus, taskResources, None, bytebuffer)))
     val ts = backend.getTaskSchedulerImpl()
     when(ts.resourceOffers(any[IndexedSeq[WorkerOffer]], any[Boolean])).thenReturn(taskDescs)
 
@@ -455,7 +455,7 @@ class CoarseGrainedSchedulerBackendSuite extends SparkFunSuite with LocalSparkCo
     val taskCpus = 1
     val taskDescs: Seq[Seq[TaskDescription]] = Seq(Seq(new TaskDescription(1, 0, "1",
       "t1", 0, 1, JobArtifactSet.emptyJobArtifactSet, new Properties(),
-      taskCpus, taskResources, bytebuffer)))
+      taskCpus, taskResources, None, bytebuffer)))
     val ts = backend.getTaskSchedulerImpl()
     when(ts.resourceOffers(any[IndexedSeq[WorkerOffer]], any[Boolean])).thenReturn(taskDescs)
 
@@ -548,7 +548,7 @@ class CoarseGrainedSchedulerBackendSuite extends SparkFunSuite with LocalSparkCo
     val taskCpus = 2
     val taskDescs: Seq[Seq[TaskDescription]] = Seq(Seq(new TaskDescription(1, 0, "1",
       "t1", 0, 1, JobArtifactSet.emptyJobArtifactSet, new Properties(),
-      taskCpus, Map.empty, bytebuffer)))
+      taskCpus, Map.empty, None, bytebuffer)))
     when(ts.resourceOffers(any[IndexedSeq[WorkerOffer]], any[Boolean])).thenReturn(taskDescs)
 
     backend.driverEndpoint.send(ReviveOffers)
@@ -606,6 +606,76 @@ class CoarseGrainedSchedulerBackendSuite extends SparkFunSuite with LocalSparkCo
     assert(mockEndpointRef.decommissionReceived)
   }
 
+  test("UpdateUserCredentials is broadcast to all registered executors") {
+    val conf = new SparkConf()
+      .setMaster("local-cluster[0, 3, 1024]")
+      .setAppName("test")
+
+    sc = new SparkContext(conf)
+    val backend = sc.schedulerBackend.asInstanceOf[CoarseGrainedSchedulerBackend]
+
+    // Register two mock executors
+    val mockEndpointRef1 = new MockExecutorRpcEndpointRef(conf)
+    val mockEndpointRef2 = new MockExecutorRpcEndpointRef(conf)
+    val mockAddress = mock[RpcAddress]
+
+    backend.driverEndpoint.askSync[Boolean](
+      RegisterExecutor("1", mockEndpointRef1, mockAddress.host, 1, Map(), Map(),
+        Map.empty, ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID))
+    backend.driverEndpoint.askSync[Boolean](
+      RegisterExecutor("2", mockEndpointRef2, mockAddress.host, 1, Map(), Map(),
+        Map.empty, ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID))
+
+    sc.listenerBus.waitUntilEmpty(executorUpTimeout.toMillis)
+
+    // Neither executor should have received credentials yet
+    assert(mockEndpointRef1.receivedUserCredentials.isEmpty)
+    assert(mockEndpointRef2.receivedUserCredentials.isEmpty)
+
+    // Send UpdateUserCredentials via DriverEndpoint
+    val testCredentials = Array[Byte](1, 2, 3, 4, 5)
+    backend.driverEndpoint.send(UpdateUserCredentials(testCredentials))
+
+    // Wait for the message to be processed
+    eventually(timeout(5 seconds)) {
+      assert(mockEndpointRef1.receivedUserCredentials.isDefined)
+      assert(mockEndpointRef2.receivedUserCredentials.isDefined)
+    }
+
+    assert(mockEndpointRef1.receivedUserCredentials.get === testCredentials)
+    assert(mockEndpointRef2.receivedUserCredentials.get === testCredentials)
+
+    // Verify SparkEnv credential store is also updated
+    assert(SparkEnv.get.userCredentials.get() === testCredentials)
+  }
+
+  test("SparkAppConfig includes current user credentials for late-registering executors") {
+    val conf = new SparkConf()
+      .setMaster("local-cluster[0, 3, 1024]")
+      .setAppName("test")
+
+    sc = new SparkContext(conf)
+    val backend = sc.schedulerBackend.asInstanceOf[CoarseGrainedSchedulerBackend]
+
+    // Simulate credential acquisition by setting credentials before any executor registers
+    val testCredentials = Array[Byte](10, 20, 30, 40, 50)
+    backend.driverEndpoint.send(UpdateUserCredentials(testCredentials))
+
+    // Wait for the message to be processed
+    eventually(timeout(5 seconds)) {
+      assert(SparkEnv.get.userCredentials.get() != null)
+    }
+
+    // Now retrieve SparkAppConfig as a late-registering executor would
+    val appConfig = backend.driverEndpoint.askSync[SparkAppConfig](
+      RetrieveSparkAppConfig(ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID))
+
+    // Verify that userCredentials is present in the response
+    assert(appConfig.userCredentials.isDefined,
+      "SparkAppConfig should include user credentials for late-registering executors")
+    assert(appConfig.userCredentials.get === testCredentials)
+  }
+
   private def testSubmitJob(sc: SparkContext, rdd: RDD[Int]): Unit = {
     sc.submitJob(
       rdd,
@@ -667,12 +737,15 @@ private[spark] class MockExecutorRpcEndpointRef(conf: SparkConf) extends RpcEndp
   // scalastyle:on executioncontextglobal
 
   var decommissionReceived = false
+  var receivedUserCredentials: Option[Array[Byte]] = None
 
   override def address: RpcAddress = null
   override def name: String = "executor"
   override def send(message: Any): Unit =
     message match {
       case DecommissionExecutor => decommissionReceived = true
+      case UpdateUserCredentials(creds) => receivedUserCredentials = Some(creds)
+      case _ =>
     }
   override def ask[T: ClassTag](message: Any, timeout: RpcTimeout): Future[T] = {
     Future{true.asInstanceOf[T]}