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 54af016ec8a1 [SPARK-57891][CORE] Add CredentialProvider SPI and
ServiceLoader discovery
54af016ec8a1 is described below
commit 54af016ec8a197d7ce9dc9532f4e35b7cab16be0
Author: Anupam Yadav <[email protected]>
AuthorDate: Tue Jul 14 20:33:59 2026 +0900
[SPARK-57891][CORE] Add CredentialProvider SPI and ServiceLoader discovery
### What changes were proposed in this pull request?
This is task 2 of the OIDC Credential Propagation SPIP
([SPARK-57703](https://issues.apache.org/jira/browse/SPARK-57703)), building on
the core types added in SPARK-57890. It introduces the pluggable provider SPI
in `org.apache.spark.security` (spark-core):
- `CredentialProvider` - a `DeveloperApi` interface that providers (AWS
STS, Azure, GCP, etc.) implement to exchange a user identity for a short-lived
`ServiceCredential`
- `CredentialResolutionException` - a `DeveloperApi` checked exception
thrown by `resolve`.
- `CredentialProviderLoader` - discovers implementations via
`ServiceLoader` and selects a provider per scheme. Scheme keys are normalized
to lowercase. Selection follows an explicit-configuration policy:
`spark.security.credentials.provider.<scheme>` names the fully-qualified
provider class to use for that scheme. If exactly one discovered provider
supports a scheme, no configuration is needed; if multiple support it and the
conf is unset, a clear error is raised listing the candidat [...]
A `FakeCredentialProvider` (and a second one sharing a scheme) plus a
`META-INF/services` registration are added under `core/src/test` to exercise
discovery and selection.
### Why are the changes needed?
There is currently no pluggable SPI for exchanging an identity token for
short-lived service credentials. This is the extension point the
credential-propagation framework builds on. See the SPIP design document,
Appendix A.
### Does this PR introduce _any_ user-facing change?
No behavior change. It adds new `DeveloperApi` types only; nothing uses
them until the follow-up (SPARK-57892 / the manager task).
### How was this patch tested?
New JUnit `CredentialProviderLoaderSuite` covering the acceptance criteria,
ServiceLoader discovery, provider selection/ambiguity, error cases, and
null-input guards.
### Was this patch authored or co-authored using generative AI tooling?
No.
Closes #57191 from yadavay-amzn/SPARK-57891.
Authored-by: Anupam Yadav <[email protected]>
Signed-off-by: Kousuke Saruta <[email protected]>
(cherry picked from commit 475a7718512a43ae05f4785c84a5d0a31b37018a)
Signed-off-by: Kousuke Saruta <[email protected]>
---
.../apache/spark/security/CredentialProvider.java | 104 +++++
.../spark/security/CredentialProviderLoader.java | 242 ++++++++++++
.../security/CredentialResolutionException.java | 54 +++
.../security/AnotherFakeCredentialProvider.java | 61 +++
.../security/CredentialProviderLoaderSuite.java | 423 +++++++++++++++++++++
.../spark/security/FakeCredentialProvider.java | 67 ++++
.../org.apache.spark.security.CredentialProvider | 19 +
7 files changed, 970 insertions(+)
diff --git
a/core/src/main/java/org/apache/spark/security/CredentialProvider.java
b/core/src/main/java/org/apache/spark/security/CredentialProvider.java
new file mode 100644
index 000000000000..24d5ed5c113a
--- /dev/null
+++ b/core/src/main/java/org/apache/spark/security/CredentialProvider.java
@@ -0,0 +1,104 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.security;
+
+import java.net.URI;
+import java.time.Duration;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.spark.annotation.DeveloperApi;
+
+/**
+ * :: DeveloperApi ::
+ * Service Provider Interface for credential resolution in the OIDC credential
propagation
+ * framework.
+ * <p>
+ * Implementations exchange a user's identity (represented by {@link
UserContext}) for a
+ * short-lived {@link ServiceCredential} scoped to a target URI. Providers are
discovered
+ * via {@link java.util.ServiceLoader} and selected based on the URI scheme.
+ * <p>
+ * Implementations must be thread-safe: {@code resolve()} may be called
concurrently from
+ * multiple threads after {@code init()} completes.
+ *
+ * @since 4.3.0
+ */
+@DeveloperApi
+public interface CredentialProvider {
+
+ /**
+ * Initializes this provider with configuration properties.
+ * <p>
+ * Called exactly once per provider instance by {@link
CredentialProviderLoader}
+ * (first-conf-wins semantics). Subsequent resolutions reuse the
already-initialized
+ * instance without re-calling this method. Implementations should capture
any configuration
+ * they need (e.g., endpoint URLs, role ARNs) from the provided map.
+ * <p>
+ * The configuration map passed to this method is scoped to keys starting
with
+ * {@code spark.security.credentials.} only. Keys from other subsystems are
not included,
+ * preventing accidental leakage of unrelated secrets to third-party
providers.
+ * <p>
+ * If init() throws, it may be retried on the next resolution attempt.
Implementations
+ * should be safe to call again after a prior failure.
+ *
+ * @param conf Spark configuration properties scoped to {@code
spark.security.credentials.*}
+ * keys (must not be null)
+ * @since 4.3.0
+ */
+ void init(Map<String, String> conf);
+
+ /**
+ * Returns the set of URI schemes this provider supports (e.g., {@code
{"s3a"}}).
+ * <p>
+ * Scheme values are compared case-insensitively (normalized to lowercase).
The returned
+ * set must be non-empty and stable across calls.
+ *
+ * @return a non-empty set of supported scheme names
+ * @since 4.3.0
+ */
+ Set<String> supportedSchemes();
+
+ /**
+ * Exchanges the user's identity for a short-lived service credential scoped
to the
+ * given target URI.
+ * <p>
+ * For example, an AWS implementation might call STS
AssumeRoleWithWebIdentity using
+ * the raw token from the {@link UserContext} and return temporary AWS
credentials as
+ * a {@link ServiceCredential}.
+ *
+ * @param user the authenticated user context containing the identity token
(must not be null)
+ * @param target the target URI for which credentials are requested (must
not be null)
+ * @return a short-lived service credential for the target
+ * @throws CredentialResolutionException if the credential exchange fails
+ * @since 4.3.0
+ */
+ ServiceCredential resolve(UserContext user, URI target) throws
CredentialResolutionException;
+
+ /**
+ * Returns the suggested time-to-live for credentials produced by this
provider.
+ * <p>
+ * The credential management layer uses this as a hint for refresh
scheduling.
+ * The default is 15 minutes.
+ *
+ * @return the suggested credential TTL (never null)
+ * @since 4.3.0
+ */
+ default Duration suggestedTtl() {
+ return Duration.ofMinutes(15);
+ }
+}
diff --git
a/core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java
b/core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java
new file mode 100644
index 000000000000..af2b9252698b
--- /dev/null
+++ b/core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java
@@ -0,0 +1,242 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.security;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Locale;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Optional;
+import java.util.ServiceLoader;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import org.apache.spark.annotation.Private;
+
+/**
+ * :: Private ::
+ * Discovers {@link CredentialProvider} implementations via {@link
ServiceLoader} and selects
+ * the appropriate provider for a given URI scheme using Binding Policy A
(explicit selection).
+ * <p>
+ * Provider discovery happens once (lazily on first call) and the list is
cached. Each provider
+ * is initialized exactly once per provider instance via {@link
CredentialProvider#init(Map)}
+ * with the configuration from the first call that selects it (first-conf-wins
semantics);
+ * subsequent resolutions reuse the already-initialized instance without
re-calling {@code init}.
+ * <p>
+ * When the configuration key {@code
spark.security.credentials.provider.<scheme>} is set
+ * (non-empty), the loader validates the configured fully-qualified class name
against the
+ * discovered candidates for that scheme regardless of candidate count. If the
configured class
+ * does not match any candidate, an {@link IllegalArgumentException} is thrown
listing the
+ * scheme, the configured class, and the available candidates. Only when the
configuration key
+ * is unset (or empty) do the count-based rules apply: a single candidate is
auto-selected;
+ * multiple candidates produce an ambiguity error; no candidates produce
{@code Optional.empty()}.
+ * <p>
+ * <b>Thread-safety:</b> This class uses synchronized access to the cached
provider list and
+ * initialization tracking. Callers may invoke {@link #providerFor(String,
Map)} from multiple
+ * threads. However, the returned {@link CredentialProvider} instances are not
guaranteed to be
+ * thread-safe; callers should synchronize on the provider or confine it to a
single thread.
+ *
+ * @since 4.3.0
+ */
+@Private
+public final class CredentialProviderLoader {
+
+ /**
+ * Configuration key prefix for explicit provider selection per scheme.
+ * When set (non-empty), the configured fully-qualified class name must
match a discovered
+ * provider that supports the scheme; a mismatch results in an {@link
IllegalArgumentException}.
+ */
+ private static final String CONF_PREFIX =
"spark.security.credentials.provider.";
+
+ /**
+ * Configuration key prefix used to scope the configuration passed to
+ * {@link CredentialProvider#init(Map)}. Only keys starting with this prefix
are forwarded,
+ * preventing unrelated secrets from leaking to third-party provider
implementations.
+ */
+ private static final String CREDENTIALS_CONF_PREFIX =
"spark.security.credentials.";
+
+ private static volatile List<CredentialProvider> cachedProviders;
+
+ /**
+ * Tracks which provider instances have already been initialized. Guarded by
the class lock.
+ * Uses identity semantics (reference equality) to handle multiple provider
instances correctly.
+ */
+ private static final Set<CredentialProvider> initializedProviders =
+ Collections.newSetFromMap(new IdentityHashMap<>());
+
+ private CredentialProviderLoader() {
+ // utility class
+ }
+
+ /**
+ * Returns the {@link CredentialProvider} for the given URI scheme, applying
Binding Policy A:
+ * <ol>
+ * <li>If no candidate supports the scheme, {@link Optional#empty()} is
returned.</li>
+ * <li>If {@code spark.security.credentials.provider.<scheme>} is set
(non-empty) in
+ * {@code conf}, the provider whose fully-qualified class name matches
is selected
+ * regardless of candidate count. If the configured class does not
match any candidate,
+ * an {@link IllegalArgumentException} is thrown naming the scheme,
the configured class,
+ * and the available candidates.</li>
+ * <li>If unset and exactly one candidate supports the scheme, that
candidate is selected.</li>
+ * <li>If unset and multiple candidates support the scheme, an
+ * {@link IllegalArgumentException} is thrown listing the
candidates.</li>
+ * </ol>
+ * The selected provider is initialized exactly once per provider instance
via
+ * {@link CredentialProvider#init(Map)} (first-conf-wins semantics); later
resolutions reuse
+ * the initialized instance without re-calling {@code init}. The
configuration passed to
+ * {@code init()} is scoped to {@code spark.security.credentials.*} keys
only; keys from
+ * other subsystems are filtered out to prevent secret leakage to
third-party providers.
+ * <p>
+ * Spark-internal callers pass their configuration as a {@code Map<String,
String>} for
+ * testability and to match the signature of {@link
CredentialProvider#init(Map)}.
+ *
+ * @param scheme the URI scheme (e.g., "s3a"); normalized to lowercase
+ * @param conf Spark configuration properties as a string map
+ * @return the selected provider, or empty if no provider supports the scheme
+ * @throws IllegalArgumentException if explicit selection names an unknown
or non-supporting
+ * class, or if multiple candidates exist without explicit selection
+ * @throws IllegalStateException if a provider returns null from {@code
supportedSchemes()}
+ */
+ public static Optional<CredentialProvider> providerFor(String scheme,
Map<String, String> conf) {
+ Objects.requireNonNull(scheme, "scheme must not be null");
+ Objects.requireNonNull(conf, "conf must not be null");
+ if (scheme.isEmpty()) {
+ throw new IllegalArgumentException("scheme must not be empty");
+ }
+ String normalizedScheme = scheme.toLowerCase(Locale.ROOT);
+ List<CredentialProvider> providers = getProviders();
+
+ List<CredentialProvider> candidates = providers.stream()
+ .filter(p -> {
+ Set<String> schemes = p.supportedSchemes();
+ if (schemes == null) {
+ throw new IllegalStateException(
+ "Provider " + p.getClass().getName()
+ + " returned null from supportedSchemes()");
+ }
+ return schemes.stream()
+ .anyMatch(s ->
s.toLowerCase(Locale.ROOT).equals(normalizedScheme));
+ })
+ .collect(Collectors.toList());
+
+ if (candidates.isEmpty()) {
+ return Optional.empty();
+ }
+
+ String confKey = CONF_PREFIX + normalizedScheme;
+ String explicitClass = conf.get(confKey);
+
+ CredentialProvider selected;
+ if (explicitClass != null && !explicitClass.isEmpty()) {
+ selected = candidates.stream()
+ .filter(p -> p.getClass().getName().equals(explicitClass))
+ .findFirst()
+ .orElseThrow(() -> new IllegalArgumentException(
+ "Configured credential provider class '" + explicitClass + "'
for scheme '"
+ + normalizedScheme + "' (key: " + confKey
+ + ") was not found among candidates or does not support the
scheme. "
+ + "Available candidates: "
+ + candidates.stream()
+ .map(p -> p.getClass().getName())
+ .collect(Collectors.joining(", "))));
+ } else if (candidates.size() == 1) {
+ selected = candidates.get(0);
+ } else {
+ String candidateNames = candidates.stream()
+ .map(p -> p.getClass().getName())
+ .collect(Collectors.joining(", "));
+ throw new IllegalArgumentException(
+ "Multiple credential providers found for scheme '" + normalizedScheme
+ + "'. Set " + confKey + " to one of: " + candidateNames);
+ }
+
+ // Initialize exactly once under the lock (first-conf-wins).
+ // Only pass spark.security.credentials.* keys to init() to avoid leaking
secrets
+ // from other subsystems to third-party ServiceLoader providers. This
follows the
+ // precedent of DataSourceV2Utils.extractSessionConfigs() which scopes
configuration
+ // to a specific prefix. We keep the full key (unlike
extractSessionConfigs which
+ // strips the prefix) so providers can distinguish sub-keys unambiguously.
+ synchronized (CredentialProviderLoader.class) {
+ if (!initializedProviders.contains(selected)) {
+ Map<String, String> filteredConf = new HashMap<>();
+ for (Map.Entry<String, String> entry : conf.entrySet()) {
+ if (entry.getKey().startsWith(CREDENTIALS_CONF_PREFIX)) {
+ filteredConf.put(entry.getKey(), entry.getValue());
+ }
+ }
+ selected.init(Collections.unmodifiableMap(filteredConf));
+ initializedProviders.add(selected);
+ }
+ }
+ return Optional.of(selected);
+ }
+
+ /**
+ * Returns the cached list of discovered providers, loading them on first
access.
+ */
+ private static List<CredentialProvider> getProviders() {
+ List<CredentialProvider> providers = cachedProviders;
+ if (providers == null) {
+ synchronized (CredentialProviderLoader.class) {
+ providers = cachedProviders;
+ if (providers == null) {
+ providers = loadProviders();
+ cachedProviders = providers;
+ }
+ }
+ }
+ return providers;
+ }
+
+ private static List<CredentialProvider> loadProviders() {
+ ClassLoader cl = Thread.currentThread().getContextClassLoader();
+ if (cl == null) {
+ cl = CredentialProvider.class.getClassLoader();
+ }
+ ServiceLoader<CredentialProvider> loader =
ServiceLoader.load(CredentialProvider.class, cl);
+ List<CredentialProvider> result = new ArrayList<>();
+ for (CredentialProvider provider : loader) {
+ result.add(provider);
+ }
+ return result;
+ }
+
+ /**
+ * Resets the cached provider list and initialization tracking. Intended for
testing only.
+ */
+ static void resetForTesting() {
+ synchronized (CredentialProviderLoader.class) {
+ cachedProviders = null;
+ initializedProviders.clear();
+ }
+ }
+
+ /**
+ * Overrides the cached provider list for testing. Intended for testing only.
+ */
+ static void setProvidersForTesting(List<CredentialProvider> providers) {
+ synchronized (CredentialProviderLoader.class) {
+ cachedProviders = providers;
+ initializedProviders.clear();
+ }
+ }
+}
diff --git
a/core/src/main/java/org/apache/spark/security/CredentialResolutionException.java
b/core/src/main/java/org/apache/spark/security/CredentialResolutionException.java
new file mode 100644
index 000000000000..eb5aaf237298
--- /dev/null
+++
b/core/src/main/java/org/apache/spark/security/CredentialResolutionException.java
@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.security;
+
+import org.apache.spark.annotation.DeveloperApi;
+
+/**
+ * :: DeveloperApi ::
+ * Thrown when a {@link CredentialProvider} fails to resolve credentials for a
target URI.
+ * <p>
+ * This is a checked exception to ensure callers handle credential resolution
failures
+ * explicitly (e.g., retry, fail the job, or fall back to another mechanism).
+ *
+ * @since 4.3.0
+ */
+@DeveloperApi
+public class CredentialResolutionException extends Exception {
+
+ private static final long serialVersionUID = 1L;
+
+ /**
+ * Constructs a new exception with the specified detail message.
+ *
+ * @param message the detail message
+ */
+ public CredentialResolutionException(String message) {
+ super(message);
+ }
+
+ /**
+ * Constructs a new exception with the specified detail message and cause.
+ *
+ * @param message the detail message
+ * @param cause the underlying cause
+ */
+ public CredentialResolutionException(String message, Throwable cause) {
+ super(message, cause);
+ }
+}
diff --git
a/core/src/test/java/org/apache/spark/security/AnotherFakeCredentialProvider.java
b/core/src/test/java/org/apache/spark/security/AnotherFakeCredentialProvider.java
new file mode 100644
index 000000000000..b4f3ae40700b
--- /dev/null
+++
b/core/src/test/java/org/apache/spark/security/AnotherFakeCredentialProvider.java
@@ -0,0 +1,61 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.security;
+
+import java.net.URI;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A second fake credential provider for testing. Supports only the "shared"
scheme
+ * to create ambiguity with {@link FakeCredentialProvider}.
+ */
+public class AnotherFakeCredentialProvider implements CredentialProvider {
+
+ /** Sentinel URI host that triggers a CredentialResolutionException. */
+ public static final String ERROR_HOST = "error.example.com";
+
+ private Map<String, String> initConf;
+
+ @Override
+ public void init(Map<String, String> conf) {
+ this.initConf = conf;
+ }
+
+ @Override
+ public Set<String> supportedSchemes() {
+ return Set.of("shared");
+ }
+
+ @Override
+ public ServiceCredential resolve(UserContext user, URI target)
+ throws CredentialResolutionException {
+ if (target.getHost() != null && target.getHost().equals(ERROR_HOST)) {
+ throw new CredentialResolutionException(
+ "Simulated failure from AnotherFakeCredentialProvider for target: "
+ target);
+ }
+ Instant expiresAt = Instant.now().plus(suggestedTtl());
+ return new ServiceCredential(Map.of("provider", "another"), expiresAt);
+ }
+
+ /** Returns the configuration map passed to {@link #init(Map)}, or null if
not yet called. */
+ public Map<String, String> getInitConf() {
+ return initConf;
+ }
+}
diff --git
a/core/src/test/java/org/apache/spark/security/CredentialProviderLoaderSuite.java
b/core/src/test/java/org/apache/spark/security/CredentialProviderLoaderSuite.java
new file mode 100644
index 000000000000..d500df40c83a
--- /dev/null
+++
b/core/src/test/java/org/apache/spark/security/CredentialProviderLoaderSuite.java
@@ -0,0 +1,423 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.security;
+
+import java.net.URI;
+import java.time.Duration;
+import java.time.Instant;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertInstanceOf;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Tests for {@link CredentialProviderLoader} covering ServiceLoader discovery,
+ * single-candidate resolution, ambiguity handling, explicit selection, and
error cases.
+ */
+public class CredentialProviderLoaderSuite {
+
+ @BeforeEach
+ public void setUp() {
+ CredentialProviderLoader.resetForTesting();
+ }
+
+ @Test
+ public void testServiceLoaderDiscoversFakeProviders() {
+ // The "fake" scheme is supported only by FakeCredentialProvider (single
candidate).
+ // If discovery works, providerFor should find it.
+ Map<String, String> conf = Map.of();
+ Optional<CredentialProvider> result =
CredentialProviderLoader.providerFor("fake", conf);
+ assertTrue(result.isPresent(), "ServiceLoader should discover
FakeCredentialProvider");
+ assertInstanceOf(FakeCredentialProvider.class, result.get());
+ }
+
+ @Test
+ public void testSingleCandidateSchemeResolvesWithNoConf() {
+ // "fake" is supported only by FakeCredentialProvider
+ Map<String, String> conf = Map.of();
+ Optional<CredentialProvider> result =
CredentialProviderLoader.providerFor("fake", conf);
+ assertTrue(result.isPresent());
+ assertInstanceOf(FakeCredentialProvider.class, result.get());
+ }
+
+ @Test
+ public void testSharedSchemeWithNoConfThrowsAmbiguity() {
+ // "shared" is supported by both FakeCredentialProvider and
AnotherFakeCredentialProvider
+ Map<String, String> conf = Map.of();
+ IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
+ () -> CredentialProviderLoader.providerFor("shared", conf));
+ assertTrue(e.getMessage().contains("Multiple credential providers"),
+ "Should mention multiple providers: " + e.getMessage());
+ assertTrue(e.getMessage().contains("shared"),
+ "Should mention the scheme: " + e.getMessage());
+
assertTrue(e.getMessage().contains("spark.security.credentials.provider.shared"),
+ "Should mention the config key: " + e.getMessage());
+ assertTrue(e.getMessage().contains(FakeCredentialProvider.class.getName()),
+ "Should list FakeCredentialProvider: " + e.getMessage());
+
assertTrue(e.getMessage().contains(AnotherFakeCredentialProvider.class.getName()),
+ "Should list AnotherFakeCredentialProvider: " + e.getMessage());
+ }
+
+ @Test
+ public void testEmptyStringConfTreatedAsUnsetThrowsAmbiguity() {
+ // An empty-string value for the explicit provider conf key should be
equivalent to unset,
+ // meaning the ambiguity error is still raised for multi-candidate schemes.
+ Map<String, String> conf = new HashMap<>();
+ conf.put("spark.security.credentials.provider.shared", "");
+ IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
+ () -> CredentialProviderLoader.providerFor("shared", conf));
+ assertTrue(e.getMessage().contains("Multiple credential providers"),
+ "Empty conf value should behave as unset: " + e.getMessage());
+ }
+
+ @Test
+ public void testSharedSchemeWithExplicitConfSelectsFake() {
+ Map<String, String> conf = Map.of(
+ "spark.security.credentials.provider.shared",
+ FakeCredentialProvider.class.getName());
+ Optional<CredentialProvider> result =
CredentialProviderLoader.providerFor("shared", conf);
+ assertTrue(result.isPresent());
+ assertInstanceOf(FakeCredentialProvider.class, result.get());
+ }
+
+ @Test
+ public void testSharedSchemeWithExplicitConfSelectsAnother() {
+ Map<String, String> conf = Map.of(
+ "spark.security.credentials.provider.shared",
+ AnotherFakeCredentialProvider.class.getName());
+ Optional<CredentialProvider> result =
CredentialProviderLoader.providerFor("shared", conf);
+ assertTrue(result.isPresent());
+ assertInstanceOf(AnotherFakeCredentialProvider.class, result.get());
+ }
+
+ @Test
+ public void testConfNamingUnknownClassThrowsClearError() {
+ Map<String, String> conf = Map.of(
+ "spark.security.credentials.provider.fake",
+ "com.example.NonExistentProvider");
+ IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
+ () -> CredentialProviderLoader.providerFor("fake", conf));
+ assertTrue(e.getMessage().contains("com.example.NonExistentProvider"),
+ "Should mention the configured class: " + e.getMessage());
+ assertTrue(e.getMessage().contains("fake"),
+ "Should mention the scheme: " + e.getMessage());
+ assertTrue(e.getMessage().contains(FakeCredentialProvider.class.getName()),
+ "Should list available candidate(s): " + e.getMessage());
+ }
+
+ @Test
+ public void testConfNamingNonSupportingClassThrowsClearError() {
+ // AnotherFakeCredentialProvider does NOT support "fake" scheme
+ Map<String, String> conf = Map.of(
+ "spark.security.credentials.provider.fake",
+ AnotherFakeCredentialProvider.class.getName());
+ IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
+ () -> CredentialProviderLoader.providerFor("fake", conf));
+
assertTrue(e.getMessage().contains(AnotherFakeCredentialProvider.class.getName()),
+ "Should mention the configured class: " + e.getMessage());
+ assertTrue(e.getMessage().contains("fake"),
+ "Should mention the scheme: " + e.getMessage());
+ assertTrue(e.getMessage().contains(FakeCredentialProvider.class.getName()),
+ "Should list available candidate(s): " + e.getMessage());
+ }
+
+ @Test
+ public void testSingleCandidateWithCorrectExplicitConfSelectsIt() {
+ // "fake" is supported only by FakeCredentialProvider; conf names the
correct class.
+ Map<String, String> conf = Map.of(
+ "spark.security.credentials.provider.fake",
+ FakeCredentialProvider.class.getName());
+ Optional<CredentialProvider> result =
CredentialProviderLoader.providerFor("fake", conf);
+ assertTrue(result.isPresent());
+ assertInstanceOf(FakeCredentialProvider.class, result.get());
+ }
+
+ @Test
+ public void testSingleCandidateWithWrongExplicitConfThrowsClearError() {
+ // "fake" is supported only by FakeCredentialProvider but conf names a
different class.
+ // This validates that explicit conf is enforced even for single-candidate
schemes.
+ Map<String, String> conf = Map.of(
+ "spark.security.credentials.provider.fake",
+ "org.apache.spark.security.SomeOtherProvider");
+ IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
+ () -> CredentialProviderLoader.providerFor("fake", conf));
+ assertTrue(e.getMessage().contains("fake"),
+ "Should mention the scheme: " + e.getMessage());
+
assertTrue(e.getMessage().contains("org.apache.spark.security.SomeOtherProvider"),
+ "Should mention the configured class: " + e.getMessage());
+ assertTrue(e.getMessage().contains(FakeCredentialProvider.class.getName()),
+ "Should list the available candidate: " + e.getMessage());
+ }
+
+ @Test
+ public void testUnknownSchemeReturnsEmpty() {
+ Map<String, String> conf = Map.of();
+ Optional<CredentialProvider> result =
+ CredentialProviderLoader.providerFor("nonexistent", conf);
+ assertFalse(result.isPresent(), "Unknown scheme should return empty");
+ }
+
+ @Test
+ public void testInitConfIsInvokedOnSelectedProvider() {
+ Map<String, String> conf = new HashMap<>();
+ conf.put("spark.security.credentials.endpoint", "https://sts.example.com";);
+ conf.put("spark.security.credentials.roleArn",
"arn:aws:iam::123456:role/test");
+
+ Optional<CredentialProvider> result =
CredentialProviderLoader.providerFor("fake", conf);
+ assertTrue(result.isPresent());
+ FakeCredentialProvider fake = (FakeCredentialProvider) result.get();
+ assertNotNull(fake.getInitConf(), "init() should have been called");
+ assertEquals("https://sts.example.com";,
+ fake.getInitConf().get("spark.security.credentials.endpoint"));
+ assertEquals("arn:aws:iam::123456:role/test",
+ fake.getInitConf().get("spark.security.credentials.roleArn"));
+ }
+
+ @Test
+ public void testProviderInitializedExactlyOnce() {
+ // Call providerFor twice for the same scheme and assert:
+ // (a) the SAME provider instance is returned
+ // (b) init was invoked EXACTLY ONCE
+ Map<String, String> conf1 = new HashMap<>();
+ conf1.put("spark.security.credentials.tag", "first-call");
+ Map<String, String> conf2 = new HashMap<>();
+ conf2.put("spark.security.credentials.tag", "second-call");
+
+ Optional<CredentialProvider> result1 =
CredentialProviderLoader.providerFor("fake", conf1);
+ Optional<CredentialProvider> result2 =
CredentialProviderLoader.providerFor("fake", conf2);
+
+ assertTrue(result1.isPresent());
+ assertTrue(result2.isPresent());
+ assertSame(result1.get(), result2.get(),
+ "providerFor should return the same cached instance");
+
+ FakeCredentialProvider fake = (FakeCredentialProvider) result1.get();
+ assertEquals(1, fake.getInitCount(),
+ "init() should be called exactly once (first-conf-wins)");
+ assertEquals("first-call",
fake.getInitConf().get("spark.security.credentials.tag"),
+ "First call's conf should win");
+ }
+
+ @Test
+ public void testNullSupportedSchemesThrowsClearError() {
+ // Inject a provider that returns null from supportedSchemes() to verify
the guard.
+ CredentialProvider nullSchemesProvider = new CredentialProvider() {
+ @Override
+ public void init(Map<String, String> conf) {}
+
+ @Override
+ public Set<String> supportedSchemes() {
+ return null;
+ }
+
+ @Override
+ public ServiceCredential resolve(UserContext user, URI target) {
+ return null;
+ }
+ };
+ CredentialProviderLoader.setProvidersForTesting(
+ List.of(nullSchemesProvider));
+
+ IllegalStateException e = assertThrows(IllegalStateException.class,
+ () -> CredentialProviderLoader.providerFor("anything", Map.of()));
+ assertTrue(e.getMessage().contains("returned null from
supportedSchemes()"),
+ "Should have a clear null-schemes message: " + e.getMessage());
+ }
+
+ @Test
+ public void testResolveReturnsExpectedServiceCredential() throws Exception {
+ Map<String, String> conf = Map.of();
+ Optional<CredentialProvider> result =
CredentialProviderLoader.providerFor("fake", conf);
+ assertTrue(result.isPresent());
+
+ UserContext user = new UserContext(
+ "testuser", "https://idp.example.com";, "token", Instant.now(), null);
+ URI target = URI.create("fake://bucket/path");
+ ServiceCredential cred = result.get().resolve(user, target);
+
+ assertNotNull(cred);
+ assertEquals("fake", cred.getProperties().get("provider"));
+ assertNotNull(cred.getExpiresAt(), "expiresAt should be set");
+ }
+
+ @Test
+ public void testResolveSentinelThrowsCredentialResolutionException() {
+ Map<String, String> conf = Map.of();
+ Optional<CredentialProvider> result =
CredentialProviderLoader.providerFor("fake", conf);
+ assertTrue(result.isPresent());
+
+ UserContext user = new UserContext(
+ "testuser", "https://idp.example.com";, "token", Instant.now(), null);
+ URI errorTarget = URI.create("fake://error.example.com/path");
+
+ CredentialResolutionException e =
assertThrows(CredentialResolutionException.class,
+ () -> result.get().resolve(user, errorTarget));
+ assertTrue(e.getMessage().contains("error.example.com"),
+ "Exception should reference the target: " + e.getMessage());
+ }
+
+ @Test
+ public void testSchemeNormalizationIsCaseInsensitive() {
+ // "FAKE" should resolve the same as "fake"
+ Map<String, String> conf = Map.of();
+ Optional<CredentialProvider> result =
CredentialProviderLoader.providerFor("FAKE", conf);
+ assertTrue(result.isPresent());
+ assertInstanceOf(FakeCredentialProvider.class, result.get());
+ }
+
+ @Test
+ public void testExplicitSelectionWithUppercaseSchemeNormalizesConfKey() {
+ // The conf key uses normalized (lowercase) scheme
+ Map<String, String> conf = Map.of(
+ "spark.security.credentials.provider.shared",
+ FakeCredentialProvider.class.getName());
+ Optional<CredentialProvider> result =
CredentialProviderLoader.providerFor("SHARED", conf);
+ assertTrue(result.isPresent());
+ assertInstanceOf(FakeCredentialProvider.class, result.get());
+ }
+
+ @Test
+ public void testNullSchemeThrowsNPE() {
+ NullPointerException e = assertThrows(NullPointerException.class,
+ () -> CredentialProviderLoader.providerFor(null, Map.of()));
+ assertTrue(e.getMessage().contains("scheme must not be null"),
+ "Should have a clear message: " + e.getMessage());
+ }
+
+ @Test
+ public void testNullConfThrowsNPE() {
+ NullPointerException e = assertThrows(NullPointerException.class,
+ () -> CredentialProviderLoader.providerFor("fake", null));
+ assertTrue(e.getMessage().contains("conf must not be null"),
+ "Should have a clear message: " + e.getMessage());
+ }
+
+ @Test
+ public void testSuggestedTtlDefaultValue() {
+ Map<String, String> conf = Map.of();
+ Optional<CredentialProvider> result =
CredentialProviderLoader.providerFor("fake", conf);
+ assertTrue(result.isPresent());
+ assertEquals(Duration.ofMinutes(15), result.get().suggestedTtl());
+ }
+
+ @Test
+ public void testInitConfScopedToCredentialsKeysOnly() {
+ // Verify that init() receives only spark.security.credentials.* keys,
+ // and foreign secrets from other subsystems are NOT leaked to providers.
+ Map<String, String> conf = new HashMap<>();
+ conf.put("spark.security.credentials.provider.fake",
+ FakeCredentialProvider.class.getName());
+ conf.put("spark.security.credentials.endpoint", "https://sts.example.com";);
+ conf.put("spark.authenticate.secret", "TOPSECRET");
+ conf.put("spark.ssl.keyPassword", "keypass");
+
+ Optional<CredentialProvider> result =
CredentialProviderLoader.providerFor("fake", conf);
+ assertTrue(result.isPresent());
+ FakeCredentialProvider fake = (FakeCredentialProvider) result.get();
+ Map<String, String> initConf = fake.getInitConf();
+ assertNotNull(initConf, "init() should have been called");
+
+ // Credentials keys should be present
+ assertEquals("https://sts.example.com";,
+ initConf.get("spark.security.credentials.endpoint"));
+
assertTrue(initConf.containsKey("spark.security.credentials.provider.fake"),
+ "Provider selection key should be included (it starts with the
prefix)");
+
+ // Foreign secrets must NOT be present
+ assertFalse(initConf.containsKey("spark.authenticate.secret"),
+ "Foreign secret should not leak to provider init()");
+ assertFalse(initConf.containsKey("spark.ssl.keyPassword"),
+ "Foreign secret should not leak to provider init()");
+ assertFalse(initConf.values().contains("TOPSECRET"),
+ "Foreign secret value should not leak to provider init()");
+ }
+
+ @Test
+ public void testEmptySchemeThrowsIllegalArgument() {
+ IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
+ () -> CredentialProviderLoader.providerFor("", Map.of()));
+ assertEquals("scheme must not be empty", e.getMessage());
+ }
+
+ @Test
+ public void testInitRetryAfterFailure() {
+ // A provider whose init() throws on the first call then succeeds on the
second.
+ // First providerFor should propagate the failure; second providerFor
should retry
+ // init() and succeed.
+ CredentialProvider failOnceThenSucceed = new CredentialProvider() {
+ private int initAttempts = 0;
+ private boolean initialized = false;
+
+ @Override
+ public void init(Map<String, String> conf) {
+ initAttempts++;
+ if (initAttempts == 1) {
+ throw new RuntimeException("Simulated transient init failure");
+ }
+ initialized = true;
+ }
+
+ @Override
+ public Set<String> supportedSchemes() {
+ return Set.of("retryscheme");
+ }
+
+ @Override
+ public ServiceCredential resolve(UserContext user, URI target) {
+ return new ServiceCredential(Map.of("ok", "true"),
Instant.now().plusSeconds(60));
+ }
+
+ @Override
+ public String toString() {
+ return "initAttempts=" + initAttempts + ",initialized=" + initialized;
+ }
+ };
+
+
CredentialProviderLoader.setProvidersForTesting(List.of(failOnceThenSucceed));
+
+ // First call: init() throws, providerFor should propagate
+ Map<String, String> conf = Map.of();
+ RuntimeException e = assertThrows(RuntimeException.class,
+ () -> CredentialProviderLoader.providerFor("retryscheme", conf));
+ assertTrue(e.getMessage().contains("Simulated transient init failure"));
+
+ // Second call: init() should be retried and succeed
+ Optional<CredentialProvider> result =
+ CredentialProviderLoader.providerFor("retryscheme", conf);
+ assertTrue(result.isPresent(), "Second providerFor should succeed after
init retry");
+
+ // Verify init was called exactly twice (proving the retry)
+ String state = result.get().toString();
+ assertTrue(state.contains("initAttempts=2"),
+ "init() should have been invoked twice: " + state);
+ assertTrue(state.contains("initialized=true"),
+ "Provider should be initialized after retry: " + state);
+ }
+}
diff --git
a/core/src/test/java/org/apache/spark/security/FakeCredentialProvider.java
b/core/src/test/java/org/apache/spark/security/FakeCredentialProvider.java
new file mode 100644
index 000000000000..9eb9be01446a
--- /dev/null
+++ b/core/src/test/java/org/apache/spark/security/FakeCredentialProvider.java
@@ -0,0 +1,67 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.security;
+
+import java.net.URI;
+import java.time.Instant;
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * A fake credential provider for testing. Supports schemes "fake" and
"shared".
+ */
+public class FakeCredentialProvider implements CredentialProvider {
+
+ /** Sentinel URI host that triggers a CredentialResolutionException. */
+ public static final String ERROR_HOST = "error.example.com";
+
+ private Map<String, String> initConf;
+ private int initCount;
+
+ @Override
+ public void init(Map<String, String> conf) {
+ this.initConf = conf;
+ this.initCount++;
+ }
+
+ @Override
+ public Set<String> supportedSchemes() {
+ return Set.of("fake", "shared");
+ }
+
+ @Override
+ public ServiceCredential resolve(UserContext user, URI target)
+ throws CredentialResolutionException {
+ if (target.getHost() != null && target.getHost().equals(ERROR_HOST)) {
+ throw new CredentialResolutionException(
+ "Simulated failure for target: " + target);
+ }
+ Instant expiresAt = Instant.now().plus(suggestedTtl());
+ return new ServiceCredential(Map.of("provider", "fake"), expiresAt);
+ }
+
+ /** Returns the configuration map passed to {@link #init(Map)}, or null if
not yet called. */
+ public Map<String, String> getInitConf() {
+ return initConf;
+ }
+
+ /** Returns the number of times {@link #init(Map)} has been called. */
+ public int getInitCount() {
+ return initCount;
+ }
+}
diff --git
a/core/src/test/resources/META-INF/services/org.apache.spark.security.CredentialProvider
b/core/src/test/resources/META-INF/services/org.apache.spark.security.CredentialProvider
new file mode 100644
index 000000000000..be245cc5429a
--- /dev/null
+++
b/core/src/test/resources/META-INF/services/org.apache.spark.security.CredentialProvider
@@ -0,0 +1,19 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+org.apache.spark.security.FakeCredentialProvider
+org.apache.spark.security.AnotherFakeCredentialProvider
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
