sarutak opened a new pull request, #57675: URL: https://github.com/apache/spark/pull/57675
### What changes were proposed in this pull request? This PR adds the credential distribution layer for the OIDC Credential Propagation SPIP ([SPARK-57703](https://issues.apache.org/jira/browse/SPARK-57703)). This PR covers Sub-task 5 ([SPARK-57894](https://issues.apache.org/jira/browse/SPARK-57894): Implement UpdateUserCredentials RPC and backend integration) and Sub-task 6 ([SPARK-57895](https://issues.apache.org/jira/browse/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 -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
