dongjoon-hyun commented on code in PR #58737:
URL: https://github.com/apache/spark/pull/58737#discussion_r3990280415
##########
core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala:
##########
@@ -554,17 +554,14 @@ private[spark] object UserCredentialManager extends
Logging {
* raises a clear error prompting explicit configuration.
*
* @param sparkConf The Spark configuration to apply properties into. Not
modified when OIDC
- * credential propagation is disabled or when `isLocal` is
true.
- * @param isLocal Whether the application runs in local mode (no scheduler
backend that starts
- * a resolution phase).
+ * credential propagation is disabled.
* @return `Some(loader)` with the [[CredentialProviderLoader]] used, to be
passed to
* [[create]] so the resolution phase reuses the same loader; `None`
when OIDC is
- * disabled or when `isLocal` is true (no loader is allocated in
those cases).
+ * disabled (no loader is allocated in that case).
*/
def applyProviderProperties(
- sparkConf: SparkConf,
- isLocal: Boolean): Option[CredentialProviderLoader] = {
- if (!sparkConf.get(SECURITY_OIDC_ENABLED) || isLocal) {
+ sparkConf: SparkConf): Option[CredentialProviderLoader] = {
+ if (!sparkConf.get(SECURITY_OIDC_ENABLED)) {
Review Comment:
With the `isLocal` guard removed, local mode now wires the provider into the
driver's Hadoop `Configuration` here, but credentials are acquired only later,
in `LocalSchedulerBackend.start()` (via `_taskScheduler.start()`). Driver-side
FS access that happens before that point in `SparkContext` initialization sees
an empty `SparkEnv.userCredentials`:
- `setCheckpointDir` (`spark.checkpoint.dir`) and `addFile` (`spark.files` /
`spark.archives`) throw, so `SparkContext` construction fails.
- `addJar` (`spark.jars`) swallows the error, so the jar is silently dropped
and tasks later fail with `ClassNotFoundException`.
For example, `local[*]` + `spark.security.oidc.enabled=true` + the AWS
provider + `spark.checkpoint.dir=s3a://bucket/ckpt` started fine before this PR
(default credential chain), but fails after it.
This is the same limitation that already exists in cluster mode, but this PR
extends it to local mode, and the new scaladoc ("points at credentials that are
actually populated") is not accurate for this window. Could you document this,
or make sure the resolution phase runs before these accesses in local mode?
##########
core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala:
##########
@@ -137,6 +143,41 @@ private[spark] class LocalSchedulerBackend(
SparkHadoopUtil.get.addDelegationTokens(tokens, conf)
}
+ /**
+ * Start the UserCredentialManager if OIDC credential propagation is
enabled, mirroring
+ * CoarseGrainedSchedulerBackend. Runs independently of
Kerberos/HadoopDelegationTokenManager.
+ *
+ * In local mode the driver and the single executor share this JVM and the
same
+ * `SparkEnv.get.userCredentials`, so the propagation callback simply
updates that reference
+ * (there is no remote executor to message); the in-JVM Executor picks up
credentials from the
+ * same store via TaskDescription. Driver-side filesystem access uses the
provider wiring that
+ * the selection phase (UserCredentialManager.applyProviderProperties)
already applied to the
+ * driver's Hadoop Configuration.
+ */
+ private def setupUserCredentialManager(): Unit = {
+ // Reuse the loader from SparkContext's selection phase (Some when OIDC is
enabled, None
+ // otherwise). Passing the Option straight through keeps SparkContext as
the single owner of
+ // the loader: create() enforces that an enabled configuration has a
loader rather than
+ // silently allocating one here that no one would close.
+ userCredentialManager = UserCredentialManager.create(conf, { (version,
credentials) =>
+ // No remote executors in local mode; update the shared credential store
directly so that
+ // subsequently dispatched tasks (and driver-side access) observe the
new credentials.
+ VersionedCredentials.updateIfNewer(SparkEnv.get.userCredentials,
version, credentials)
Review Comment:
This callback looks up the global `SparkEnv.get` every time it fires,
instead of the env this backend belongs to.
`UserCredentialManager.stop()` waits only 10 seconds for the renewal thread.
If a renewal is stuck longer than that (e.g., a slow STS call that ignores
interrupts) and a new `SparkContext` is created in the same JVM (notebooks,
tests, Spark Connect session restart), the late renewal writes the old
application's credentials (version N >= 2) into the new application's store.
Since the new application's version counter restarts at 1, `updateIfNewer` then
rejects its own renewals until its version exceeds N, so it keeps using the
previous application's (possibly different principal's) credentials.
`CoarseGrainedSchedulerBackend` doesn't have this issue because it routes
the update through its own (already stopped) `driverEndpoint`. Capturing the
env once in `setupUserCredentialManager()` (e.g., `val env = SparkEnv.get`) and
using it in both the callback and the initial store would avoid this.
--
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]
