[SPARK-57893][CORE] Implement UserCredentialManager#57387
Conversation
… renewal loop) This adds the UserCredentialManager, which orchestrates credential acquisition, renewal scheduling, and propagation triggering on the driver. It is a sibling of HadoopDelegationTokenManager and handles the OIDC credential propagation path independently. Key features: - Reads identity token via TokenIngestor - Calls CredentialProvider.resolve() for each configured scheme - Schedules renewal based on min(token expiry, credential expiry) - safetyMargin - Retries with exponential backoff on failure - Fail-fast on initial acquisition (application won't start with null credentials) - ObjectInputFilter for safe deserialization - Adds discoverAllSchemes() to CredentialProviderLoader - Adds SECURITY_CREDENTIALS_* config keys Part of SPARK-57703 (OIDC Credential Propagation SPIP), Sub-task 4.
…quality - Add binding policy exceptions for new config keys (CI fix) - Per-provider error isolation in resolveCredentials: one provider failure no longer aborts the entire resolution loop - discoverSchemes: remove dead parameter, document ambiguity handling - start(): add double-invocation guard (require) - create(): replace early return with if/else (Spark convention) - serializeUserCredentials: explicit close for both streams - Add intent comment on consecutiveFailures reset after propagation failure - Fix ObjectInputFilter test to use java.io.File (actually tests filter rejection) - Differentiate start() vs renewal log messages - Extract SYNTHETIC_TARGET_AUTHORITY constant with documentation - Add OIDC config section header in config/package.scala
- Add @volatile to renewalExecutor for visibility across threads (shutdown hook) - Add CredentialProviderLoader.resetForTesting() in beforeEach for test isolation - Fix deserializeUserCredentials: nested try/finally ensures bis is closed even if ObjectInputStream constructor throws StreamCorruptedException - Add per-provider error isolation tests: - One provider failure does not block other providers from resolving - All providers failing throws IllegalStateException
Adds a test that verifies: when the TokenIngestor returns a different UserContext on subsequent calls (simulating token file rotation), the renewal loop acquires new credentials and invokes the propagation callback. This directly covers the acceptance criterion: 'On TokenIngestor rotation, new credentials are acquired and propagated.'
New configs must declare a ConfigBindingPolicy via .withBindingPolicy() rather than being added to the frozen exceptions file. The exceptions file is a legacy list that must only shrink. All four SECURITY_CREDENTIALS_* configs use NOT_APPLICABLE since they are Spark Core runtime configs that do not affect SQL view/UDF/procedure resolution.
resetForTesting() was package-private in org.apache.spark.security but called from UserCredentialManagerSuite in org.apache.spark.deploy.security. Java's package-private access does not allow cross-package access. Make it public since it is already documented as test-only.
HyukjinKwon
left a comment
HyukjinKwon
left a comment
There was a problem hiding this comment.
0 blocking, 3 non-blocking, 2 nits.
Solid, closely tracks the HadoopDelegationTokenManager pattern; thorough tests. One correctness robustness issue on the fail-fast path is worth addressing, the rest are optional.
Correctness (1)
- UserCredentialManager.scala:91:
start()leaks the renewal executor and locks out retry on the fail-fast throw path — see inline
Suggestions (2)
- CredentialProviderLoader.java:238:
discoverAllSchemesnever uses itsconfparameter — see inline - UserCredentialManager.scala:71:
maxBackoffMsis a hardcoded magic constant vs. config-driven peers — see inline
Nits: 2 minor items (see inline comments).
PR description suggestions
- Update the "Modified files" list: it lists
configs-without-binding-policy-exceptions("Registered new config keys"), but the diff no longer touches that file — the configs usewithBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE)instead.
| * @throws IllegalStateException if the initial credential acquisition fails. | ||
| */ | ||
| def start(): Array[Byte] = { | ||
| require(renewalExecutor == null, "start() must not be called more than once") |
There was a problem hiding this comment.
start() assigns renewalExecutor (next line) before the fail-fast acquisition, which can throw at line 101 (token missing) or via resolveCredentials at line 232 (no credentials resolved). On that throw, the daemon-thread executor has been created but is never returned or stopped, and because start() opens with this require(renewalExecutor == null, ...), a caller that catches the exception can never retry — the guard now trips.
The analogue HadoopDelegationTokenManager avoids this: its updateTokensTask() catches and reschedules rather than throwing out of start(), so it never leaves start() with a live unmanaged executor.
Since the executor isn't used for the synchronous initial acquisition, the cleanest fix is to create it after the acquisition succeeds, just before scheduleRenewal(renewalDelay). Alternatively, wrap the acquisition in a try that calls renewalExecutor.shutdownNow() and nulls the field before rethrowing.
| * @param conf Spark configuration properties (used for potential future filtering) | ||
| * @return a set of all supported scheme names (lowercased), possibly empty | ||
| */ | ||
| public static Set<String> discoverAllSchemes(Map<String, String> conf) { |
There was a problem hiding this comment.
The conf parameter is never used in this method body — the Javadoc notes it's "used for potential future filtering." An unused parameter is YAGNI: consider dropping it until the filtering actually lands (the caller discoverSchemes() can add it back with the feature). Non-blocking.
| // Thread-safe counter for exponential backoff calculation. | ||
| // Accessed from both the caller of start() and the scheduled renewal thread. | ||
| private val consecutiveFailures = new AtomicInteger(0) | ||
| private val maxBackoffMs: Long = TimeUnit.MINUTES.toMillis(10) |
There was a problem hiding this comment.
maxBackoffMs is a hardcoded 10-minute magic constant, while its scheduling peers safetyMargin and minInterval are both configs (SECURITY_CREDENTIALS_RENEWAL_*). For consistency, consider making this a config too, or at least a named constant on the companion object beside DEFAULT_RENEWAL_INTERVAL_NO_EXPIRY_MS. Non-blocking.
| * safetyMargin` | ||
| * 5. Retries with exponential backoff on failure | ||
| * | ||
| * Started from `CoarseGrainedSchedulerBackend.start()` when |
There was a problem hiding this comment.
Nit: this cites CoarseGrainedSchedulerBackend.start() as the caller, but no such call site exists yet on this branch (this is a WIP subtask). It reads as accurate to someone holding the full SPIP context, but is misleading on master today. Consider softening to "intended to be started from ..." or deferring the claim to the subtask that adds the wiring.
| * Resets the cached provider list and initialization tracking. Intended for testing only. | ||
| */ | ||
| static void resetForTesting() { | ||
| public static void resetForTesting() { |
There was a problem hiding this comment.
Nit: resetForTesting() is widened from package-private to public here. This is needed because the suite lives in a different package (o.a.s.deploy.security vs this class's o.a.s.security), and it's defensible on a @Private class — but annotating it @VisibleForTesting (used elsewhere in core) would document that the widened visibility is for tests, not general use.
shrirangmhalgi
left a comment
shrirangmhalgi
left a comment
There was a problem hiding this comment.
LGTM. The design follows the HadoopDelegationTokenManager sibling pattern correctly, with improvements (@volatile on renewalExecutor, require() double-start guard, per-provider error isolation).
What I Verified:
ObjectInputFiltercovers the full serialization graph (UserCredentials → HashMap → ServiceCredential → HashMap + Instant)computeRenewalDelaycorrectly takes min(tokenExpiry, credentialExpiry) - safetyMargin- Per-provider error isolation: one STS failure doesn't block other schemes
- Exponential backoff capped at 10min with jitter
- Fail-fast on startup
Added 2 non-blocking comments -- see inline
| */ | ||
| private def renewCredentialsTask(): Unit = { | ||
| try { | ||
| val userContext = tokenIngestor.load() |
There was a problem hiding this comment.
(non-blocking): tokenIngestor.load() here does Files.readAllBytes on the renewal thread with no timeout. In Kubernetes environments with projected service account tokens, the token volume can become temporarily unresponsive under kubelet pressure - blocking the single renewal thread indefinitely and causing credentials to go stale with no retry path.
HDTM has the same pattern (no timeout on checkTGTAndReloginFromKeytab), but OIDC deployments are more likely to use network-mounted token files (FUSE, CSI drivers) than Kerberos keytabs. Would a Future-based timeout be worth tracking as a follow-up?
| // even if onCredentialsUpdate failed above: the backoff counter tracks credential | ||
| // *resolution* failures (STS/provider issues), not propagation failures. | ||
| // Propagation failures are transient and will be retried on next scheduled renewal. | ||
| consecutiveFailures.set(0) |
There was a problem hiding this comment.
nit (non-blocking): When onCredentialsUpdate throws here, consecutiveFailures is reset to 0 and the next renewal is at the full scheduled interval. This means executors could run with stale credentials for the entire renewal period even though fresh credentials are already resolved and sitting in memory.
Would a faster propagation-only retry (shorter delay, skip tokenIngestor.load() + resolveCredentials()) be worth adding for the case where resolution succeeds but distribution fails?
3 participants
What changes were proposed in this pull request?
This is a part of the SPIP: OIDC Credential Propagation (SPARK-57703).
This PR adds
UserCredentialManager, the driver-side credential renewal loop.UserCredentialManageris a sibling ofHadoopDelegationTokenManagerand handles the OIDC credential propagation path independently. Both managers run on independent threads and can both be active simultaneously.Responsibilities:
TokenIngestor.load()CredentialProvider.resolve()for each configured schemeUserCredentialsand invokes the propagation callbackmin(identity token expiry, service credential expiry) - safetyMarginKey design decisions:
UserCredentials, preventing deserialization attacks.New files:
core/.../deploy/security/UserCredentialManager.scala: Main implementationcore/.../deploy/security/UserCredentialManagerSuite.scala: Unit testsModified files:
CredentialProviderLoader.java: AddeddiscoverAllSchemes()for auto-discovery of registered schemesconfig/package.scala: Addedspark.security.credentials.*configuration keysWhy are the changes needed?
There is no component to orchestrate credential acquisition, renewal scheduling, and propagation triggering on the driver for OIDC-based environments. This is the core driver-side engine that coordinates
TokenIngestorandCredentialProviderto maintain fresh credentials.Does this PR introduce any user-facing change?
No. All new behavior is gated by
spark.security.credentials.enabled=false(default). No existing APIs are changed.How was this patch tested?
Unit tests in
UserCredentialManagerSuitecovering:computeRenewalDelayedge cases (safety margin, min interval, no expiry)create) with various configurationsWas this patch authored or co-authored using generative AI tooling?
Kiro CLI / Claude