[SPARK-58472][CORE][FOLLOWUP] Scope credential providers to manager lifecycle

[cloud-fan]

What changes were proposed in this pull request?

Followup to #57677.

This PR scopes CredentialProviderLoader state to each UserCredentialManager. It converts
provider discovery and initialization tracking from static state to loader-instance state, prevents
provider reinitialization after closeAll(), and gives each manager its own loader while retaining
the existing constructor for callers.

It also adds coverage for sequential managers and for shutdown while credential renewal is in
flight.

Why are the changes needed?

The original change stores providers in global static state. After one manager closes those
providers, a later manager in the same JVM can rediscover and reinitialize the released provider
instances. Instance-scoped loaders align provider ownership with the manager lifecycle and allow a
later manager to discover fresh providers. Marking a loader closed also prevents accidental reuse
after its resources have been released.

Does this PR introduce any user-facing change?

No.

How was this patch tested?

Added tests for closed-loader reuse, sequential manager lifecycles, and concurrent renewal shutdown.

JAVA_HOME=/usr/lib/jvm/java-17-openjdk-amd64 build/sbt \
  'core/testOnly org.apache.spark.security.CredentialProviderLoaderSuite' \
  'core/testOnly org.apache.spark.deploy.security.UserCredentialManagerSuite'

All 28 CredentialProviderLoaderSuite tests and all 22 UserCredentialManagerSuite tests passed.

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

Generated-by: OpenAI Codex (GPT-5)

[cloud-fan]

cc @sarutak

[uros-b]

Thank you @cloud-fan!

[sarutak]

Thanks for the followup, @cloud-fan. I agree that the closeAll() cache inconsistency is a real issue in that after one manager closes providers, the stale instances remain in cachedProviders and could be reinitialized by a later SparkContext.

However, I think the root cause is simpler than what this PR addresses: closeAll() clears initializedProviders but does not invalidate cachedProviders. I think the following minimal fix is sufficient.

public static void closeAll() throws Exception {
    synchronized (CredentialProviderLoader.class) {
        providersClosed = true;
        cachedProviders = null;  // force fresh ServiceLoader discovery on next use
        toClose = new ArrayList<>(initializedProviders);
        initializedProviders.clear();
    }
    // ... close logic
}

This forces the next providerFor() call to re-run ServiceLoader discovery, producing fresh instances. The providersClosed flag prevents accidental reuse between close and the next SparkContext.

[cloud-fan]

Thanks for taking a look. I agree that clearing cachedProviders addresses the immediate stale-instance issue. However, the proposed snippet sets providersClosed = true without a safe point to reset it for the next manager. Resetting global state during the next discovery would also leave lifecycle races where one manager can invalidate or close providers used by another.

Scoping the loader to UserCredentialManager makes provider ownership and shutdown state explicit: stopping one manager closes only its providers, while a later manager receives fresh instances. For that reason, I’d prefer to keep the instance-scoped approach.

[sarutak]

Thanks. On the two concerns:

1. Reset point for providersClosed: We can simplify by dropping the flag entirely. Setting cachedProviders = null in closeAll() is sufficient. The null cache naturally means "rediscover on next use." No separate closed/open state is needed.

2. Lifecycle races between managers: Spark enforces a single active SparkContext per JVM via SparkContext.assertNoOtherContextIsRunning. Two UserCredentialManager instances cannot coexist. The lifecycle is strictly sequential (manager A stops, then manager B starts). There is no window for one manager to invalidate providers used by another.

I intentionally chose a JVM-scoped loader because classpath-based ServiceLoader discovery is a JVM-level invariant (the discovered providers don't change between SparkContexts). Additionally, if we extend this to Spark Connect multi-session in the future, a shared server-level loader would be more appropriate. CredentialProvider.resolve() takes UserContext as an argument, so session isolation is achieved at the call site without per-session loaders. Instance-per-manager would mean redundant ServiceLoader discovery and separate HTTP connection pools per session.