You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by we...@apache.org on 2026/08/25 17:24:32 UTC
(spark) branch master updated: [SPARK-58472][CORE][FOLLOWUP] Scope credential providers to manager lifecycle
This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 7d617d840176 [SPARK-58472][CORE][FOLLOWUP] Scope credential providers to manager lifecycle
7d617d840176 is described below
commit 7d617d84017693e87326e43b75b2aad8b8967d40
Author: Wenchen Fan <we...@databricks.com>
AuthorDate: Wed Aug 26 01:24:07 2026 +0800
[SPARK-58472][CORE][FOLLOWUP] Scope credential providers to manager lifecycle
### What changes were proposed in this pull request?
Followup to https://github.com/apache/spark/pull/57677.
This PR scopes `CredentialProviderLoader` state to each `UserCredentialManager`. Each manager owns
one loader and its provider discovery, initialization, and closed state. `stop()` interrupts the
renewal executor, waits up to 10 seconds for an in-flight renewal to finish, and then closes the
manager's providers. Calls through a retired loader, including scheme discovery, are rejected.
It also adds coverage for sequential managers, closed-loader reuse, stale lifecycle callers, and
shutdown while credential renewal is in flight.
### Why are the changes needed?
The original implementation stores provider discovery and initialization state globally. Clearing
the static provider cache during shutdown would prevent a later manager from reusing released
instances, but it would not distinguish calls from different manager lifecycles.
If the bounded shutdown wait expires, a renewal task from the retiring manager can remain alive
after `stop()` returns. A later SparkContext must reopen shared static state for its new manager,
which would also let the stale task enter that state. With instance-scoped loaders, the stale task
retains the permanently closed loader from its own lifecycle, while the new manager receives a
fresh loader and can initialize fresh providers independently.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Added tests for closed-loader reuse and scheme discovery, sequential manager lifecycles, stale
lifecycle callers, and concurrent renewal shutdown.
```
build/sbt \
'core/testOnly org.apache.spark.security.CredentialProviderLoaderSuite' \
'core/testOnly org.apache.spark.deploy.security.UserCredentialManagerSuite'
```
All 29 `CredentialProviderLoaderSuite` tests and all 26 `UserCredentialManagerSuite` tests passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex (GPT-5)
Closes #57954 from cloud-fan/SPARK-58472-followup.
Authored-by: Wenchen Fan <we...@databricks.com>
Signed-off-by: Wenchen Fan <we...@databricks.com>
---
.../apache/spark/security/CredentialProvider.java | 10 +-
.../spark/security/CredentialProviderLoader.java | 76 ++++++-----
.../deploy/security/UserCredentialManager.scala | 41 ++++--
.../security/CredentialProviderLoaderSuite.java | 145 +++++++++++++++------
.../security/OidcCredentialIntegrationSuite.scala | 2 -
.../security/UserCredentialManagerSuite.scala | 104 ++++++++++++++-
6 files changed, 289 insertions(+), 89 deletions(-)
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 27015576cbfa..2d3e51f8ca10 100644
--- a/core/src/main/java/org/apache/spark/security/CredentialProvider.java
+++ b/core/src/main/java/org/apache/spark/security/CredentialProvider.java
@@ -138,11 +138,11 @@ public interface CredentialProvider extends AutoCloseable {
* is a no-op; providers that allocate long-lived resources in {@link #init(Map)} should
* override this method to clean them up.
* <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.
+ * Shutdown interrupts the renewal thread and waits a bounded time for in-flight calls to
+ * complete. If the wait times out or the shutdown thread is interrupted, {@code close()} may
+ * be invoked while another thread is still executing {@link #resolve(UserContext, URI)}.
+ * Implementations must tolerate a concurrent or subsequent {@code resolve()} failing after
+ * resources have been released, and {@code close()} itself must not block indefinitely.
* <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
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 50d2a24546e9..918d11f5b7b7 100644
--- a/core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java
+++ b/core/src/main/java/org/apache/spark/security/CredentialProviderLoader.java
@@ -40,8 +40,9 @@ import org.apache.spark.annotation.Private;
* Discovers {@link CredentialProvider} implementations via {@link ServiceLoader} and selects
* the appropriate provider for a given URI scheme using Binding Policy A (explicit selection).
* <p>
- * Provider discovery happens once (lazily on first call) and the list is cached. Each provider
- * is initialized exactly once per provider instance via {@link CredentialProvider#init(Map)}
+ * Provider discovery happens once per loader (lazily on first call) and the list is cached.
+ * Each provider is initialized exactly once per provider instance via
+ * {@link CredentialProvider#init(Map)}
* with the configuration from the first call that selects it (first-conf-wins semantics);
* subsequent resolutions reuse the already-initialized instance without re-calling {@code init}.
* <p>
@@ -53,7 +54,7 @@ import org.apache.spark.annotation.Private;
* is unset (or empty) do the count-based rules apply: a single candidate is auto-selected;
* multiple candidates produce an ambiguity error; no candidates produce {@code Optional.empty()}.
* <p>
- * <b>Thread-safety:</b> This class uses synchronized access to the cached provider list and
+ * <b>Thread-safety:</b> Each loader uses synchronized access to its cached provider list and
* initialization tracking, and callers may invoke {@link #providerFor(String, Map)} from
* multiple threads. A provider instance is cached and shared across callers; per the
* {@link CredentialProvider} contract, implementations must be thread-safe, so a returned
@@ -78,18 +79,19 @@ public final class CredentialProviderLoader {
*/
private static final String OIDC_CONF_PREFIX = "spark.security.oidc.";
- private static volatile List<CredentialProvider> cachedProviders;
+ private volatile List<CredentialProvider> cachedProviders;
/**
- * Tracks which provider instances have already been initialized. Guarded by the class lock.
+ * Tracks which provider instances have already been initialized. Guarded by this loader's lock.
* Uses identity semantics (reference equality) to handle multiple provider instances correctly.
*/
- private static final Set<CredentialProvider> initializedProviders =
+ private final Set<CredentialProvider> initializedProviders =
Collections.newSetFromMap(new IdentityHashMap<>());
- private CredentialProviderLoader() {
- // utility class
- }
+ /** Guarded by this loader's lock. Prevents providers from being reinitialized after shutdown. */
+ private boolean providersClosed;
+
+ public CredentialProviderLoader() {}
/**
* Returns the {@link CredentialProvider} for the given URI scheme, applying Binding Policy A:
@@ -118,14 +120,20 @@ public final class CredentialProviderLoader {
* @return the selected provider, or empty if no provider supports the scheme
* @throws IllegalArgumentException if explicit selection names an unknown or non-supporting
* class, or if multiple candidates exist without explicit selection
- * @throws IllegalStateException if a provider returns null from {@code supportedSchemes()}
+ * @throws IllegalStateException if providers have already been closed or if a provider returns
+ * null from {@code supportedSchemes()}
*/
- public static Optional<CredentialProvider> providerFor(String scheme, Map<String, String> conf) {
+ public Optional<CredentialProvider> providerFor(String scheme, Map<String, String> conf) {
Objects.requireNonNull(scheme, "scheme must not be null");
Objects.requireNonNull(conf, "conf must not be null");
if (scheme.isEmpty()) {
throw new IllegalArgumentException("scheme must not be empty");
}
+ synchronized (this) {
+ if (providersClosed) {
+ throw new IllegalStateException("Credential providers have already been closed");
+ }
+ }
String normalizedScheme = scheme.toLowerCase(Locale.ROOT);
List<CredentialProvider> providers = getProviders();
@@ -179,7 +187,11 @@ public final class CredentialProviderLoader {
// precedent of DataSourceV2Utils.extractSessionConfigs() which scopes configuration
// to a specific prefix. We keep the full key (unlike extractSessionConfigs which
// strips the prefix) so providers can distinguish sub-keys unambiguously.
- synchronized (CredentialProviderLoader.class) {
+ synchronized (this) {
+ // Re-check under the initialization lock in case closeAll() ran during selection.
+ if (providersClosed) {
+ throw new IllegalStateException("Credential providers have already been closed");
+ }
if (!initializedProviders.contains(selected)) {
Map<String, String> filteredConf = new HashMap<>();
for (Map.Entry<String, String> entry : conf.entrySet()) {
@@ -197,10 +209,10 @@ public final class CredentialProviderLoader {
/**
* Returns the cached list of discovered providers, loading them on first access.
*/
- private static List<CredentialProvider> getProviders() {
+ private List<CredentialProvider> getProviders() {
List<CredentialProvider> providers = cachedProviders;
if (providers == null) {
- synchronized (CredentialProviderLoader.class) {
+ synchronized (this) {
providers = cachedProviders;
if (providers == null) {
providers = loadProviders();
@@ -211,7 +223,7 @@ public final class CredentialProviderLoader {
return providers;
}
- private static List<CredentialProvider> loadProviders() {
+ private List<CredentialProvider> loadProviders() {
ClassLoader cl = Thread.currentThread().getContextClassLoader();
if (cl == null) {
cl = CredentialProvider.class.getClassLoader();
@@ -236,8 +248,14 @@ public final class CredentialProviderLoader {
* (e.g., {@code spark.security.oidc.provider.<scheme>}) is provided.
*
* @return a set of all supported scheme names (lowercased), possibly empty
+ * @throws IllegalStateException if providers have already been closed
*/
- public static Set<String> discoverAllSchemes() {
+ public Set<String> discoverAllSchemes() {
+ synchronized (this) {
+ if (providersClosed) {
+ throw new IllegalStateException("Credential providers have already been closed");
+ }
+ }
List<CredentialProvider> providers = getProviders();
Set<String> schemes = new HashSet<>();
for (CredentialProvider provider : providers) {
@@ -256,25 +274,23 @@ public final class CredentialProviderLoader {
* <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.
+ * on each. The first exception is retained, later exceptions are suppressed onto it, and the
+ * first exception is rethrown 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).
+ * After shutdown begins, subsequent {@link #providerFor} calls fail rather than
+ * re-initializing a cached provider whose resources have already been released.
* <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 {
+ public void closeAll() throws Exception {
List<CredentialProvider> toClose;
- synchronized (CredentialProviderLoader.class) {
+ synchronized (this) {
// Copy and clear under the lock to prevent double-close if closeAll() is called
// again concurrently, and to avoid ConcurrentModificationException.
+ providersClosed = true;
toClose = new ArrayList<>(initializedProviders);
initializedProviders.clear();
}
@@ -301,10 +317,11 @@ public final class CredentialProviderLoader {
* Resets the cached provider list and initialization tracking. Intended for testing only.
*/
@VisibleForTesting
- public static void resetForTesting() {
- synchronized (CredentialProviderLoader.class) {
+ public void resetForTesting() {
+ synchronized (this) {
cachedProviders = null;
initializedProviders.clear();
+ providersClosed = false;
}
}
@@ -312,10 +329,11 @@ public final class CredentialProviderLoader {
* Overrides the cached provider list for testing. Intended for testing only.
*/
@VisibleForTesting
- static void setProvidersForTesting(List<CredentialProvider> providers) {
- synchronized (CredentialProviderLoader.class) {
+ void setProvidersForTesting(List<CredentialProvider> providers) {
+ synchronized (this) {
cachedProviders = providers;
initializedProviders.clear();
+ providersClosed = false;
}
}
}
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 a06bf5f1e33b..a5e395e547b9 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
@@ -61,7 +61,16 @@ import org.apache.spark.util.{ThreadUtils, Utils}
private[spark] class UserCredentialManager(
sparkConf: SparkConf,
tokenIngestor: TokenIngestor,
- onCredentialsUpdate: (Long, Array[Byte]) => Unit) extends Logging {
+ onCredentialsUpdate: (Long, Array[Byte]) => Unit,
+ credentialProviderLoader: CredentialProviderLoader)
+ extends Logging {
+
+ def this(
+ sparkConf: SparkConf,
+ tokenIngestor: TokenIngestor,
+ onCredentialsUpdate: (Long, Array[Byte]) => Unit) = {
+ this(sparkConf, tokenIngestor, onCredentialsUpdate, new CredentialProviderLoader())
+ }
private val safetyMargin = sparkConf.get(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN)
private val minInterval = sparkConf.get(SECURITY_OIDC_RENEWAL_MIN_INTERVAL)
@@ -166,21 +175,35 @@ private[spark] class UserCredentialManager(
}
def stop(): Unit = {
+ var interrupted = false
if (renewalExecutor != null) {
renewalExecutor.shutdownNow()
+ try {
+ if (!renewalExecutor.awaitTermination(
+ UserCredentialManager.RENEWAL_SHUTDOWN_TIMEOUT_SECONDS, TimeUnit.SECONDS)) {
+ logWarning(log"Timed out waiting for credential renewal to stop; " +
+ log"closing credential providers while renewal may still be running.")
+ }
+ } catch {
+ case e: InterruptedException =>
+ interrupted = true
+ logWarning(log"Interrupted while waiting for credential renewal to stop.", e)
+ }
}
// 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.
+ // This loader belongs to this manager, so closing it cannot affect a later SparkContext.
try {
- CredentialProviderLoader.closeAll()
+ credentialProviderLoader.closeAll()
} catch {
case e: InterruptedException =>
- Thread.currentThread().interrupt()
+ interrupted = true
logWarning(log"Interrupted while closing credential providers during shutdown.", e)
case NonFatal(e) =>
logWarning(log"Error closing credential providers during shutdown.", e)
+ } finally {
+ if (interrupted) {
+ Thread.currentThread().interrupt()
+ }
}
}
@@ -262,7 +285,7 @@ private[spark] class UserCredentialManager(
for (scheme <- schemes) {
try {
- val providerOpt = CredentialProviderLoader.providerFor(scheme, credentialConfMap)
+ val providerOpt = credentialProviderLoader.providerFor(scheme, credentialConfMap)
if (providerOpt.isPresent) {
val provider = providerOpt.get()
// Use a synthetic target URI with just the scheme for initial resolution.
@@ -339,7 +362,7 @@ private[spark] class UserCredentialManager(
// available on the classpath by probing CredentialProviderLoader.
// This covers both built-in providers (e.g., connector/credential-aws for "s3a")
// and third-party providers registered via ServiceLoader.
- CredentialProviderLoader.discoverAllSchemes().asScala.toSet
+ credentialProviderLoader.discoverAllSchemes().asScala.toSet
}
}
@@ -415,6 +438,8 @@ private[spark] class UserCredentialManager(
private[spark] object UserCredentialManager {
+ private val RENEWAL_SHUTDOWN_TIMEOUT_SECONDS = 10L
+
/**
* Synthetic authority used in target URIs for scheme-based provider resolution.
* Providers should not rely on this value; it signals that no specific endpoint
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 34261f2524c2..b325fba1ff5d 100644
--- a/core/src/test/java/org/apache/spark/security/CredentialProviderLoaderSuite.java
+++ b/core/src/test/java/org/apache/spark/security/CredentialProviderLoaderSuite.java
@@ -25,6 +25,8 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.atomic.AtomicReference;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
@@ -43,9 +45,12 @@ import static org.junit.jupiter.api.Assertions.assertTrue;
*/
public class CredentialProviderLoaderSuite {
+ private CredentialProviderLoader loader;
+
@BeforeEach
public void setUp() {
- CredentialProviderLoader.resetForTesting();
+ loader = new CredentialProviderLoader();
+ loader.resetForTesting();
}
@Test
@@ -53,7 +58,7 @@ public class CredentialProviderLoaderSuite {
// The "fake" scheme is supported only by FakeCredentialProvider (single candidate).
// If discovery works, providerFor should find it.
Map<String, String> conf = Map.of();
- Optional<CredentialProvider> result = CredentialProviderLoader.providerFor("fake", conf);
+ Optional<CredentialProvider> result = loader.providerFor("fake", conf);
assertTrue(result.isPresent(), "ServiceLoader should discover FakeCredentialProvider");
assertInstanceOf(FakeCredentialProvider.class, result.get());
}
@@ -62,7 +67,7 @@ public class CredentialProviderLoaderSuite {
public void testSingleCandidateSchemeResolvesWithNoConf() {
// "fake" is supported only by FakeCredentialProvider
Map<String, String> conf = Map.of();
- Optional<CredentialProvider> result = CredentialProviderLoader.providerFor("fake", conf);
+ Optional<CredentialProvider> result = loader.providerFor("fake", conf);
assertTrue(result.isPresent());
assertInstanceOf(FakeCredentialProvider.class, result.get());
}
@@ -72,7 +77,7 @@ public class CredentialProviderLoaderSuite {
// "shared" is supported by both FakeCredentialProvider and AnotherFakeCredentialProvider
Map<String, String> conf = Map.of();
IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
- () -> CredentialProviderLoader.providerFor("shared", conf));
+ () -> loader.providerFor("shared", conf));
assertTrue(e.getMessage().contains("Multiple credential providers"),
"Should mention multiple providers: " + e.getMessage());
assertTrue(e.getMessage().contains("shared"),
@@ -92,7 +97,7 @@ public class CredentialProviderLoaderSuite {
Map<String, String> conf = new HashMap<>();
conf.put("spark.security.oidc.provider.shared", "");
IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
- () -> CredentialProviderLoader.providerFor("shared", conf));
+ () -> loader.providerFor("shared", conf));
assertTrue(e.getMessage().contains("Multiple credential providers"),
"Empty conf value should behave as unset: " + e.getMessage());
}
@@ -102,7 +107,7 @@ public class CredentialProviderLoaderSuite {
Map<String, String> conf = Map.of(
"spark.security.oidc.provider.shared",
FakeCredentialProvider.class.getName());
- Optional<CredentialProvider> result = CredentialProviderLoader.providerFor("shared", conf);
+ Optional<CredentialProvider> result = loader.providerFor("shared", conf);
assertTrue(result.isPresent());
assertInstanceOf(FakeCredentialProvider.class, result.get());
}
@@ -112,7 +117,7 @@ public class CredentialProviderLoaderSuite {
Map<String, String> conf = Map.of(
"spark.security.oidc.provider.shared",
AnotherFakeCredentialProvider.class.getName());
- Optional<CredentialProvider> result = CredentialProviderLoader.providerFor("shared", conf);
+ Optional<CredentialProvider> result = loader.providerFor("shared", conf);
assertTrue(result.isPresent());
assertInstanceOf(AnotherFakeCredentialProvider.class, result.get());
}
@@ -123,7 +128,7 @@ public class CredentialProviderLoaderSuite {
"spark.security.oidc.provider.fake",
"com.example.NonExistentProvider");
IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
- () -> CredentialProviderLoader.providerFor("fake", conf));
+ () -> loader.providerFor("fake", conf));
assertTrue(e.getMessage().contains("com.example.NonExistentProvider"),
"Should mention the configured class: " + e.getMessage());
assertTrue(e.getMessage().contains("fake"),
@@ -139,7 +144,7 @@ public class CredentialProviderLoaderSuite {
"spark.security.oidc.provider.fake",
AnotherFakeCredentialProvider.class.getName());
IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
- () -> CredentialProviderLoader.providerFor("fake", conf));
+ () -> loader.providerFor("fake", conf));
assertTrue(e.getMessage().contains(AnotherFakeCredentialProvider.class.getName()),
"Should mention the configured class: " + e.getMessage());
assertTrue(e.getMessage().contains("fake"),
@@ -154,7 +159,7 @@ public class CredentialProviderLoaderSuite {
Map<String, String> conf = Map.of(
"spark.security.oidc.provider.fake",
FakeCredentialProvider.class.getName());
- Optional<CredentialProvider> result = CredentialProviderLoader.providerFor("fake", conf);
+ Optional<CredentialProvider> result = loader.providerFor("fake", conf);
assertTrue(result.isPresent());
assertInstanceOf(FakeCredentialProvider.class, result.get());
}
@@ -167,7 +172,7 @@ public class CredentialProviderLoaderSuite {
"spark.security.oidc.provider.fake",
"org.apache.spark.security.SomeOtherProvider");
IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
- () -> CredentialProviderLoader.providerFor("fake", conf));
+ () -> loader.providerFor("fake", conf));
assertTrue(e.getMessage().contains("fake"),
"Should mention the scheme: " + e.getMessage());
assertTrue(e.getMessage().contains("org.apache.spark.security.SomeOtherProvider"),
@@ -180,7 +185,7 @@ public class CredentialProviderLoaderSuite {
public void testUnknownSchemeReturnsEmpty() {
Map<String, String> conf = Map.of();
Optional<CredentialProvider> result =
- CredentialProviderLoader.providerFor("nonexistent", conf);
+ loader.providerFor("nonexistent", conf);
assertFalse(result.isPresent(), "Unknown scheme should return empty");
}
@@ -190,7 +195,7 @@ public class CredentialProviderLoaderSuite {
conf.put("spark.security.oidc.endpoint", "https://sts.example.com");
conf.put("spark.security.oidc.roleArn", "arn:aws:iam::123456:role/test");
- Optional<CredentialProvider> result = CredentialProviderLoader.providerFor("fake", conf);
+ Optional<CredentialProvider> result = loader.providerFor("fake", conf);
assertTrue(result.isPresent());
FakeCredentialProvider fake = (FakeCredentialProvider) result.get();
assertNotNull(fake.getInitConf(), "init() should have been called");
@@ -210,8 +215,8 @@ public class CredentialProviderLoaderSuite {
Map<String, String> conf2 = new HashMap<>();
conf2.put("spark.security.oidc.tag", "second-call");
- Optional<CredentialProvider> result1 = CredentialProviderLoader.providerFor("fake", conf1);
- Optional<CredentialProvider> result2 = CredentialProviderLoader.providerFor("fake", conf2);
+ Optional<CredentialProvider> result1 = loader.providerFor("fake", conf1);
+ Optional<CredentialProvider> result2 = loader.providerFor("fake", conf2);
assertTrue(result1.isPresent());
assertTrue(result2.isPresent());
@@ -242,11 +247,11 @@ public class CredentialProviderLoaderSuite {
return null;
}
};
- CredentialProviderLoader.setProvidersForTesting(
+ loader.setProvidersForTesting(
List.of(nullSchemesProvider));
IllegalStateException e = assertThrows(IllegalStateException.class,
- () -> CredentialProviderLoader.providerFor("anything", Map.of()));
+ () -> loader.providerFor("anything", Map.of()));
assertTrue(e.getMessage().contains("returned null from supportedSchemes()"),
"Should have a clear null-schemes message: " + e.getMessage());
}
@@ -254,7 +259,7 @@ public class CredentialProviderLoaderSuite {
@Test
public void testResolveReturnsExpectedServiceCredential() throws Exception {
Map<String, String> conf = Map.of();
- Optional<CredentialProvider> result = CredentialProviderLoader.providerFor("fake", conf);
+ Optional<CredentialProvider> result = loader.providerFor("fake", conf);
assertTrue(result.isPresent());
UserContext user = new UserContext(
@@ -270,7 +275,7 @@ public class CredentialProviderLoaderSuite {
@Test
public void testResolveSentinelThrowsCredentialResolutionException() {
Map<String, String> conf = Map.of();
- Optional<CredentialProvider> result = CredentialProviderLoader.providerFor("fake", conf);
+ Optional<CredentialProvider> result = loader.providerFor("fake", conf);
assertTrue(result.isPresent());
UserContext user = new UserContext(
@@ -287,7 +292,7 @@ public class CredentialProviderLoaderSuite {
public void testSchemeNormalizationIsCaseInsensitive() {
// "FAKE" should resolve the same as "fake"
Map<String, String> conf = Map.of();
- Optional<CredentialProvider> result = CredentialProviderLoader.providerFor("FAKE", conf);
+ Optional<CredentialProvider> result = loader.providerFor("FAKE", conf);
assertTrue(result.isPresent());
assertInstanceOf(FakeCredentialProvider.class, result.get());
}
@@ -298,7 +303,7 @@ public class CredentialProviderLoaderSuite {
Map<String, String> conf = Map.of(
"spark.security.oidc.provider.shared",
FakeCredentialProvider.class.getName());
- Optional<CredentialProvider> result = CredentialProviderLoader.providerFor("SHARED", conf);
+ Optional<CredentialProvider> result = loader.providerFor("SHARED", conf);
assertTrue(result.isPresent());
assertInstanceOf(FakeCredentialProvider.class, result.get());
}
@@ -306,7 +311,7 @@ public class CredentialProviderLoaderSuite {
@Test
public void testNullSchemeThrowsNPE() {
NullPointerException e = assertThrows(NullPointerException.class,
- () -> CredentialProviderLoader.providerFor(null, Map.of()));
+ () -> loader.providerFor(null, Map.of()));
assertTrue(e.getMessage().contains("scheme must not be null"),
"Should have a clear message: " + e.getMessage());
}
@@ -314,7 +319,7 @@ public class CredentialProviderLoaderSuite {
@Test
public void testNullConfThrowsNPE() {
NullPointerException e = assertThrows(NullPointerException.class,
- () -> CredentialProviderLoader.providerFor("fake", null));
+ () -> loader.providerFor("fake", null));
assertTrue(e.getMessage().contains("conf must not be null"),
"Should have a clear message: " + e.getMessage());
}
@@ -322,7 +327,7 @@ public class CredentialProviderLoaderSuite {
@Test
public void testSuggestedTtlDefaultValue() {
Map<String, String> conf = Map.of();
- Optional<CredentialProvider> result = CredentialProviderLoader.providerFor("fake", conf);
+ Optional<CredentialProvider> result = loader.providerFor("fake", conf);
assertTrue(result.isPresent());
assertEquals(Duration.ofMinutes(15), result.get().suggestedTtl());
}
@@ -338,7 +343,7 @@ public class CredentialProviderLoaderSuite {
conf.put("spark.authenticate.secret", "TOPSECRET");
conf.put("spark.ssl.keyPassword", "keypass");
- Optional<CredentialProvider> result = CredentialProviderLoader.providerFor("fake", conf);
+ Optional<CredentialProvider> result = loader.providerFor("fake", conf);
assertTrue(result.isPresent());
FakeCredentialProvider fake = (FakeCredentialProvider) result.get();
Map<String, String> initConf = fake.getInitConf();
@@ -362,7 +367,7 @@ public class CredentialProviderLoaderSuite {
@Test
public void testEmptySchemeThrowsIllegalArgument() {
IllegalArgumentException e = assertThrows(IllegalArgumentException.class,
- () -> CredentialProviderLoader.providerFor("", Map.of()));
+ () -> loader.providerFor("", Map.of()));
assertEquals("scheme must not be empty", e.getMessage());
}
@@ -370,13 +375,13 @@ public class CredentialProviderLoaderSuite {
public void testCloseAllClosesInitializedProviders() throws Exception {
// Initialize a provider by calling providerFor
Map<String, String> conf = Map.of();
- Optional<CredentialProvider> result = CredentialProviderLoader.providerFor("fake", conf);
+ Optional<CredentialProvider> result = loader.providerFor("fake", conf);
assertTrue(result.isPresent());
FakeCredentialProvider fake = (FakeCredentialProvider) result.get();
assertEquals(0, fake.getCloseCount(), "close() not yet called");
// Call closeAll
- CredentialProviderLoader.closeAll();
+ loader.closeAll();
assertEquals(1, fake.getCloseCount(), "close() should be called exactly once");
}
@@ -406,18 +411,18 @@ public class CredentialProviderLoaderSuite {
FakeCredentialProvider fakeProvider = new FakeCredentialProvider();
- CredentialProviderLoader.setProvidersForTesting(
+ loader.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);
+ loader.providerFor("throwing", conf);
+ loader.providerFor("fake", conf);
// closeAll should throw (from throwingProvider) but still close fakeProvider
Exception e = assertThrows(Exception.class,
- () -> CredentialProviderLoader.closeAll());
+ () -> loader.closeAll());
assertTrue(e.getMessage().contains("Simulated close failure"));
assertEquals(1, fakeProvider.getCloseCount(),
"Second provider should still be closed even when first throws");
@@ -425,9 +430,73 @@ public class CredentialProviderLoaderSuite {
@Test
public void testCloseAllWithNoInitializedProvidersIsNoOp() throws Exception {
- // No providers initialized — closeAll should not throw
- CredentialProviderLoader.closeAll();
- // If we reach here, no exception was thrown — success
+ // No providers initialized; closeAll should not throw
+ loader.closeAll();
+ // If we reach here, no exception was thrown: success
+ }
+
+ @Test
+ public void testProviderCannotBeReinitializedAfterCloseAll() throws Exception {
+ Map<String, String> conf = Map.of();
+ CredentialProvider first = loader.providerFor("fake", conf).orElseThrow();
+
+ loader.closeAll();
+
+ IllegalStateException e = assertThrows(IllegalStateException.class,
+ () -> loader.providerFor("fake", conf));
+ assertEquals("Credential providers have already been closed", e.getMessage());
+ assertThrows(IllegalStateException.class,
+ () -> loader.providerFor("nonexistent", conf));
+ IllegalStateException discoverError = assertThrows(
+ IllegalStateException.class, loader::discoverAllSchemes);
+ assertEquals("Credential providers have already been closed", discoverError.getMessage());
+
+ CredentialProviderLoader nextLoader = new CredentialProviderLoader();
+ CredentialProvider second = nextLoader.providerFor("fake", conf).orElseThrow();
+ assertInstanceOf(FakeCredentialProvider.class, second);
+ assertTrue(first != second, "A new loader should discover a fresh provider instance");
+ }
+
+ @Test
+ public void testStaleLifecycleCannotUseNextLifecycleProviders() throws Exception {
+ Map<String, String> conf = Map.of();
+ CredentialProvider retiredProvider = loader.providerFor("fake", conf).orElseThrow();
+ CountDownLatch releaseStaleCaller = new CountDownLatch(1);
+ AtomicReference<Throwable> staleFailure = new AtomicReference<>();
+ Thread staleCaller = new Thread(() -> {
+ try {
+ releaseStaleCaller.await();
+ loader.providerFor("fake", conf);
+ } catch (Throwable t) {
+ staleFailure.set(t);
+ }
+ });
+ staleCaller.start();
+
+ try {
+ loader.closeAll();
+
+ CredentialProviderLoader nextLoader = new CredentialProviderLoader();
+ try {
+ CredentialProvider nextProvider = nextLoader.providerFor("fake", conf).orElseThrow();
+ assertTrue(retiredProvider != nextProvider,
+ "The next lifecycle should discover a fresh provider instance");
+
+ releaseStaleCaller.countDown();
+ staleCaller.join(10000);
+
+ assertFalse(staleCaller.isAlive(), "The stale caller should have completed");
+ assertInstanceOf(IllegalStateException.class, staleFailure.get());
+ assertEquals("Credential providers have already been closed",
+ staleFailure.get().getMessage());
+ } finally {
+ nextLoader.closeAll();
+ }
+ } finally {
+ releaseStaleCaller.countDown();
+ staleCaller.interrupt();
+ staleCaller.join(10000);
+ }
}
@Test
@@ -464,17 +533,17 @@ public class CredentialProviderLoaderSuite {
}
};
- CredentialProviderLoader.setProvidersForTesting(List.of(failOnceThenSucceed));
+ loader.setProvidersForTesting(List.of(failOnceThenSucceed));
// First call: init() throws, providerFor should propagate
Map<String, String> conf = Map.of();
RuntimeException e = assertThrows(RuntimeException.class,
- () -> CredentialProviderLoader.providerFor("retryscheme", conf));
+ () -> loader.providerFor("retryscheme", conf));
assertTrue(e.getMessage().contains("Simulated transient init failure"));
// Second call: init() should be retried and succeed
Optional<CredentialProvider> result =
- CredentialProviderLoader.providerFor("retryscheme", conf);
+ loader.providerFor("retryscheme", conf);
assertTrue(result.isPresent(), "Second providerFor should succeed after init retry");
// Verify init was called exactly twice (proving the retry)
diff --git a/core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala b/core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala
index be660693b319..47d19e05a2a5 100644
--- a/core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala
+++ b/core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala
@@ -57,7 +57,6 @@ class OidcCredentialIntegrationSuite extends SparkFunSuite {
override def beforeEach(): Unit = {
super.beforeEach()
- CredentialProviderLoader.resetForTesting()
tokenFile = File.createTempFile("oidc-token-", ".jwt")
tokenFile.deleteOnExit()
writeTokenFile("fake.jwt.token.workload")
@@ -67,7 +66,6 @@ class OidcCredentialIntegrationSuite extends SparkFunSuite {
try {
if (tokenFile != null) tokenFile.delete()
} finally {
- CredentialProviderLoader.resetForTesting()
super.afterEach()
}
}
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 14e0f1cf5a11..af8e02c11e51 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
@@ -20,6 +20,7 @@ package org.apache.spark.deploy.security
import java.time.Instant
import java.util
import java.util.Optional
+import java.util.concurrent.{CountDownLatch, TimeUnit}
import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
import scala.concurrent.duration._
@@ -32,11 +33,6 @@ import org.apache.spark.security._
class UserCredentialManagerSuite extends SparkFunSuite {
- override def beforeEach(): Unit = {
- super.beforeEach()
- CredentialProviderLoader.resetForTesting()
- }
-
private def createSparkConf(): SparkConf = {
new SparkConf(loadDefaults = false)
.set(SECURITY_OIDC_ENABLED, true)
@@ -497,6 +493,7 @@ class UserCredentialManagerSuite extends SparkFunSuite {
val conf = createSparkConf()
val ctx = createUserContext()
val callbackRef = new AtomicReference[Array[Byte]]()
+ val loader = new CredentialProviderLoader()
conf.set("spark.security.oidc.provider.fake",
"org.apache.spark.security.FakeCredentialProvider")
@@ -504,12 +501,13 @@ class UserCredentialManagerSuite extends SparkFunSuite {
val manager = new UserCredentialManager(
conf,
createIngestor(ctx),
- (_, bytes) => callbackRef.set(bytes))
+ (_, bytes) => callbackRef.set(bytes),
+ loader)
manager.start()
// Get the FakeCredentialProvider instance to verify close was called
- val providerOpt = CredentialProviderLoader.providerFor("fake",
+ val providerOpt = loader.providerFor("fake",
new util.HashMap[String, String]())
assert(providerOpt.isPresent)
val fakeProvider = providerOpt.get().asInstanceOf[FakeCredentialProvider]
@@ -521,6 +519,98 @@ class UserCredentialManagerSuite extends SparkFunSuite {
"stop() should close initialized providers exactly once")
}
+ test("a later manager uses fresh credential providers after stop") {
+ val conf = createSparkConf()
+ val ctx = createUserContext()
+ conf.set("spark.security.oidc.provider.fake",
+ "org.apache.spark.security.FakeCredentialProvider")
+
+ val firstLoader = new CredentialProviderLoader()
+ val firstManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, _) => (),
+ firstLoader)
+ firstManager.start()
+ val firstProvider = firstLoader.providerFor("fake",
+ new util.HashMap[String, String]()).get().asInstanceOf[FakeCredentialProvider]
+ firstManager.stop()
+ assert(firstProvider.getCloseCount === 1)
+
+ val secondLoader = new CredentialProviderLoader()
+ val secondManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, _) => (),
+ secondLoader)
+ try {
+ secondManager.start()
+ val secondProvider = secondLoader.providerFor("fake",
+ new util.HashMap[String, String]()).get().asInstanceOf[FakeCredentialProvider]
+ assert(secondProvider ne firstProvider)
+ assert(secondProvider.getCloseCount === 0)
+ } finally {
+ secondManager.stop()
+ }
+ }
+
+ test("stop() waits for credential renewal before closing providers") {
+ val conf = createSparkConf()
+ val ctx = createUserContext(expiresInSeconds = 6)
+ val callbackCount = new AtomicInteger()
+ val renewalStarted = new CountDownLatch(1)
+ val renewalInterrupted = new CountDownLatch(1)
+ val releaseRenewal = new CountDownLatch(1)
+ val renewalCompleted = new CountDownLatch(1)
+ val loader = new CredentialProviderLoader()
+
+ conf.set("spark.security.oidc.provider.fake",
+ "org.apache.spark.security.FakeCredentialProvider")
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, _) => {
+ if (callbackCount.incrementAndGet() > 1) {
+ renewalStarted.countDown()
+ var released = false
+ while (!released) {
+ try {
+ releaseRenewal.await()
+ released = true
+ } catch {
+ case _: InterruptedException => renewalInterrupted.countDown()
+ }
+ }
+ renewalCompleted.countDown()
+ }
+ },
+ loader)
+
+ manager.start()
+ val provider = loader.providerFor("fake",
+ new util.HashMap[String, String]()).get().asInstanceOf[FakeCredentialProvider]
+ assert(renewalStarted.await(10, TimeUnit.SECONDS))
+
+ val stopThread = new Thread(() => manager.stop())
+ stopThread.start()
+
+ try {
+ assert(renewalInterrupted.await(10, TimeUnit.SECONDS))
+ assert(stopThread.isAlive, "stop() should wait for credential renewal to finish")
+ assert(provider.getCloseCount === 0,
+ "stop() should not close providers while credential renewal is still running")
+ } finally {
+ releaseRenewal.countDown()
+ stopThread.join(10000)
+ }
+
+ assert(renewalCompleted.await(10, TimeUnit.SECONDS))
+ assert(!stopThread.isAlive, "stop() should finish after credential renewal exits")
+ assert(provider.getCloseCount === 1,
+ "stop() should close providers after credential renewal exits")
+ }
+
// ========== additionalSparkProperties application ==========
test("start() applies additionalSparkProperties from active providers") {
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org