This is an automated email from the ASF dual-hosted git repository.
sarutak pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new ce072651373e [SPARK-58789][CORE] Add
CredentialProvider.additionalSparkProperties() SPI and auto-config wiring
ce072651373e is described below
commit ce072651373e203efe6e39bf1285fcd3a4c5e3ca
Author: Shrirang Mhalgi <[email protected]>
AuthorDate: Thu Aug 20 05:58:34 2026 +0900
[SPARK-58789][CORE] Add CredentialProvider.additionalSparkProperties() SPI
and auto-config wiring
### What changes were proposed in this pull request?
Adds a default method `additionalSparkProperties()` to the
`CredentialProvider` interface that allows provider modules to declare Spark
configuration properties needed when the provider is active. After successful
credential acquisition, `UserCredentialManager.start()` iterates active
providers (those that successfully resolved credentials) and applies their
declared properties to `SparkConf` -- only if the user has not already set them
explicitly.
`AwsStsCredentialProvider` overrides this to return
`{"spark.hadoop.fs.s3a.aws.credentials.provider":
"org.apache.spark.security.aws.SparkOidcAwsCredentialsProvider"}`, enabling
zero-config executor-side S3A credential resolution when OIDC is enabled.
### Why are the changes needed?
This is a Sub-task of the OIDC Credential Propagation SPIP. PR #57998
(`SparkOidcAwsCredentialsProvider`) originally placed auto-config logic in
`CoarseGrainedSchedulerBackend`, which was a layer violation -- core
referencing an optional module's class name. This PR moves the mechanism to a
generic SPI where each provider module declares its own properties, keeping
vendor-specific knowledge out of core.
### Does this PR introduce _any_ user-facing change?
Yes. When `spark.security.oidc.enabled=true` and `AwsStsCredentialProvider`
is on the classpath, executors automatically use
`SparkOidcAwsCredentialsProvider` for S3A access without requiring manual
Hadoop configuration. User-provided values are never overwritten.
### How was this patch tested?
- `AwsStsCredentialProviderSuite` (56 tests) -- verifies the override
returns the correct S3A mapping
- `UserCredentialManagerSuite` (24 tests) -- verifies properties are
applied to SparkConf, user-set values are not overwritten, null map handling,
and exception isolation (provider throwing from `additionalSparkProperties()`
does not crash `start()` or block other providers)
- `OidcCredentialIntegrationSuite` (13 tests) -- exercises the full
credential acquisition flow
Tested with both Maven and SBT.
### Was this patch authored or co-authored using generative AI tooling?
Yes. Co-Authored using Claude Opus 4.8
Closes #58018 from shrirangmhalgi/SPARK-58789-additional-spark-properties.
Authored-by: Shrirang Mhalgi <[email protected]>
Signed-off-by: Kousuke Saruta <[email protected]>
(cherry picked from commit e424fc652ea41fcc7be981b005899116b80686e0)
Signed-off-by: Kousuke Saruta <[email protected]>
---
.../security/aws/AwsStsCredentialProvider.java | 7 ++
.../aws/AwsStsCredentialProviderSuite.java | 29 +++++++
.../apache/spark/security/CredentialProvider.java | 29 +++++++
.../deploy/security/UserCredentialManager.scala | 41 +++++++++-
.../security/AnotherFakeCredentialProvider.java | 14 ++++
.../spark/security/FakeCredentialProvider.java | 6 ++
.../security/UserCredentialManagerSuite.scala | 89 ++++++++++++++++++++++
7 files changed, 211 insertions(+), 4 deletions(-)
diff --git
a/connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java
b/connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java
index f8b967d1fd83..db537fee6639 100644
---
a/connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java
+++
b/connector/credential-aws/src/main/java/org/apache/spark/security/aws/AwsStsCredentialProvider.java
@@ -414,6 +414,13 @@ public class AwsStsCredentialProvider implements
CredentialProvider {
return Duration.ofMinutes(15);
}
+ @Override
+ public Map<String, String> additionalSparkProperties() {
+ return Map.of(
+ "spark.hadoop.fs.s3a.aws.credentials.provider",
+ "org.apache.spark.security.aws.SparkOidcAwsCredentialsProvider");
+ }
+
/**
* Walks the cause chain of the given throwable and checks whether the raw
token
* appears in any layer's message. Used to decide whether the original
exception
diff --git
a/connector/credential-aws/src/test/java/org/apache/spark/security/aws/AwsStsCredentialProviderSuite.java
b/connector/credential-aws/src/test/java/org/apache/spark/security/aws/AwsStsCredentialProviderSuite.java
index 5f6daa102c8a..54e54d989f4d 100644
---
a/connector/credential-aws/src/test/java/org/apache/spark/security/aws/AwsStsCredentialProviderSuite.java
+++
b/connector/credential-aws/src/test/java/org/apache/spark/security/aws/AwsStsCredentialProviderSuite.java
@@ -1002,4 +1002,33 @@ public class AwsStsCredentialProviderSuite {
.thenReturn(response);
return mockSts;
}
+
+ // ========== additionalSparkProperties() ==========
+
+ @Test
+ public void testAdditionalSparkPropertiesReturnsS3aProviderMapping() {
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider();
+ Map<String, String> props = provider.additionalSparkProperties();
+ assertEquals(1, props.size());
+ assertEquals(
+ "org.apache.spark.security.aws.SparkOidcAwsCredentialsProvider",
+ props.get("spark.hadoop.fs.s3a.aws.credentials.provider"));
+ }
+
+ @Test
+ public void testAdditionalSparkPropertiesKeyIncludesSparkHadoopPrefix() {
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider();
+ Map<String, String> props = provider.additionalSparkProperties();
+ props.keySet().forEach(key ->
+ assertTrue(key.startsWith("spark.hadoop."),
+ "Key must include spark.hadoop. prefix: " + key));
+ }
+
+ @Test
+ public void testAdditionalSparkPropertiesIsUnmodifiable() {
+ AwsStsCredentialProvider provider = new AwsStsCredentialProvider();
+ Map<String, String> props = provider.additionalSparkProperties();
+ assertThrows(UnsupportedOperationException.class,
+ () -> props.put("foo", "bar"));
+ }
}
diff --git
a/core/src/main/java/org/apache/spark/security/CredentialProvider.java
b/core/src/main/java/org/apache/spark/security/CredentialProvider.java
index a306b427535b..27015576cbfa 100644
--- a/core/src/main/java/org/apache/spark/security/CredentialProvider.java
+++ b/core/src/main/java/org/apache/spark/security/CredentialProvider.java
@@ -102,6 +102,35 @@ public interface CredentialProvider extends AutoCloseable {
return Duration.ofMinutes(15);
}
+ /**
+ * Returns additional Spark configuration properties that should be set when
this
+ * provider is active.
+ * <p>
+ * This method is called after {@link #init(Map)} and a successful
+ * {@link #resolve(UserContext, URI)} invocation. Implementations may
+ * assume that provider state is fully initialized when this is called.
+ * <p>
+ * The credential management layer applies these entries to {@code
SparkConf} after
+ * successful startup, only if the user has not already set them explicitly.
This
+ * allows provider modules to declare executor-side wiring (e.g., the Hadoop
+ * credentials provider class for a particular filesystem scheme) without
requiring
+ * core to have vendor-specific knowledge.
+ * <p>
+ * Keys must use the {@code spark.} prefix to be effective (SparkConf
convention).
+ * Keys with the {@code spark.hadoop.} prefix are propagated to executor-side
+ * Hadoop {@code Configuration} with the prefix stripped. Other {@code
spark.*}
+ * keys are applied as Spark-internal configuration.
+ * <p>
+ * The default implementation returns an empty map (no additional
properties).
+ *
+ * @return an unmodifiable map of property key-value pairs (never null).
+ * Keys and values within the map must not be {@code null}.
+ * @since 4.4.0
+ */
+ default Map<String, String> additionalSparkProperties() {
+ return Map.of();
+ }
+
/**
* Releases any resources held by this provider (e.g., HTTP clients,
connection pools).
* <p>
diff --git
a/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala
b/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala
index 5a8d6e46530a..a06bf5f1e33b 100644
---
a/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala
+++
b/core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala
@@ -112,7 +112,7 @@ private[spark] class UserCredentialManager(
log"${MDC(LogKeys.PRINCIPAL, ctx.getPrincipal)} " +
log"(issuer: ${MDC(LogKeys.URI, ctx.getIssuer)})")
- val (credentials, earliestExpiry) = resolveCredentials(ctx)
+ val (credentials, earliestExpiry, activeProviders) =
resolveCredentials(ctx)
val serialized =
UserCredentialManager.serializeUserCredentials(credentials)
val version = credentialVersion.incrementAndGet()
@@ -131,6 +131,37 @@ private[spark] class UserCredentialManager(
logInfo(log"Credential acquisition successful. Next renewal in " +
log"${MDC(LogKeys.TIME_UNITS, UIUtils.formatDuration(renewalDelay))}.")
+
+ // Apply additional Spark properties declared by active providers.
+ // Only providers that successfully resolved credentials contribute
properties.
+ // This allows provider modules to wire executor-side configuration
+ // (e.g., fs.s3a.aws.credentials.provider) without core having
+ // vendor-specific knowledge. Properties are only set if the user
+ // has not already configured them explicitly.
+ for (provider <- activeProviders) {
+ try {
+ val props = provider.additionalSparkProperties()
+ if (props != null) {
+ props.forEach { (key, value) =>
+ if (!sparkConf.contains(key)) {
+ sparkConf.set(key, value)
+ logInfo(log"Auto-configured ${MDC(LogKeys.CONFIG, key)} from " +
+ log"${MDC(LogKeys.CLASS_NAME, provider.getClass.getName)}")
+ } else {
+ logDebug(log"Skipped ${MDC(LogKeys.CONFIG, key)} from " +
+ log"${MDC(LogKeys.CLASS_NAME, provider.getClass.getName)} " +
+ log"(already configured)")
+ }
+ }
+ }
+ } catch {
+ case scala.util.control.NonFatal(e) =>
+ logWarning(log"Failed to apply additionalSparkProperties from " +
+ log"${MDC(LogKeys.CLASS_NAME, provider.getClass.getName)}. " +
+ log"Skipping.", e)
+ }
+ }
+
(version, serialized)
}
@@ -170,7 +201,7 @@ private[spark] class UserCredentialManager(
log"${MDC(LogKeys.PRINCIPAL, ctx.getPrincipal)} " +
log"(issuer: ${MDC(LogKeys.URI, ctx.getIssuer)})")
- val (credentials, earliestExpiry) = resolveCredentials(ctx)
+ val (credentials, earliestExpiry, _) = resolveCredentials(ctx)
val serialized =
UserCredentialManager.serializeUserCredentials(credentials)
val version = credentialVersion.incrementAndGet()
@@ -222,10 +253,11 @@ private[spark] class UserCredentialManager(
* @return Tuple of (UserCredentials, earliest expiry across all service
credentials)
*/
private def resolveCredentials(
- ctx: UserContext): (UserCredentials, Option[Instant]) = {
+ ctx: UserContext): (UserCredentials, Option[Instant],
Seq[CredentialProvider]) = {
val schemes = discoverSchemes()
val credentialMap = new mutable.HashMap[String, ServiceCredential]()
+ val activeProviders = new mutable.ArrayBuffer[CredentialProvider]()
var earliestExpiry: Option[Instant] = None
for (scheme <- schemes) {
@@ -245,6 +277,7 @@ private[spark] class UserCredentialManager(
log"returned null; skipping.")
} else {
credentialMap.put(scheme, credential)
+ activeProviders += provider
val expiry = credential.getExpiresAt
if (expiry != null) {
@@ -271,7 +304,7 @@ private[spark] class UserCredentialManager(
"Check that providers are on the classpath and configured
correctly.")
}
- (new UserCredentials(credentialMap.asJava), earliestExpiry)
+ (new UserCredentials(credentialMap.asJava), earliestExpiry,
activeProviders.toSeq)
}
/**
diff --git
a/core/src/test/java/org/apache/spark/security/AnotherFakeCredentialProvider.java
b/core/src/test/java/org/apache/spark/security/AnotherFakeCredentialProvider.java
index b4f3ae40700b..ccc987e7588c 100644
---
a/core/src/test/java/org/apache/spark/security/AnotherFakeCredentialProvider.java
+++
b/core/src/test/java/org/apache/spark/security/AnotherFakeCredentialProvider.java
@@ -31,6 +31,12 @@ public class AnotherFakeCredentialProvider implements
CredentialProvider {
/** Sentinel URI host that triggers a CredentialResolutionException. */
public static final String ERROR_HOST = "error.example.com";
+ /**
+ * When set to true, {@link #additionalSparkProperties()} throws a
RuntimeException.
+ * Used to test exception isolation in UserCredentialManager.
+ */
+ public static volatile boolean throwOnProperties = false;
+
private Map<String, String> initConf;
@Override
@@ -58,4 +64,12 @@ public class AnotherFakeCredentialProvider implements
CredentialProvider {
public Map<String, String> getInitConf() {
return initConf;
}
+
+ @Override
+ public Map<String, String> additionalSparkProperties() {
+ if (throwOnProperties) {
+ throw new RuntimeException("Simulated failure in
additionalSparkProperties");
+ }
+ return Map.of();
+ }
}
diff --git
a/core/src/test/java/org/apache/spark/security/FakeCredentialProvider.java
b/core/src/test/java/org/apache/spark/security/FakeCredentialProvider.java
index a733803576cf..b9108980a473 100644
--- a/core/src/test/java/org/apache/spark/security/FakeCredentialProvider.java
+++ b/core/src/test/java/org/apache/spark/security/FakeCredentialProvider.java
@@ -76,4 +76,10 @@ public class FakeCredentialProvider implements
CredentialProvider {
public int getCloseCount() {
return closeCount.get();
}
+
+ @Override
+ public Map<String, String> additionalSparkProperties() {
+ return Map.of("spark.hadoop.fs.fake.credentials.provider",
+ "org.apache.spark.security.FakeExecutorCredentialProvider");
+ }
}
diff --git
a/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala
b/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala
index d30d54beb345..14e0f1cf5a11 100644
---
a/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala
+++
b/core/src/test/scala/org/apache/spark/deploy/security/UserCredentialManagerSuite.scala
@@ -520,4 +520,93 @@ class UserCredentialManagerSuite extends SparkFunSuite {
assert(fakeProvider.getCloseCount === 1,
"stop() should close initialized providers exactly once")
}
+
+ // ========== additionalSparkProperties application ==========
+
+ test("start() applies additionalSparkProperties from active providers") {
+ val conf = createSparkConf()
+ conf.set("spark.security.oidc.provider.fake",
+ "org.apache.spark.security.FakeCredentialProvider")
+ val ctx = createUserContext()
+
+ val manager = new UserCredentialManager(
+ conf, createIngestor(ctx), (_, _) => ())
+
+ try {
+ manager.start()
+ assert(conf.get("spark.hadoop.fs.fake.credentials.provider") ===
+ "org.apache.spark.security.FakeExecutorCredentialProvider")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("start() does not overwrite user-set properties") {
+ val conf = createSparkConf()
+ conf.set("spark.security.oidc.provider.fake",
+ "org.apache.spark.security.FakeCredentialProvider")
+ // User explicitly sets the property before start()
+ conf.set("spark.hadoop.fs.fake.credentials.provider", "user.Custom")
+ val ctx = createUserContext()
+
+ val manager = new UserCredentialManager(
+ conf, createIngestor(ctx), (_, _) => ())
+
+ try {
+ manager.start()
+ // User-set value must NOT be overwritten
+ assert(conf.get("spark.hadoop.fs.fake.credentials.provider") ===
"user.Custom")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("start() handles provider returning null from
additionalSparkProperties") {
+ // AnotherFakeCredentialProvider uses default (empty map), not null.
+ // This test verifies the defensive null check doesn't crash
+ // with a provider that inherits the default empty map.
+ val conf = createSparkConf()
+ conf.set("spark.security.oidc.provider.fake",
+ "org.apache.spark.security.FakeCredentialProvider")
+ val ctx = createUserContext()
+
+ val manager = new UserCredentialManager(
+ conf, createIngestor(ctx), (_, _) => ())
+
+ try {
+ // Should not throw
+ manager.start()
+ assert(conf.contains("spark.hadoop.fs.fake.credentials.provider"))
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("start() succeeds when a provider throws from
additionalSparkProperties") {
+ // AnotherFakeCredentialProvider is configured to throw;
FakeCredentialProvider should
+ // still have its properties applied (exception isolation via NonFatal
catch).
+ val conf = createSparkConf()
+ conf.set("spark.security.oidc.provider.fake",
+ "org.apache.spark.security.FakeCredentialProvider")
+ conf.set("spark.security.oidc.provider.shared",
+ "org.apache.spark.security.AnotherFakeCredentialProvider")
+ val ctx = createUserContext()
+
+ AnotherFakeCredentialProvider.throwOnProperties = true
+ try {
+ val manager = new UserCredentialManager(
+ conf, createIngestor(ctx), (_, _) => ())
+ try {
+ // start() must not fail even though AnotherFakeCredentialProvider
throws
+ manager.start()
+ // FakeCredentialProvider's property must still be applied
+ assert(conf.get("spark.hadoop.fs.fake.credentials.provider") ===
+ "org.apache.spark.security.FakeExecutorCredentialProvider")
+ } finally {
+ manager.stop()
+ }
+ } finally {
+ AnotherFakeCredentialProvider.throwOnProperties = false
+ }
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
