[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.

[dongjoon-hyun]

stop() calls shutdownNow() (interrupt only, no wait) and then closeAll() immediately, so close() can run while a renewal task is still inside resolve(). Since this is an SPI, that concurrency contract should be spelled out for implementors:

Suggested change

	* <p>
	* <p>
	* {@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.
	* <p>

[dongjoon-hyun]

Scala's NonFatal does not match InterruptedException, so if a provider's close() throws it (plausible for HTTP client shutdown waits), it escapes stop() and can disrupt the caller's shutdown path. A shutdown hook should swallow it after restoring the interrupt flag.

Suggested change

	case NonFatal(e) =>
	logWarning(log"Error closing credential providers during shutdown.", e)
	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)

[dongjoon-hyun]

+1, LGTM except two parts.

• CredentialProvider Contract Documentation
• Exception Handling in UserCredentialManager

Thank you, @sarutak .

[sarutak]

Merge Summary:

• merged into master eb3b138
• merged into branch-4.x 1623ebd

Posted by merge_spark_pr.py

[sarutak]

Merged. Thank you @dongjoon-hyun for reviewing!