[SPARK-58472][CORE] Add AutoCloseable lifecycle to CredentialProvider SPI

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,20 @@ public interface CredentialProvider {
   default Duration suggestedTtl() {
     return Duration.ofMinutes(15);
   }
+
+  /**
+   * Releases any resources held by this provider (e.g., HTTP clients, connection pools).
+   * <p>
+   * 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.
+   * <p>
+   * 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 {}
 }

core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java

@@ -251,6 +251,52 @@ public static Set<String> discoverAllSchemes() {
     return schemes;
   }
 
+  /**
+   * Closes all initialized providers, suppressing individual close exceptions.
+   * <p>
+   * 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.
+   * <p>
+   * 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).
+   * <p>
+   * <b>Contract:</b> {@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<CredentialProvider> 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<CredentialProvider> providers) {
     synchronized (CredentialProviderLoader.class) {
       cachedProviders = providers;

core/src/main/scala/org/apache/spark/deploy/security/UserCredentialManager.scala

@@ -130,6 +130,16 @@ 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 NonFatal(e) =>
+        logWarning(log"Error closing credential providers during shutdown.", e)
+    }
   }
 
   /**

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<String, String> conf = Map.of();
+    Optional<CredentialProvider> 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<String, String> conf) {}
+
+      @Override
+      public Set<String> 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<String, String> 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.

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<String, String> initConf;
-  private int initCount;
+  private volatile Map<String, String> initConf;
+  private final AtomicInteger initCount = new AtomicInteger();
+  private final AtomicInteger closeCount = new AtomicInteger();
 
   @Override
   public void init(Map<String, String> conf) {
     this.initConf = conf;
-    this.initCount++;
+    this.initCount.incrementAndGet();
   }
 
   @Override
@@ -55,13 +57,23 @@ 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<String, String> getInitConf() {
     return initConf;
   }
 
   /** 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();
   }
 }

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")
+  }
 }