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

[sarutak]

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: Implement UpdateUserCredentials RPC and backend integration) and Sub-task 6 (SPARK-57895: Implement 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) 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.

Key implementation details:

• UpdateUserCredentials(credentials: Array[Byte]): New RPC message
• SparkEnv.userCredentials: AtomicReference[Array[Byte]]: Credential store on both driver and executor
• CoarseGrainedSchedulerBackend: Manages AtomicReference, starts/stops UserCredentialManager, handles UpdateUserCredentials in DriverEndpoint.receive (thread-safe access to executorDataMap)
• CoarseGrainedExecutorBackend: Handles UpdateUserCredentials by updating SparkEnv.userCredentials; applies initial credentials from SparkAppConfig on startup
• TaskDescription: New userCredentials: Option[Array[Byte]] field with encode/decode
• Executor: Applies credentials from TaskDescription 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, SparkAppConfig.userCredentials field
CoarseGrainedSchedulerBackend.scala	updateUserCredentials(), setupUserCredentialManager(), stopUserCredentialManager()
CoarseGrainedExecutorBackend.scala	UpdateUserCredentials handler, initial credential application from SparkAppConfig
SparkEnv.scala	userCredentials: AtomicReference[Array[Byte]] credential store
TaskDescription.scala	userCredentials field, encode/decode
TaskSetManager.scala	Include current credentials in TaskDescription
Executor.scala	Apply credentials before task execution
KubernetesExecutorBackend.scala	Apply initial credentials from SparkAppConfig (same as CoarseGrainedExecutorBackend)

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 userCredentials (Some and None)
• CoarseGrainedSchedulerBackendSuite: UpdateUserCredentials broadcast to all registered executors; SparkEnv.userCredentials updated; SparkAppConfig includes credentials for late-registering executors
• Existing test updates: All TaskDescription constructor call sites updated with new userCredentials parameter

Was this patch authored or co-authored using generative AI tooling?

Kiro CLI / Claude