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

[sarutak]

What changes were proposed in this pull request?

This PR adds AutoCloseable to the CredentialProvider interface so that implementations holding long-lived resources (e.g., HTTP connection pools in StsClient) can be properly cleaned up on shutdown.

• CredentialProvider now extends AutoCloseable with a default no-op close()
• CredentialProviderLoader.closeAll() closes all initialized providers with exception suppression (first exception wins, others attached via addSuppressed)
• UserCredentialManager.stop() calls closeAll() during shutdown

Why are the changes needed?

Unlike HadoopDelegationTokenProvider (which is stateless, and receives config on each obtainDelegationTokens() call), CredentialProvider uses an init() pattern where implementations construct long-lived resources. For example, AwsStsCredentialProvider (#57655) holds an StsClient with HTTP connection pools that must be closed. Without a lifecycle hook, these resources leak on application shutdown.

Adding AutoCloseable (rather than a standalone close() method) follows the same pattern as KVStore and DataWriter in Spark . It serves as a lifecycle contract marker indicating that implementations may hold resources, and the framework is responsible for calling close() at shutdown. It is not intended for use with try-with-resources.

Does this PR introduce any user-facing change?

No. The default close() is a no-op; existing CredentialProvider implementations are unaffected.

How was this patch tested?

• CredentialProviderLoaderSuite: 3 new tests for closeAll()
• UserCredentialManagerSuite: 1 new test for stop() closing providers
• All 47 tests pass

Was this patch authored or co-authored using generative AI tooling?

Kiro CLI / Claude

[dongjoon-hyun]

closeAll() invokes third-party close() implementations while holding the class lock. Since providerFor() takes the same lock on every call, a close() that blocks (e.g., draining an HTTP connection pool) stalls all credential resolutions — and if a close() implementation waits on a thread that is calling providerFor(), we get a deadlock. The javadoc contract ("must not call back into CredentialProviderLoader") only covers direct re-entry, not indirect waiting.

Since the copy-and-clear already happens under the lock, moving just the close loop outside the lock removes the risk:

Suggested change

	public static void closeAll() throws Exception {
	synchronized (CredentialProviderLoader.class) {
	// Copy and clear first to prevent double-close if closeAll() is called again
	// concurrently or re-entrantly, and to avoid ConcurrentModificationException
	// if a close() implementation were to interact with this class.
	List<CredentialProvider> toClose = new ArrayList<>(initializedProviders);
	initializedProviders.clear();
	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;
	}
	}
	}
	public static void closeAll() throws Exception {
	List<CredentialProvider> toClose;
	synchronized (CredentialProviderLoader.class) {
	// Copy and clear first to prevent double-close if closeAll() is called again,
	// and to avoid ConcurrentModificationException if a close() implementation
	// were to interact with this class.
	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;
	}
	}

[sarutak]

Thank you, @dongjoon-hyun. I've applied suggested change.