supportedSchemes();
+
+ /**
+ * Exchanges the user's identity for a short-lived service credential scoped to the
+ * given target URI.
+ *
+ * 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.
+ *
+ * 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 0000000000000..7a3d5aa57caf0
--- /dev/null
+++ b/core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java
@@ -0,0 +1,218 @@
+/*
+ * 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.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).
+ *
+ * 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}.
+ *
+ * When the configuration key {@code spark.security.credentials.provider.} 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()}.
+ *
+ * Thread-safety: 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.";
+
+ private static volatile List 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 initializedProviders =
+ Collections.newSetFromMap(new IdentityHashMap<>());
+
+ private CredentialProviderLoader() {
+ // utility class
+ }
+
+ /**
+ * Returns the {@link CredentialProvider} for the given URI scheme, applying Binding Policy A:
+ *
+ * - If no candidate supports the scheme, {@link Optional#empty()} is returned.
+ * - If {@code spark.security.credentials.provider.} 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.
+ * - If unset and exactly one candidate supports the scheme, that candidate is selected.
+ * - If unset and multiple candidates support the scheme, an
+ * {@link IllegalArgumentException} is thrown listing the candidates.
+ *
+ * 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}.
+ *
+ * Spark-internal callers pass their configuration as a {@code Map} 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 providerFor(String scheme, Map conf) {
+ Objects.requireNonNull(scheme, "scheme must not be null");
+ Objects.requireNonNull(conf, "conf must not be null");
+ String normalizedScheme = scheme.toLowerCase(Locale.ROOT);
+ List providers = getProviders();
+
+ List candidates = providers.stream()
+ .filter(p -> {
+ Set 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).
+ synchronized (CredentialProviderLoader.class) {
+ if (!initializedProviders.contains(selected)) {
+ selected.init(conf);
+ initializedProviders.add(selected);
+ }
+ }
+ return Optional.of(selected);
+ }
+
+ /**
+ * Returns the cached list of discovered providers, loading them on first access.
+ */
+ private static List getProviders() {
+ List providers = cachedProviders;
+ if (providers == null) {
+ synchronized (CredentialProviderLoader.class) {
+ providers = cachedProviders;
+ if (providers == null) {
+ providers = loadProviders();
+ cachedProviders = providers;
+ }
+ }
+ }
+ return providers;
+ }
+
+ private static List loadProviders() {
+ ClassLoader cl = Thread.currentThread().getContextClassLoader();
+ if (cl == null) {
+ cl = CredentialProvider.class.getClassLoader();
+ }
+ ServiceLoader loader = ServiceLoader.load(CredentialProvider.class, cl);
+ List 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 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 0000000000000..eb5aaf2372984
--- /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.
+ *
+ * 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 0000000000000..b4f3ae40700b0
--- /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 initConf;
+
+ @Override
+ public void init(Map conf) {
+ this.initConf = conf;
+ }
+
+ @Override
+ public Set 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 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 0000000000000..a1a0b4264d5c1
--- /dev/null
+++ b/core/src/test/java/org/apache/spark/security/CredentialProviderLoaderSuite.java
@@ -0,0 +1,327 @@
+/*
+ * 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 conf = Map.of();
+ Optional 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 conf = Map.of();
+ Optional 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 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 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 conf = Map.of(
+ "spark.security.credentials.provider.shared",
+ FakeCredentialProvider.class.getName());
+ Optional result = CredentialProviderLoader.providerFor("shared", conf);
+ assertTrue(result.isPresent());
+ assertInstanceOf(FakeCredentialProvider.class, result.get());
+ }
+
+ @Test
+ public void testSharedSchemeWithExplicitConfSelectsAnother() {
+ Map conf = Map.of(
+ "spark.security.credentials.provider.shared",
+ AnotherFakeCredentialProvider.class.getName());
+ Optional result = CredentialProviderLoader.providerFor("shared", conf);
+ assertTrue(result.isPresent());
+ assertInstanceOf(AnotherFakeCredentialProvider.class, result.get());
+ }
+
+ @Test
+ public void testConfNamingUnknownClassThrowsClearError() {
+ Map 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 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 conf = Map.of(
+ "spark.security.credentials.provider.fake",
+ FakeCredentialProvider.class.getName());
+ Optional 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 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 conf = Map.of();
+ Optional result =
+ CredentialProviderLoader.providerFor("nonexistent", conf);
+ assertFalse(result.isPresent(), "Unknown scheme should return empty");
+ }
+
+ @Test
+ public void testInitConfIsInvokedOnSelectedProvider() {
+ Map conf = new HashMap<>();
+ conf.put("spark.app.name", "test-app");
+ conf.put("custom.key", "custom-value");
+
+ Optional result = CredentialProviderLoader.providerFor("fake", conf);
+ assertTrue(result.isPresent());
+ FakeCredentialProvider fake = (FakeCredentialProvider) result.get();
+ assertNotNull(fake.getInitConf(), "init() should have been called");
+ assertEquals("test-app", fake.getInitConf().get("spark.app.name"));
+ assertEquals("custom-value", fake.getInitConf().get("custom.key"));
+ }
+
+ @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 conf1 = new HashMap<>();
+ conf1.put("spark.app.name", "first-call");
+ Map conf2 = new HashMap<>();
+ conf2.put("spark.app.name", "second-call");
+
+ Optional result1 = CredentialProviderLoader.providerFor("fake", conf1);
+ Optional 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.app.name"),
+ "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 conf) {}
+
+ @Override
+ public Set 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 conf = Map.of();
+ Optional 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 conf = Map.of();
+ Optional 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 conf = Map.of();
+ Optional 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 conf = Map.of(
+ "spark.security.credentials.provider.shared",
+ FakeCredentialProvider.class.getName());
+ Optional 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 conf = Map.of();
+ Optional result = CredentialProviderLoader.providerFor("fake", conf);
+ assertTrue(result.isPresent());
+ assertEquals(Duration.ofMinutes(15), result.get().suggestedTtl());
+ }
+}
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 0000000000000..9eb9be01446aa
--- /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 initConf;
+ private int initCount;
+
+ @Override
+ public void init(Map conf) {
+ this.initConf = conf;
+ this.initCount++;
+ }
+
+ @Override
+ public Set 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 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 0000000000000..be245cc5429a6
--- /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