You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by sa...@apache.org on 2026/08/06 14:23:27 UTC
(spark) branch master updated: [SPARK-57894][SPARK-57895][CORE][K8S] Implement `UpdateUserCredentials` RPC, backend integration, and initial credential delivery
This is an automated email from the ASF dual-hosted git repository.
sarutak pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 247405ce1c77 [SPARK-57894][SPARK-57895][CORE][K8S] Implement `UpdateUserCredentials` RPC, backend integration, and initial credential delivery
247405ce1c77 is described below
commit 247405ce1c77ec12e6b0b07c663ba411fcd0f795
Author: Kousuke Saruta <sa...@apache.org>
AuthorDate: Thu Aug 6 23:23:14 2026 +0900
[SPARK-57894][SPARK-57895][CORE][K8S] Implement `UpdateUserCredentials` RPC, backend integration, and initial credential delivery
### What changes were proposed in this pull request?
This PR adds the credential distribution layer for the OIDC Credential Propagation SPIP (SPARK-57703). This PR covers Sub-task 5 (SPARK-57894): UpdateUserCredentials RPC and backend integration) and Sub-task 6 (SPARK-57895): initial credential delivery for newly registered executors).
Sub-task 6 was originally scoped as a separate PR, but during implementation it became clear that the two are inseparable: the `AtomicReference` storing current credentials (Sub-task 5) is the same state that `SparkAppConfig` reads from for initial delivery (Sub-task 6). Splitting them would produce an artificial boundary with no meaningful independent reviewability.
**Credential delivery via three complementary paths:**
1. **SparkAppConfig** (Sub-task 6): When a new executor registers, `RetrieveSparkAppConfig` includes the current credentials in the response. This ensures late-registering executors (e.g., dynamic allocation, Executor Plugins that access S3 during `init()`) receive credentials immediately without waiting for a refresh cycle or task dispatch.
2. **UpdateUserCredentials RPC** (Sub-task 5): On credential renewal, `UserCredentialManager`'s callback sends `UpdateUserCredentials` to the `DriverEndpoint`, which broadcasts to all registered executors. This refreshes credentials during long-running tasks.
3. **TaskDescription** (Sub-task 5): Each task carries the current credentials at dispatch time. This eliminates the race between RPC broadcast and task execution, guaranteeing credentials are available before any task code runs. Unlike Hadoop delegation tokens (which have hours-long TTLs), OIDC credentials are short-lived (minutes), making this race a practical concern.
**Version guard for stale-credential prevention:**
Each credential renewal increments a monotonic version counter in `UserCredentialManager`. Credentials are transmitted as `(version: Long, bytes: Array[Byte])` across all three delivery paths. On the executor side, both `CoarseGrainedExecutorBackend.receive` and `Executor.TaskRunner` use `AtomicReference.updateAndGet` to atomically apply credentials only if the incoming version is strictly greater than the current one. This prevents a delayed `TaskDescription` (carrying an older versi [...]
**Key implementation details:**
- `UpdateUserCredentials(version: Long, credentials: Array[Byte])`: New RPC message
- `VersionedCredentials(version: Long, bytes: Array[Byte])`: Case class for the credential store
- `SparkEnv.userCredentials: AtomicReference[VersionedCredentials]`: Credential store on both driver and executors
- `CoarseGrainedSchedulerBackend`: Manages credential store, starts/stops `UserCredentialManager`, handles `UpdateUserCredentials` in `DriverEndpoint.receive` (thread-safe access to `executorDataMap`)
- `CoarseGrainedExecutorBackend`: Handles `UpdateUserCredentials` with version guard; applies initial credentials from `SparkAppConfig` on startup
- `TaskDescription`: New `userCredentials: Option[(Long, Array[Byte])]` field with encode/decode
- `Executor`: Applies credentials with version guard before task execution
**Security:** Raw identity tokens (`UserContext.rawToken`) are never present in any RPC payload, `TaskDescription`, or `SparkAppConfig`. Only derived `ServiceCredential` bundles (serialized `UserCredentials`) are transmitted.
**Modified files:**
| File | Change |
|------|--------|
| `CoarseGrainedClusterMessage.scala` | `UpdateUserCredentials` message with version, `SparkAppConfig.userCredentials` field |
| `CoarseGrainedSchedulerBackend.scala` | `updateUserCredentials()`, `setupUserCredentialManager()`, `stopUserCredentialManager()` |
| `CoarseGrainedExecutorBackend.scala` | `UpdateUserCredentials` handler with version guard, initial credential application |
| `SparkEnv.scala` | `VersionedCredentials` case class, `userCredentials` credential store |
| `TaskDescription.scala` | `userCredentials` field with version, encode/decode |
| `TaskSetManager.scala` | Include current versioned credentials in `TaskDescription` |
| `Executor.scala` | Apply credentials with atomic version guard before task execution |
| `KubernetesExecutorBackend.scala` | Apply initial credentials from `SparkAppConfig` |
| `UserCredentialManager.scala` | Add monotonic version counter, update callback signature |
### Why are the changes needed?
After Sub-task 4 (#57387, `UserCredentialManager`) acquires and renews credentials on the driver, there is no mechanism to distribute them to executors. Without this PR, the credential propagation framework has no delivery layer, and executors would never receive the short-lived service credentials needed to access external services (e.g., S3 via STS-derived credentials).
### Does this PR introduce _any_ user-facing change?
No. All new behavior is gated by `spark.security.oidc.enabled=false` (default). No existing APIs or behaviors are changed.
### How was this patch tested?
- **TaskDescriptionSuite**: Encode/decode round-trip with versioned `userCredentials` (Some and None)
- **CoarseGrainedSchedulerBackendSuite**:
- `UpdateUserCredentials` broadcast to all registered executors
- `SparkAppConfig` includes versioned credentials for late-registering executors
- Version guard prevents stale credentials from overwriting newer ones (v3 -> v1 rejected -> v5 accepted)
- Executor-side credential store version guard rejects stale and accepts newer (unit test for `updateAndGet` logic)
- **Existing test updates**: All `TaskDescription` and `UserCredentialManager` call sites updated with versioned signatures
### Was this patch authored or co-authored using generative AI tooling?
Kiro CLI / Claude
Closes #57675 from sarutak/oidc-propagation/subtask5-update-user-credentials-rpc.
Authored-by: Kousuke Saruta <sa...@apache.org>
Signed-off-by: Kousuke Saruta <sa...@apache.org>
---
.../java/org/apache/spark/internal/LogKeys.java | 1 +
.../src/main/scala/org/apache/spark/SparkEnv.scala | 44 ++++++
.../deploy/security/UserCredentialManager.scala | 23 +++-
.../executor/CoarseGrainedExecutorBackend.scala | 12 ++
.../scala/org/apache/spark/executor/Executor.scala | 9 ++
.../apache/spark/scheduler/TaskDescription.scala | 34 ++++-
.../apache/spark/scheduler/TaskSetManager.scala | 1 +
.../cluster/CoarseGrainedClusterMessage.scala | 4 +
.../cluster/CoarseGrainedSchedulerBackend.scala | 51 ++++++-
.../security/UserCredentialManagerSuite.scala | 65 +++++----
.../CoarseGrainedExecutorBackendSuite.scala | 6 +-
.../org/apache/spark/executor/ExecutorSuite.scala | 1 +
.../CoarseGrainedSchedulerBackendSuite.scala | 148 ++++++++++++++++++++-
.../spark/scheduler/TaskDescriptionSuite.scala | 65 +++++++++
.../cluster/k8s/KubernetesExecutorBackend.scala | 5 +
15 files changed, 429 insertions(+), 40 deletions(-)
diff --git a/common/utils-java/src/main/java/org/apache/spark/internal/LogKeys.java b/common/utils-java/src/main/java/org/apache/spark/internal/LogKeys.java
index 91bff3120f00..bf53e4d2b773 100644
--- a/common/utils-java/src/main/java/org/apache/spark/internal/LogKeys.java
+++ b/common/utils-java/src/main/java/org/apache/spark/internal/LogKeys.java
@@ -140,6 +140,7 @@ public enum LogKeys implements LogKey {
CREATED_POOL_NAME,
CREATION_SITE,
CREDENTIALS_RENEWAL_INTERVAL_RATIO,
+ CREDENTIAL_VERSION,
CROSS_VALIDATION_METRIC,
CROSS_VALIDATION_METRICS,
CSV_HEADER_COLUMN_NAME,
diff --git a/core/src/main/scala/org/apache/spark/SparkEnv.scala b/core/src/main/scala/org/apache/spark/SparkEnv.scala
index ca48ee473eb0..8c9be740027f 100644
--- a/core/src/main/scala/org/apache/spark/SparkEnv.scala
+++ b/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,19 @@ class SparkEnv (
private[spark] var executorBackend: Option[ExecutorBackend] = None
+ /**
+ * Versioned credential store for OIDC-based user credentials on both driver and 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).
+ *
+ * The version field is a monotonically increasing counter assigned by `UserCredentialManager`
+ * on each credential renewal. It is used to guard against stale credentials from delayed
+ * `TaskDescription` delivery overwriting fresher credentials delivered via RPC broadcast.
+ */
+ private[spark] val userCredentials: AtomicReference[VersionedCredentials] =
+ new AtomicReference[VersionedCredentials]()
+
private[spark] def stop(): Unit = {
if (!isStopped) {
@@ -825,3 +839,33 @@ object SparkEnv extends Logging {
"Metrics Properties" -> metricsProperties.toSeq.sorted)
}
}
+
+/**
+ * Container for versioned OIDC user credentials.
+ *
+ * @param version Monotonically increasing counter assigned by `UserCredentialManager` on each
+ * credential renewal. Used to prevent stale credentials from overwriting fresher
+ * ones on executors.
+ * @param bytes Serialized `UserCredentials` payload (no raw identity token).
+ */
+private[spark] case class VersionedCredentials(version: Long, bytes: Array[Byte])
+
+private[spark] object VersionedCredentials {
+ /**
+ * Atomically update a credential store only if the given version is strictly newer
+ * than what is currently stored. This prevents stale credentials (e.g., from a delayed
+ * `TaskDescription`) from overwriting fresher credentials delivered via RPC broadcast.
+ *
+ * Uses `AtomicReference.updateAndGet` to ensure the check-and-set is atomic even
+ * when called concurrently from multiple task threads and the RPC dispatcher thread.
+ */
+ def updateIfNewer(
+ store: AtomicReference[VersionedCredentials],
+ version: Long,
+ bytes: Array[Byte]): Unit = {
+ val newValue = VersionedCredentials(version, bytes)
+ store.updateAndGet { current =>
+ if (current == null || version > current.version) newValue else current
+ }
+ }
+}
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 0627a87d9436..5a8d6e46530a 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
@@ -22,6 +22,7 @@ import java.net.URI
import java.nio.file.Paths
import java.time.Instant
import java.util.concurrent.{RejectedExecutionException, ScheduledExecutorService, TimeUnit}
+import java.util.concurrent.atomic.AtomicLong
import scala.collection.mutable
import scala.jdk.CollectionConverters._
@@ -60,11 +61,17 @@ import org.apache.spark.util.{ThreadUtils, Utils}
private[spark] class UserCredentialManager(
sparkConf: SparkConf,
tokenIngestor: TokenIngestor,
- onCredentialsUpdate: Array[Byte] => Unit) extends Logging {
+ onCredentialsUpdate: (Long, Array[Byte]) => Unit) extends Logging {
private val safetyMargin = sparkConf.get(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN)
private val minInterval = sparkConf.get(SECURITY_OIDC_RENEWAL_MIN_INTERVAL)
+ // Monotonically increasing version counter for credential updates.
+ // Incremented on each successful credential acquisition (initial + renewals).
+ // Used by executors to guard against stale TaskDescription credentials overwriting
+ // fresher credentials delivered via RPC broadcast.
+ private val credentialVersion = new AtomicLong(0)
+
// Counter for exponential backoff calculation.
// Only accessed from the single-thread renewal executor.
private var consecutiveFailures: Int = 0
@@ -84,10 +91,10 @@ private[spark] class UserCredentialManager(
* no credentials can be resolved, an exception is thrown. Subsequent renewal failures
* are handled with exponential backoff.
*
- * @return The serialized initial [[UserCredentials]].
+ * @return The version and serialized initial [[UserCredentials]].
* @throws IllegalStateException if the initial credential acquisition fails.
*/
- def start(): Array[Byte] = {
+ def start(): (Long, Array[Byte]) = {
require(renewalExecutor == null, "start() must not be called more than once")
// Initial acquisition is fail-fast (no retry/backoff).
@@ -107,9 +114,10 @@ private[spark] class UserCredentialManager(
val (credentials, earliestExpiry) = resolveCredentials(ctx)
val serialized = UserCredentialManager.serializeUserCredentials(credentials)
+ val version = credentialVersion.incrementAndGet()
// Propagate initial credentials
- onCredentialsUpdate(serialized)
+ onCredentialsUpdate(version, serialized)
// Create the renewal executor only after successful initial acquisition.
// This avoids leaking a daemon thread if the fail-fast path throws, and
@@ -123,7 +131,7 @@ private[spark] class UserCredentialManager(
logInfo(log"Credential acquisition successful. Next renewal in " +
log"${MDC(LogKeys.TIME_UNITS, UIUtils.formatDuration(renewalDelay))}.")
- serialized
+ (version, serialized)
}
def stop(): Unit = {
@@ -164,11 +172,12 @@ private[spark] class UserCredentialManager(
val (credentials, earliestExpiry) = resolveCredentials(ctx)
val serialized = UserCredentialManager.serializeUserCredentials(credentials)
+ val version = credentialVersion.incrementAndGet()
// Propagate credentials to executors. Errors here are logged separately
// so that credential-fetch success is not conflated with distribution failure.
try {
- onCredentialsUpdate(serialized)
+ onCredentialsUpdate(version, serialized)
} catch {
case e: Exception =>
logWarning(log"Credentials were resolved successfully but failed to propagate " +
@@ -417,7 +426,7 @@ private[spark] object UserCredentialManager {
*/
def create(
sparkConf: SparkConf,
- onCredentialsUpdate: Array[Byte] => Unit): Option[UserCredentialManager] = {
+ onCredentialsUpdate: (Long, Array[Byte]) => Unit): Option[UserCredentialManager] = {
if (!sparkConf.get(SECURITY_OIDC_ENABLED)) {
None
} else {
diff --git a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
index 206a6a0fe385..7429774cec60 100644
--- a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
+++ b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
@@ -223,6 +223,12 @@ private[spark] class CoarseGrainedExecutorBackend(
logInfo(log"Received tokens of ${MDC(LogKeys.NUM_BYTES, tokenBytes.length)} bytes")
SparkHadoopUtil.get.addDelegationTokens(tokenBytes, env.conf)
+ case UpdateUserCredentials(version, credentials) =>
+ logInfo(log"Received user credentials of " +
+ log"${MDC(LogKeys.NUM_BYTES, credentials.length)} bytes " +
+ log"(version ${MDC(LogKeys.CREDENTIAL_VERSION, version)})")
+ VersionedCredentials.updateIfNewer(env.userCredentials, version, credentials)
+
case DecommissionExecutor =>
decommissionSelf()
}
@@ -505,6 +511,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 { case (version, credentials) =>
+ env.userCredentials.set(VersionedCredentials(version, credentials))
+ }
+
// Set the application attemptId in the BlockStoreClient if available.
val appAttemptId = env.conf.get(APP_ATTEMPT_ID)
appAttemptId.foreach(attemptId =>
diff --git a/core/src/main/scala/org/apache/spark/executor/Executor.scala b/core/src/main/scala/org/apache/spark/executor/Executor.scala
index f288ceacef7e..a67e700bc372 100644
--- a/core/src/main/scala/org/apache/spark/executor/Executor.scala
+++ b/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.
+ // Only apply if the version is newer than what the store already has, preventing
+ // a delayed TaskDescription from overwriting fresher credentials delivered via RPC.
+ taskDescription.userCredentials.foreach { case (version, creds) =>
+ VersionedCredentials.updateIfNewer(env.userCredentials, version, creds)
+ }
+
updateDependencies(
taskDescription.artifacts.files,
taskDescription.artifacts.jars,
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskDescription.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskDescription.scala
index 7eba116c5680..42b0d33d724c 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskDescription.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskDescription.scala
@@ -61,6 +61,16 @@ 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]],
+ // OIDC credentials with version for stale-update prevention.
+ // The version is a monotonic counter from UserCredentialManager; executors only apply
+ // credentials if the version is newer than what they already have. This prevents a
+ // delayed TaskDescription from overwriting fresher credentials delivered via RPC.
+ // Trade-off: every TaskDescription carries the full serialized UserCredentials (a few KB).
+ // This is acceptable because OIDC credentials are short-lived (minutes) unlike Hadoop
+ // delegation tokens (hours/days), making the race between RPC broadcast and task dispatch
+ // a practical concern. For short-task-heavy workloads, the overhead is bounded by
+ // credential size x tasks-in-flight (not total task count).
+ val userCredentials: Option[(Long, Array[Byte])],
val serializedTask: ByteBuffer) {
assert(cpus > 0, "CPUs per task should be > 0")
@@ -121,6 +131,17 @@ private[spark] object TaskDescription {
// Write resources.
serializeResources(taskDescription.resources, dataOut)
+ // Write user credentials (OIDC).
+ taskDescription.userCredentials match {
+ case Some((version, creds)) =>
+ dataOut.writeBoolean(true)
+ dataOut.writeLong(version)
+ 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 +249,21 @@ private[spark] object TaskDescription {
// Read resources.
val resources = deserializeResources(dataIn)
+ // Read user credentials (OIDC).
+ val userCredentials = if (dataIn.readBoolean()) {
+ val version = dataIn.readLong()
+ val length = dataIn.readInt()
+ val creds = new Array[Byte](length)
+ dataIn.readFully(creds)
+ Some((version, 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)
}
}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
index 4fbe197cb838..9920d7ad971f 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskSetManager.scala
@@ -604,6 +604,7 @@ private[spark] class TaskSetManager(
task.localProperties,
taskCpus,
taskResourceAssignments,
+ Option(env.userCredentials.get()).map(vc => (vc.version, vc.bytes)),
serializedTask)
}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala
index 84a6f9b3be2a..e1be2af9294b 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedClusterMessage.scala
+++ b/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[(Long, 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(version: Long, credentials: Array[Byte])
+ extends CoarseGrainedClusterMessage
+
// Executors to driver
case class RegisterExecutor(
executorId: String,
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 99fdc534a6fb..8871a4b5b880 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
@@ -26,8 +26,9 @@ import scala.concurrent.Future
import com.google.common.cache.CacheBuilder
-import org.apache.spark.{ExecutorAllocationClient, SparkEnv, TaskState}
+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}
@@ -132,6 +133,9 @@ 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")
@@ -220,6 +224,9 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
case UpdateDelegationTokens(newDelegationTokens) =>
updateDelegationTokens(newDelegationTokens)
+ case UpdateUserCredentials(version, newCredentials) =>
+ updateUserCredentials(version, 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 +365,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
sparkProperties,
SparkEnv.get.securityManager.getIOEncryptionKey(),
Option(delegationTokens.get()),
+ Option(SparkEnv.get.userCredentials.get()).map(vc => (vc.version, vc.bytes)),
rp,
currentLogLevel)
context.reply(reply)
@@ -618,6 +626,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 +657,7 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp
cleanupService.foreach(_.shutdownNow())
stopExecutors()
stopTokenManager()
+ stopUserCredentialManager()
try {
if (driverEndpoint != null) {
driverEndpoint.askSync[Boolean](StopDriver)
@@ -1047,6 +1057,45 @@ 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(version: Long, credentials: Array[Byte]): Unit = {
+ VersionedCredentials.updateIfNewer(SparkEnv.get.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.
+ */
+ private def setupUserCredentialManager(): Unit = {
+ 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))
+ })
+ 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())
+ }
+
/**
* 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.
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 5e881c8d3a3e..d30d54beb345 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
@@ -80,10 +80,10 @@ class UserCredentialManagerSuite extends SparkFunSuite {
val manager = new UserCredentialManager(
conf,
createIngestor(ctx),
- bytes => callbackRef.set(bytes))
+ (_, bytes) => callbackRef.set(bytes))
try {
- val result = manager.start()
+ val (_, result) = manager.start()
assert(result != null, "start() should return serialized credentials")
assert(callbackRef.get() != null, "callback should have been invoked")
@@ -105,7 +105,7 @@ class UserCredentialManagerSuite extends SparkFunSuite {
val manager = new UserCredentialManager(
conf,
createFailingIngestor(),
- _ => ())
+ (_: Long, _: Array[Byte]) => ())
try {
val ex = intercept[IllegalStateException] {
@@ -129,7 +129,8 @@ class UserCredentialManagerSuite extends SparkFunSuite {
val original = new UserCredentials(credsMap)
val conf = createSparkConf()
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
+ val manager = new UserCredentialManager(
+ conf, createFailingIngestor(), (_: Long, _: Array[Byte]) => ())
try {
val serialized = UserCredentialManager.serializeUserCredentials(original)
@@ -168,7 +169,8 @@ class UserCredentialManagerSuite extends SparkFunSuite {
.set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 10000L) // 10s
.set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 5000L) // 5s
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
+ val manager = new UserCredentialManager(
+ conf, createFailingIngestor(), (_: Long, _: Array[Byte]) => ())
try {
// Token expires in 60s, credential expires in 30s
// Expected: min(60s, 30s) - 10s = 20s
@@ -189,7 +191,8 @@ class UserCredentialManagerSuite extends SparkFunSuite {
.set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 10000L)
.set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 5000L)
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
+ val manager = new UserCredentialManager(
+ conf, createFailingIngestor(), (_: Long, _: Array[Byte]) => ())
try {
// Token expires in 5s, safetyMargin is 10s -> computed delay would be negative
// Should be bounded by minInterval (5s)
@@ -208,7 +211,8 @@ class UserCredentialManagerSuite extends SparkFunSuite {
.set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 10000L) // 10s
.set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 5000L) // 5s
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
+ val manager = new UserCredentialManager(
+ conf, createFailingIngestor(), (_: Long, _: Array[Byte]) => ())
try {
// Token expires in 60s, no credential expiry
// Expected: 60s - 10s = 50s
@@ -224,7 +228,8 @@ class UserCredentialManagerSuite extends SparkFunSuite {
test("computeRenewalDelay returns default when no expiry information") {
val conf = createSparkConf()
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
+ val manager = new UserCredentialManager(
+ conf, createFailingIngestor(), (_: Long, _: Array[Byte]) => ())
try {
// UserContext with null expiresAt
val ctx = new UserContext(
@@ -243,7 +248,8 @@ class UserCredentialManagerSuite extends SparkFunSuite {
val conf = createSparkConf()
.set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
+ val manager = new UserCredentialManager(
+ conf, createFailingIngestor(), (_: Long, _: Array[Byte]) => ())
try {
val failuresField = classOf[UserCredentialManager].getDeclaredField("consecutiveFailures")
failuresField.setAccessible(true)
@@ -274,7 +280,8 @@ class UserCredentialManagerSuite extends SparkFunSuite {
val conf = createSparkConf()
.set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
+ val manager = new UserCredentialManager(
+ conf, createFailingIngestor(), (_: Long, _: Array[Byte]) => ())
try {
val failuresField = classOf[UserCredentialManager].getDeclaredField("consecutiveFailures")
failuresField.setAccessible(true)
@@ -293,7 +300,8 @@ class UserCredentialManagerSuite extends SparkFunSuite {
val conf = createSparkConf()
.set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
- val manager = new UserCredentialManager(conf, createFailingIngestor(), _ => ())
+ val manager = new UserCredentialManager(
+ conf, createFailingIngestor(), (_: Long, _: Array[Byte]) => ())
try {
val failuresField = classOf[UserCredentialManager].getDeclaredField("consecutiveFailures")
failuresField.setAccessible(true)
@@ -314,13 +322,13 @@ class UserCredentialManagerSuite extends SparkFunSuite {
val conf = new SparkConf(loadDefaults = false)
.set(SECURITY_OIDC_ENABLED, false)
- val result = UserCredentialManager.create(conf, _ => ())
+ val result = UserCredentialManager.create(conf, (_: Long, _: Array[Byte]) => ())
assert(result.isEmpty)
}
test("UserCredentialManager.create returns Some when enabled with valid config") {
val conf = createSparkConf()
- val result = UserCredentialManager.create(conf, _ => ())
+ val result = UserCredentialManager.create(conf, (_: Long, _: Array[Byte]) => ())
assert(result.isDefined)
}
@@ -330,7 +338,7 @@ class UserCredentialManagerSuite extends SparkFunSuite {
// Deliberately not setting SECURITY_OIDC_IDENTITY_TOKEN_FILE
val ex = intercept[IllegalArgumentException] {
- UserCredentialManager.create(conf, _ => ())
+ UserCredentialManager.create(conf, (_: Long, _: Array[Byte]) => ())
}
assert(ex.getMessage.contains("spark.security.oidc.identityToken.file"))
}
@@ -345,10 +353,10 @@ class UserCredentialManagerSuite extends SparkFunSuite {
val manager = new UserCredentialManager(
conf,
createIngestor(ctx),
- _ => { callbackCount += 1 })
+ (_: Long, _: Array[Byte]) => { callbackCount += 1 })
try {
- val result = manager.start()
+ val (_, result) = manager.start()
assert(result != null)
assert(callbackCount === 1, "callback should be invoked once on start")
} finally {
@@ -362,7 +370,8 @@ class UserCredentialManagerSuite extends SparkFunSuite {
"org.apache.spark.security.FakeCredentialProvider")
val ctx = createUserContext()
- val manager = new UserCredentialManager(conf, createIngestor(ctx), _ => ())
+ val manager = new UserCredentialManager(
+ conf, createIngestor(ctx), (_: Long, _: Array[Byte]) => ())
manager.start()
// Should not throw
manager.stop()
@@ -386,10 +395,10 @@ class UserCredentialManagerSuite extends SparkFunSuite {
val manager = new UserCredentialManager(
conf,
createIngestor(ctx),
- bytes => callbackRef.set(bytes))
+ (_, bytes) => callbackRef.set(bytes))
try {
- val result = manager.start()
+ val (_, result) = manager.start()
assert(result != null, "start() should return serialized credentials")
// Verify that "fake" credentials were resolved despite "shared" failing
@@ -412,7 +421,8 @@ class UserCredentialManagerSuite extends SparkFunSuite {
val ctx = createUserContext()
- val manager = new UserCredentialManager(conf, createIngestor(ctx), _ => ())
+ val manager = new UserCredentialManager(
+ conf, createIngestor(ctx), (_: Long, _: Array[Byte]) => ())
try {
val ex = intercept[IllegalStateException] {
@@ -450,14 +460,14 @@ class UserCredentialManagerSuite extends SparkFunSuite {
}
}
- val callbacks = new java.util.concurrent.CopyOnWriteArrayList[Array[Byte]]()
+ val callbacks = new java.util.concurrent.CopyOnWriteArrayList[(Long, Array[Byte])]()
val manager = new UserCredentialManager(
conf,
rotatingIngestor,
- bytes => callbacks.add(bytes))
+ (version, bytes) => callbacks.add((version, bytes)))
try {
- val initial = manager.start()
+ val (_, initial) = manager.start()
assert(initial != null)
assert(callbacks.size() === 1, "Should have one callback from start()")
@@ -471,6 +481,13 @@ class UserCredentialManagerSuite extends SparkFunSuite {
// Verify that the ingestor was called more than once (rotation detected)
assert(callCount.get() >= 2,
s"TokenIngestor should have been called at least twice, got ${callCount.get()}")
+
+ // Verify version monotonicity: each callback receives a strictly increasing version
+ val versions = (0 until callbacks.size()).map(i => callbacks.get(i)._1)
+ assert(versions === versions.sorted,
+ s"Versions should be monotonically increasing: $versions")
+ assert(versions.head === 1L, "First version should be 1")
+ assert(versions(1) === 2L, "Second version should be 2")
} finally {
manager.stop()
}
@@ -487,7 +504,7 @@ class UserCredentialManagerSuite extends SparkFunSuite {
val manager = new UserCredentialManager(
conf,
createIngestor(ctx),
- bytes => callbackRef.set(bytes))
+ (_, bytes) => callbackRef.set(bytes))
manager.start()
diff --git a/core/src/test/scala/org/apache/spark/executor/CoarseGrainedExecutorBackendSuite.scala b/core/src/test/scala/org/apache/spark/executor/CoarseGrainedExecutorBackendSuite.scala
index a038d8e8613e..c7307066c78d 100644
--- a/core/src/test/scala/org/apache/spark/executor/CoarseGrainedExecutorBackendSuite.scala
+++ b/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)
diff --git a/core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala b/core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala
index 4362eb2f816f..80b0bf69d45c 100644
--- a/core/src/test/scala/org/apache/spark/executor/ExecutorSuite.scala
+++ b/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)
}
diff --git a/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
index 99b757496379..462daa046835 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
@@ -18,7 +18,7 @@
package org.apache.spark.scheduler
import java.util.Properties
-import java.util.concurrent.atomic.AtomicBoolean
+import java.util.concurrent.atomic.{AtomicBoolean, AtomicReference}
import scala.collection.mutable
import scala.concurrent.Future
@@ -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,143 @@ 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(1L, testCredentials))
+
+ // Wait for the message to be processed
+ eventually(timeout(5 seconds)) {
+ assert(mockEndpointRef1.receivedUserCredentials.isDefined)
+ assert(mockEndpointRef2.receivedUserCredentials.isDefined)
+ }
+
+ assert(mockEndpointRef1.receivedUserCredentials.get._2 === testCredentials)
+ assert(mockEndpointRef2.receivedUserCredentials.get._2 === testCredentials)
+
+ // Verify SparkEnv credential store is also updated
+ assert(SparkEnv.get.userCredentials.get().bytes === 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(1L, 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._2 === testCredentials)
+ }
+
+ test("version guard prevents stale credentials from overwriting newer ones") {
+ val conf = new SparkConf()
+ .setMaster("local-cluster[0, 3, 1024]")
+ .setAppName("test")
+
+ sc = new SparkContext(conf)
+ val backend = sc.schedulerBackend.asInstanceOf[CoarseGrainedSchedulerBackend]
+
+ // Send version 3 credentials first
+ val credsV3 = Array[Byte](30, 30, 30)
+ backend.driverEndpoint.send(UpdateUserCredentials(3L, credsV3))
+
+ eventually(timeout(5 seconds)) {
+ assert(SparkEnv.get.userCredentials.get() != null)
+ assert(SparkEnv.get.userCredentials.get().version === 3L)
+ }
+
+ // Now send version 1 (stale) -- should be rejected
+ val credsV1 = Array[Byte](10, 10, 10)
+ backend.driverEndpoint.send(UpdateUserCredentials(1L, credsV1))
+
+ // Flush the DriverEndpoint mailbox by sending a synchronous request.
+ // Since DriverEndpoint is single-threaded, when this returns, v1 has been processed.
+ backend.driverEndpoint.askSync[SparkAppConfig](
+ RetrieveSparkAppConfig(ResourceProfile.DEFAULT_RESOURCE_PROFILE_ID))
+
+ // Store should still hold v3 (stale v1 was rejected by version guard)
+ assert(SparkEnv.get.userCredentials.get().version === 3L,
+ "Stale version 1 should not overwrite newer version 3")
+
+ // Send version 5 (newer) -- should be accepted
+ val credsV5 = Array[Byte](50, 50, 50)
+ backend.driverEndpoint.send(UpdateUserCredentials(5L, credsV5))
+
+ eventually(timeout(5 seconds)) {
+ assert(SparkEnv.get.userCredentials.get().version === 5L)
+ }
+
+ // Verify version 5 credentials are in the store (not v1 or v3)
+ assert(SparkEnv.get.userCredentials.get().bytes === credsV5)
+ }
+
+ test("executor-side credential store version guard rejects stale and accepts newer") {
+ // This tests VersionedCredentials.updateIfNewer -- the same method used in
+ // CoarseGrainedExecutorBackend.receive and Executor.TaskRunner.
+ val store = new AtomicReference[VersionedCredentials]()
+
+ // Initial write to null store should succeed
+ VersionedCredentials.updateIfNewer(store, 2L, Array[Byte](20, 20))
+ assert(store.get().version === 2L)
+ assert(store.get().bytes === Array[Byte](20, 20))
+
+ // Stale version (1) should be rejected
+ VersionedCredentials.updateIfNewer(store, 1L, Array[Byte](10, 10))
+ assert(store.get().version === 2L, "Stale version should not overwrite newer")
+
+ // Same version (2) should also be rejected (strict >)
+ VersionedCredentials.updateIfNewer(store, 2L, Array[Byte](22, 22))
+ assert(store.get().version === 2L)
+ assert(store.get().bytes === Array[Byte](20, 20), "Same version should not overwrite")
+
+ // Newer version (5) should be accepted
+ VersionedCredentials.updateIfNewer(store, 5L, Array[Byte](50, 50))
+ assert(store.get().version === 5L)
+ assert(store.get().bytes === Array[Byte](50, 50))
+ }
+
private def testSubmitJob(sc: SparkContext, rdd: RDD[Int]): Unit = {
sc.submitJob(
rdd,
@@ -667,12 +804,15 @@ private[spark] class MockExecutorRpcEndpointRef(conf: SparkConf) extends RpcEndp
// scalastyle:on executioncontextglobal
var decommissionReceived = false
+ var receivedUserCredentials: Option[(Long, 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(version, creds) => receivedUserCredentials = Some((version, creds))
+ case _ =>
}
override def ask[T: ClassTag](message: Any, timeout: RpcTimeout): Future[T] = {
Future{true.asInstanceOf[T]}
diff --git a/core/src/test/scala/org/apache/spark/scheduler/TaskDescriptionSuite.scala b/core/src/test/scala/org/apache/spark/scheduler/TaskDescriptionSuite.scala
index dcb22f8b0d7d..4744b41809bb 100644
--- a/core/src/test/scala/org/apache/spark/scheduler/TaskDescriptionSuite.scala
+++ b/core/src/test/scala/org/apache/spark/scheduler/TaskDescriptionSuite.scala
@@ -85,6 +85,7 @@ class TaskDescriptionSuite extends SparkFunSuite {
originalProperties,
cpus = 2,
originalResources,
+ None,
taskBuffer
)
@@ -102,6 +103,7 @@ class TaskDescriptionSuite extends SparkFunSuite {
assert(decodedTaskDescription.properties.equals(originalTaskDescription.properties))
assert(decodedTaskDescription.cpus.equals(originalTaskDescription.cpus))
assert(decodedTaskDescription.resources === originalTaskDescription.resources)
+ assert(decodedTaskDescription.userCredentials === None)
assert(decodedTaskDescription.serializedTask.equals(taskBuffer))
}
@@ -119,6 +121,7 @@ class TaskDescriptionSuite extends SparkFunSuite {
new Properties(),
cpus,
Map.empty,
+ None,
ByteBuffer.wrap(Array[Byte](1, 2, 3, 4)))
val decoded = TaskDescription.decode(TaskDescription.encode(taskDescription))
// The decoded amount is what the executor echoes back in its status update and what the
@@ -129,4 +132,66 @@ class TaskDescriptionSuite extends SparkFunSuite {
}
}
+ test("encoding and decoding TaskDescription with userCredentials preserves credentials") {
+ val taskBuffer = ByteBuffer.wrap(Array[Byte](1, 2, 3, 4))
+ val properties = new Properties()
+ properties.put("key", "value")
+
+ val credentialBytes = Array[Byte](10, 20, 30, 40, 50)
+
+ val taskDescription = new TaskDescription(
+ taskId = 42,
+ attemptNumber = 0,
+ executorId = "exec-1",
+ name = "task with credentials",
+ index = 0,
+ partitionId = 0,
+ JobArtifactSet.emptyJobArtifactSet,
+ properties,
+ cpus = 1,
+ Map.empty,
+ Some((1L, credentialBytes)),
+ taskBuffer
+ )
+
+ val serialized = TaskDescription.encode(taskDescription)
+ val decoded = TaskDescription.decode(serialized)
+
+ assert(decoded.taskId === 42)
+ assert(decoded.name === "task with credentials")
+ assert(decoded.userCredentials.isDefined)
+ assert(decoded.userCredentials.get._1 === 1L)
+ assert(decoded.userCredentials.get._2 === credentialBytes)
+ assert(decoded.serializedTask.equals(taskBuffer))
+ }
+
+ test("encoding and decoding TaskDescription without userCredentials round-trips correctly") {
+ val taskBuffer = ByteBuffer.wrap(Array[Byte](5, 6, 7, 8))
+ val properties = new Properties()
+
+ val taskDescription = new TaskDescription(
+ taskId = 99,
+ attemptNumber = 1,
+ executorId = "exec-2",
+ name = "task without credentials",
+ index = 3,
+ partitionId = 2,
+ JobArtifactSet.emptyJobArtifactSet,
+ properties,
+ cpus = 2,
+ Map.empty,
+ None,
+ taskBuffer
+ )
+
+ val serialized = TaskDescription.encode(taskDescription)
+ val decoded = TaskDescription.decode(serialized)
+
+ assert(decoded.taskId === 99)
+ assert(decoded.attemptNumber === 1)
+ assert(decoded.name === "task without credentials")
+ assert(decoded.userCredentials === None)
+ assert(decoded.serializedTask.equals(taskBuffer))
+ }
+
}
diff --git a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBackend.scala b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBackend.scala
index e44d7e29ef60..5b8a30fdab1c 100644
--- a/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBackend.scala
+++ b/resource-managers/kubernetes/core/src/main/scala/org/apache/spark/scheduler/cluster/k8s/KubernetesExecutorBackend.scala
@@ -128,6 +128,11 @@ private[spark] object KubernetesExecutorBackend extends Logging {
val env = SparkEnv.createExecutorEnv(driverConf, execId, arguments.bindAddress,
arguments.hostname, arguments.cores, cfg.ioEncryptionKey, isLocal = false)
+ // Apply initial user credentials to the executor credential store.
+ cfg.userCredentials.foreach { case (version, credentials) =>
+ env.userCredentials.set(VersionedCredentials(version, credentials))
+ }
+
val backend = backendCreateFn(env.rpcEnv, arguments, env, cfg.resourceProfile, execId)
env.rpcEnv.setupEndpoint("Executor", backend)
arguments.workerUrl.foreach { url =>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org