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 610962a594945..aae7f0f430a9b 100644
--- a/core/src/main/java/org/apache/spark/security/CredentialProvider.java
+++ b/core/src/main/java/org/apache/spark/security/CredentialProvider.java
@@ -39,7 +39,7 @@
* @since 4.3.0
*/
@DeveloperApi
-public interface CredentialProvider {
+public interface CredentialProvider extends AutoCloseable {
/**
* Initializes this provider with configuration properties.
@@ -101,4 +101,26 @@ public interface CredentialProvider {
default Duration suggestedTtl() {
return Duration.ofMinutes(15);
}
+
+ /**
+ * Releases any resources held by this provider (e.g., HTTP clients, connection pools).
+ *
+ * Called by the credential management layer during shutdown. The default implementation
+ * is a no-op; providers that allocate long-lived resources in {@link #init(Map)} should
+ * override this method to clean them up.
+ *
+ * {@code close()} may be invoked while another thread is still executing
+ * {@link #resolve(UserContext, URI)}: shutdown interrupts the renewal thread but does
+ * not wait for in-flight calls to complete. Implementations must tolerate a concurrent
+ * or subsequent {@code resolve()} failing after resources have been released, and
+ * {@code close()} itself must not block indefinitely.
+ *
+ * Implementations that do not throw checked exceptions may narrow the {@code throws}
+ * clause in their override (e.g., declare {@code close()} with no {@code throws} or
+ * with a more specific exception type).
+ *
+ * @since 4.3.0
+ */
+ @Override
+ default void close() throws Exception {}
}
diff --git a/core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java b/core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java
index 9d6e9e6fedf08..84aaaca0a1961 100644
--- a/core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java
+++ b/core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java
@@ -251,6 +251,52 @@ public static Set discoverAllSchemes() {
return schemes;
}
+ /**
+ * Closes all initialized providers, suppressing individual close exceptions.
+ *
+ * This method iterates over all providers that have been initialized via
+ * {@link CredentialProvider#init(Map)} and calls {@link CredentialProvider#close()}
+ * on each. If any provider's {@code close()} throws, the exception is suppressed
+ * and attached to the first exception encountered. If at least one exception occurred,
+ * it is thrown after all providers have been attempted.
+ *
+ * After this method returns (normally or exceptionally), the initialization tracking
+ * is cleared, but the cached provider list is retained. This means providers would be
+ * re-initialized on the next {@link #providerFor} call (which is not expected after
+ * shutdown).
+ *
+ * Contract: {@code close()} implementations must not call back into
+ * {@code CredentialProviderLoader} methods (e.g., {@code providerFor}).
+ *
+ * @throws Exception if one or more providers threw during close
+ */
+ public static void closeAll() throws Exception {
+ List toClose;
+ synchronized (CredentialProviderLoader.class) {
+ // Copy and clear under the lock to prevent double-close if closeAll() is called
+ // again concurrently, and to avoid ConcurrentModificationException.
+ toClose = new ArrayList<>(initializedProviders);
+ initializedProviders.clear();
+ }
+ // Close outside the lock so a slow or blocking close() cannot stall
+ // providerFor() callers or deadlock against them.
+ Exception firstException = null;
+ for (CredentialProvider provider : toClose) {
+ try {
+ provider.close();
+ } catch (Exception e) {
+ if (firstException == null) {
+ firstException = e;
+ } else {
+ firstException.addSuppressed(e);
+ }
+ }
+ }
+ if (firstException != null) {
+ throw firstException;
+ }
+ }
+
/**
* Resets the cached provider list and initialization tracking. Intended for testing only.
*/
@@ -265,6 +311,7 @@ public static void resetForTesting() {
/**
* Overrides the cached provider list for testing. Intended for testing only.
*/
+ @VisibleForTesting
static void setProvidersForTesting(List providers) {
synchronized (CredentialProviderLoader.class) {
cachedProviders = providers;
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 c98771ee02659..0627a87d9436e 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
@@ -130,6 +130,19 @@ private[spark] class UserCredentialManager(
if (renewalExecutor != null) {
renewalExecutor.shutdownNow()
}
+ // Close all initialized credential providers to release resources (e.g., HTTP clients).
+ // CredentialProviderLoader.closeAll() operates on global static state, which is safe
+ // because Spark enforces a single SparkContext (and thus a single UserCredentialManager)
+ // per JVM.
+ try {
+ CredentialProviderLoader.closeAll()
+ } catch {
+ case e: InterruptedException =>
+ Thread.currentThread().interrupt()
+ logWarning(log"Interrupted while closing credential providers during shutdown.", e)
+ case NonFatal(e) =>
+ logWarning(log"Error closing credential providers during shutdown.", e)
+ }
}
/**
diff --git a/core/src/test/java/org/apache/spark/security/CredentialProviderLoaderSuite.java b/core/src/test/java/org/apache/spark/security/CredentialProviderLoaderSuite.java
index 54c64db893338..34261f2524c2c 100644
--- a/core/src/test/java/org/apache/spark/security/CredentialProviderLoaderSuite.java
+++ b/core/src/test/java/org/apache/spark/security/CredentialProviderLoaderSuite.java
@@ -366,6 +366,70 @@ public void testEmptySchemeThrowsIllegalArgument() {
assertEquals("scheme must not be empty", e.getMessage());
}
+ @Test
+ public void testCloseAllClosesInitializedProviders() throws Exception {
+ // Initialize a provider by calling providerFor
+ Map conf = Map.of();
+ Optional result = CredentialProviderLoader.providerFor("fake", conf);
+ assertTrue(result.isPresent());
+ FakeCredentialProvider fake = (FakeCredentialProvider) result.get();
+ assertEquals(0, fake.getCloseCount(), "close() not yet called");
+
+ // Call closeAll
+ CredentialProviderLoader.closeAll();
+
+ assertEquals(1, fake.getCloseCount(), "close() should be called exactly once");
+ }
+
+ @Test
+ public void testCloseAllSuppressesExceptionsAndClosesAll() throws Exception {
+ // Two providers: first throws on close, second should still be closed
+ CredentialProvider throwingProvider = new CredentialProvider() {
+ @Override
+ public void init(Map conf) {}
+
+ @Override
+ public Set supportedSchemes() {
+ return Set.of("throwing");
+ }
+
+ @Override
+ public ServiceCredential resolve(UserContext user, URI target) {
+ return new ServiceCredential(Map.of(), Instant.now().plusSeconds(60));
+ }
+
+ @Override
+ public void close() throws Exception {
+ throw new RuntimeException("Simulated close failure");
+ }
+ };
+
+ FakeCredentialProvider fakeProvider = new FakeCredentialProvider();
+
+ CredentialProviderLoader.setProvidersForTesting(
+ List.of(throwingProvider, fakeProvider));
+
+ // Initialize both by selecting them
+ Map conf = new HashMap<>();
+ conf.put("spark.security.oidc.provider.throwing", throwingProvider.getClass().getName());
+ CredentialProviderLoader.providerFor("throwing", conf);
+ CredentialProviderLoader.providerFor("fake", conf);
+
+ // closeAll should throw (from throwingProvider) but still close fakeProvider
+ Exception e = assertThrows(Exception.class,
+ () -> CredentialProviderLoader.closeAll());
+ assertTrue(e.getMessage().contains("Simulated close failure"));
+ assertEquals(1, fakeProvider.getCloseCount(),
+ "Second provider should still be closed even when first throws");
+ }
+
+ @Test
+ public void testCloseAllWithNoInitializedProvidersIsNoOp() throws Exception {
+ // No providers initialized — closeAll should not throw
+ CredentialProviderLoader.closeAll();
+ // If we reach here, no exception was thrown — success
+ }
+
@Test
public void testInitRetryAfterFailure() {
// A provider whose init() throws on the first call then succeeds on the second.
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 9eb9be01446aa..a733803576cf5 100644
--- a/core/src/test/java/org/apache/spark/security/FakeCredentialProvider.java
+++ b/core/src/test/java/org/apache/spark/security/FakeCredentialProvider.java
@@ -21,6 +21,7 @@
import java.time.Instant;
import java.util.Map;
import java.util.Set;
+import java.util.concurrent.atomic.AtomicInteger;
/**
* A fake credential provider for testing. Supports schemes "fake" and "shared".
@@ -30,13 +31,14 @@ 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;
+ private volatile Map initConf;
+ private final AtomicInteger initCount = new AtomicInteger();
+ private final AtomicInteger closeCount = new AtomicInteger();
@Override
public void init(Map conf) {
this.initConf = conf;
- this.initCount++;
+ this.initCount.incrementAndGet();
}
@Override
@@ -55,6 +57,11 @@ public ServiceCredential resolve(UserContext user, URI target)
return new ServiceCredential(Map.of("provider", "fake"), expiresAt);
}
+ @Override
+ public void close() {
+ this.closeCount.incrementAndGet();
+ }
+
/** Returns the configuration map passed to {@link #init(Map)}, or null if not yet called. */
public Map getInitConf() {
return initConf;
@@ -62,6 +69,11 @@ public Map getInitConf() {
/** Returns the number of times {@link #init(Map)} has been called. */
public int getInitCount() {
- return initCount;
+ return initCount.get();
+ }
+
+ /** Returns the number of times {@link #close()} has been called. */
+ public int getCloseCount() {
+ return closeCount.get();
}
}
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 b5ee471013b38..5e881c8d3a3ef 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
@@ -475,4 +475,32 @@ class UserCredentialManagerSuite extends SparkFunSuite {
manager.stop()
}
}
+
+ test("stop() closes initialized credential providers") {
+ val conf = createSparkConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ conf.set("spark.security.oidc.provider.fake",
+ "org.apache.spark.security.FakeCredentialProvider")
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ bytes => callbackRef.set(bytes))
+
+ manager.start()
+
+ // Get the FakeCredentialProvider instance to verify close was called
+ val providerOpt = CredentialProviderLoader.providerFor("fake",
+ new util.HashMap[String, String]())
+ assert(providerOpt.isPresent)
+ val fakeProvider = providerOpt.get().asInstanceOf[FakeCredentialProvider]
+ assert(fakeProvider.getCloseCount === 0, "close() not yet called before stop()")
+
+ manager.stop()
+
+ assert(fakeProvider.getCloseCount === 1,
+ "stop() should close initialized providers exactly once")
+ }
}