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 a6dddaebdc16 [SPARK-57894][SPARK-57895][CORE][K8S][DOC][FOLLOWUP] Add
documentation comments and `@volatile` fix for credential delivery
a6dddaebdc16 is described below
commit a6dddaebdc16b7bc9d362cdb5525a1bdeea312ea
Author: Kousuke Saruta <[email protected]>
AuthorDate: Mon Aug 10 14:56:47 2026 +0900
[SPARK-57894][SPARK-57895][CORE][K8S][DOC][FOLLOWUP] Add documentation
comments and `@volatile` fix for credential delivery
### What changes were proposed in this pull request?
Follow-up to #57675 addressing non-blocking review comments.
1. **Document missed-update window for late-registering executors**: Added
comments explaining the narrow window where a renewal broadcast (vN+1) could
arrive between an executor's `SparkAppConfig` reply (vN) and its registration
in `executorDataMap`, and noting that the `TaskDescription` path covers this
case.
2. **Document why initial credential application uses `set()` instead of
`updateIfNewer()`**: The store is guaranteed null at executor startup (before
any RPC or task is received), so the version guard comparison is unnecessary.
3. **Add `volatile` to `MockExecutorRpcEndpointRef` test fields**:
`decommissionReceived` and `receivedUserCredentials` are written by the RPC
dispatcher thread and read by the test thread inside `eventually`, so
visibility must be guaranteed.
4. **Clarify TaskDescription overhead comment**: Qualified as "in-memory
overhead at any instant" to distinguish from cumulative network transfer cost.
### Why are the changes needed?
These address non-blocking review feedback on #57675 to improve code
documentation and test correctness.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
GA.
### Was this patch authored or co-authored using generative AI tooling?
Kiro CLI / Claude
Closes #57877 from sarutak/oidc-propagation/subtask5-followup-nits.
Authored-by: Kousuke Saruta <[email protected]>
Signed-off-by: Kousuke Saruta <[email protected]>
---
.../org/apache/spark/executor/CoarseGrainedExecutorBackend.scala | 6 ++++++
.../src/main/scala/org/apache/spark/scheduler/TaskDescription.scala | 2 +-
.../apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala | 4 ++--
.../spark/scheduler/cluster/k8s/KubernetesExecutorBackend.scala | 6 ++++++
4 files changed, 15 insertions(+), 3 deletions(-)
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 7429774cec60..375682383b1f 100644
---
a/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
+++
b/core/src/main/scala/org/apache/spark/executor/CoarseGrainedExecutorBackend.scala
@@ -513,6 +513,12 @@ private[spark] object CoarseGrainedExecutorBackend extends
Logging {
arguments.hostname, arguments.cores, cfg.ioEncryptionKey, isLocal =
false)
// Apply initial user credentials to the executor credential store.
+ // Uses unconditional set() rather than updateIfNewer() because the
store is guaranteed
+ // null at this point (executor startup, before any RPC or task is
received).
+ // Note: there is a narrow window where a renewal broadcast (vN+1) could
arrive between
+ // the SparkAppConfig reply (vN) and executor registration in
executorDataMap, leaving
+ // this executor on vN until vN+2. The TaskDescription path covers this
case since
+ // every dispatched task carries the latest credentials.
cfg.userCredentials.foreach { case (version, credentials) =>
env.userCredentials.set(VersionedCredentials(version, credentials))
}
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 42b0d33d724c..db67b69fd9d7 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/TaskDescription.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/TaskDescription.scala
@@ -69,7 +69,7 @@ private[spark] class TaskDescription(
// 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).
+ // credential size x tasks-in-flight at any instant (not total task count).
val userCredentials: Option[(Long, Array[Byte])],
val serializedTask: ByteBuffer) {
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 462daa046835..10eb884619b5 100644
---
a/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
+++
b/core/src/test/scala/org/apache/spark/scheduler/CoarseGrainedSchedulerBackendSuite.scala
@@ -803,8 +803,8 @@ private[spark] class MockExecutorRpcEndpointRef(conf:
SparkConf) extends RpcEndp
import scala.concurrent.ExecutionContext.Implicits.global
// scalastyle:on executioncontextglobal
- var decommissionReceived = false
- var receivedUserCredentials: Option[(Long, Array[Byte])] = None
+ @volatile var decommissionReceived = false
+ @volatile var receivedUserCredentials: Option[(Long, Array[Byte])] = None
override def address: RpcAddress = null
override def name: String = "executor"
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 5b8a30fdab1c..385bc82a35e2 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
@@ -129,6 +129,12 @@ private[spark] object KubernetesExecutorBackend extends
Logging {
arguments.hostname, arguments.cores, cfg.ioEncryptionKey, isLocal =
false)
// Apply initial user credentials to the executor credential store.
+ // Uses unconditional set() rather than updateIfNewer() because the
store is guaranteed
+ // null at this point (executor startup, before any RPC or task is
received).
+ // Note: there is a narrow window where a renewal broadcast (vN+1) could
arrive between
+ // the SparkAppConfig reply (vN) and executor registration in
executorDataMap, leaving
+ // this executor on vN until vN+2. The TaskDescription path covers this
case since
+ // every dispatched task carries the latest credentials.
cfg.userCredentials.foreach { case (version, credentials) =>
env.userCredentials.set(VersionedCredentials(version, credentials))
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
