This is an automated email from the ASF dual-hosted git repository.
sarutak pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new 95cb4c37147d [SPARK-57894][SPARK-57895][CORE][K8S] Implement
`UpdateUserCredentials` RPC, backend integration, and initial credential
delivery
95cb4c37147d is described below
commit 95cb4c37147d1c1b6032db0a70313779608a4648
Author: Kousuke Saruta <[email protected]>
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 <[email protected]>
Signed-off-by: Kousuke Saruta <[email protected]>
(cherry picked from commit 247405ce1c77ec12e6b0b07c663ba411fcd0f795)
Signed-off-by: Kousuke Saruta <[email protected]>
---
.../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: [email protected]
For additional commands, e-mail: [email protected]
