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