This is an automated email from the ASF dual-hosted git repository.
cloud-fan pushed a commit to branch branch-4.x
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.x by this push:
new c77095080627 [SPARK-38954][CORE][FOLLOWUP] Isolate credential provider
requirement failures
c77095080627 is described below
commit c770950806277c2bf893aee5d4536e05cf7772bd
Author: Wenchen Fan <[email protected]>
AuthorDate: Mon Aug 10 13:22:17 2026 +0800
[SPARK-38954][CORE][FOLLOWUP] Isolate credential provider requirement
failures
### What changes were proposed in this pull request?
Followup to https://github.com/apache/spark/pull/57285.
Catch exceptions from `delegationTokensRequired` so one faulty provider
does not prevent other
direct credential providers from renewing and distributing credentials. Add
regression coverage
for renewal eligibility and token acquisition.
If every active direct provider fails its eligibility check during initial
setup, renewal remains
disabled and no retry is scheduled; the failures are logged. Once renewal
has started, provider
failures during a renewal cycle contribute to the existing failure count
and retry behavior.
### Why are the changes needed?
A provider throwing from `delegationTokensRequired` currently aborts
processing for every
provider. Failures should be isolated so healthy providers can continue
supplying credentials.
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Ran `NonKerberosCredentialsSuite`.
Without the production fix, 11 of 14 tests failed because the simulated
provider exception
escaped. With the fix, all 14 tests passed.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Codex GPT-5
Closes #57828 from cloud-fan/SPARK-38954-followup.
Authored-by: Wenchen Fan <[email protected]>
Signed-off-by: Wenchen Fan <[email protected]>
(cherry picked from commit 2a62c37e7e0f1e6f3c1c264ef9e6c52ce73e4134)
Signed-off-by: Wenchen Fan <[email protected]>
---
.../security/HadoopDelegationTokenManager.scala | 40 +++++++++++---------
.../spark/scheduler/SupportsDelegationToken.scala | 5 ++-
...he.spark.security.HadoopDelegationTokenProvider | 1 +
.../security/NonKerberosCredentialsSuite.scala | 44 ++++++++++++++++++++++
4 files changed, 71 insertions(+), 19 deletions(-)
diff --git
a/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala
b/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala
index ca202bd9bbfb..b98a71b13cba 100644
---
a/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala
+++
b/core/src/main/scala/org/apache/spark/deploy/security/HadoopDelegationTokenManager.scala
@@ -24,6 +24,7 @@ import java.util.ServiceLoader
import java.util.concurrent.{ScheduledExecutorService, TimeUnit}
import scala.collection.mutable
+import scala.util.control.NonFatal
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.FileSystem
@@ -111,7 +112,16 @@ private[spark] class HadoopDelegationTokenManager(
def renewalEnabled: Boolean = {
hasKerberosCredentials ||
(sparkConf.get(DIRECT_CREDENTIAL_PROVIDERS_ENABLED) &&
-
delegationTokenProviders.values.exists(_.delegationTokensRequired(sparkConf,
hadoopConf)))
+ delegationTokenProviders.values.exists { provider =>
+ try {
+ provider.delegationTokensRequired(sparkConf, hadoopConf)
+ } catch {
+ case NonFatal(e) =>
+ logWarning(log"Failed to determine whether credentials are
required from " +
+ log"${MDC(LogKeys.SERVICE_NAME, provider.serviceName)}.", e)
+ false
+ }
+ })
}
/**
@@ -198,24 +208,20 @@ private[spark] class HadoopDelegationTokenManager(
val creds = new Credentials()
var failureCount = 0
val nextRenewal = delegationTokenProviders.values.flatMap { provider =>
- if (provider.delegationTokensRequired(sparkConf, hadoopConf)) {
- if (isolateFailures) {
- try {
- provider.obtainDelegationTokens(hadoopConf, sparkConf, creds)
- } catch {
- case e: Exception =>
- logWarning(log"Failed to obtain credentials from " +
- log"${MDC(LogKeys.SERVICE_NAME, provider.serviceName)}.", e)
- failureCount += 1
- None
- }
- } else {
+ try {
+ if (provider.delegationTokensRequired(sparkConf, hadoopConf)) {
provider.obtainDelegationTokens(hadoopConf, sparkConf, creds)
+ } else {
+ logDebug(s"Service ${provider.serviceName} does not require a
token." +
+ s" Check your configuration to see if security is disabled or
not.")
+ None
}
- } else {
- logDebug(s"Service ${provider.serviceName} does not require a token." +
- s" Check your configuration to see if security is disabled or not.")
- None
+ } catch {
+ case NonFatal(e) if isolateFailures =>
+ logWarning(log"Failed to obtain credentials from " +
+ log"${MDC(LogKeys.SERVICE_NAME, provider.serviceName)}.", e)
+ failureCount += 1
+ None
}
}.foldLeft(Long.MaxValue)(math.min)
(creds, nextRenewal, failureCount)
diff --git
a/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala
b/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala
index 3fdbe4c25017..6edcc7806e01 100644
---
a/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala
+++
b/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala
@@ -43,8 +43,9 @@ private[spark] trait SupportsDelegationToken {
protected def updateDelegationTokens(tokens: Array[Byte]): Unit
/**
- * Whether the token manager should be started. Returns true if Hadoop
security is enabled
- * or if direct credential providers are configured.
+ * Whether the token manager should be started. The default implementation
returns true when
+ * Hadoop security is enabled. Backends that support direct credential
providers override this
+ * method to also check whether those providers are configured.
*/
protected def tokenManagerRequired(): Boolean =
UserGroupInformation.isSecurityEnabled
diff --git
a/core/src/test/resources/META-INF/services/org.apache.spark.security.HadoopDelegationTokenProvider
b/core/src/test/resources/META-INF/services/org.apache.spark.security.HadoopDelegationTokenProvider
index 8621d66256e0..add25b0c26fe 100644
---
a/core/src/test/resources/META-INF/services/org.apache.spark.security.HadoopDelegationTokenProvider
+++
b/core/src/test/resources/META-INF/services/org.apache.spark.security.HadoopDelegationTokenProvider
@@ -19,4 +19,5 @@
org.apache.spark.deploy.security.ExceptionThrowingDelegationTokenProvider
org.apache.spark.deploy.security.TestNonKerberosTokenProvider
org.apache.spark.deploy.security.TestDisabledProvider
org.apache.spark.deploy.security.TestFailingProvider
+org.apache.spark.deploy.security.TestRequirementFailingProvider
org.apache.spark.deploy.security.TestNoExpiryProvider
diff --git
a/core/src/test/scala/org/apache/spark/deploy/security/NonKerberosCredentialsSuite.scala
b/core/src/test/scala/org/apache/spark/deploy/security/NonKerberosCredentialsSuite.scala
index d6ecc954cfe0..f9f77b4d2798 100644
---
a/core/src/test/scala/org/apache/spark/deploy/security/NonKerberosCredentialsSuite.scala
+++
b/core/src/test/scala/org/apache/spark/deploy/security/NonKerberosCredentialsSuite.scala
@@ -79,6 +79,24 @@ private class TestFailingProvider extends
HadoopDelegationTokenProvider {
}
}
+private class TestRequirementFailingProvider extends
HadoopDelegationTokenProvider {
+ override def serviceName: String = "test-requirement-failing"
+
+ override def delegationTokensRequired(
+ sparkConf: SparkConf, hadoopConf: Configuration): Boolean = {
+ if (sparkConf.getBoolean(
+ "spark.test.requirement-failing.throw", false)) {
+ throw new RuntimeException("Simulated provider requirement failure")
+ }
+ false
+ }
+
+ override def obtainDelegationTokens(
+ hadoopConf: Configuration,
+ sparkConf: SparkConf,
+ creds: Credentials): Option[Long] = None
+}
+
// Adds a credential but reports no expiry (returns None), mimicking providers
such as
// HBaseDelegationTokenProvider. This exercises the nextRenewal ==
Long.MaxValue case where
// credentials were nevertheless obtained.
@@ -149,6 +167,32 @@ class NonKerberosCredentialsSuite extends SparkFunSuite {
assert(new String(creds.getSecretKey(new Text("test.direct.credential")))
=== "test-token")
}
+ test("provider requirement failure does not prevent other providers from
running") {
+ val sparkConf = baseConf
+ .set("spark.security.credentials.test-requirement-failing.enabled",
"true")
+ .set("spark.test.requirement-failing.throw", "true")
+ val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null)
+
+ assert(manager.renewalEnabled)
+
+ val creds = new Credentials()
+ manager.obtainDelegationTokens(creds)
+
+ assert(creds.getSecretKey(new Text("test.direct.credential")) != null)
+ }
+
+ test("provider requirement failure does not abort renewalEnabled") {
+ val sparkConf = baseConf
+ .set("spark.security.credentials.test-direct.enabled", "false")
+ .set("spark.security.credentials.test-failing.enabled", "false")
+ .set("spark.security.credentials.test-noexpiry.enabled", "false")
+ .set("spark.security.credentials.test-requirement-failing.enabled",
"true")
+ .set("spark.test.requirement-failing.throw", "true")
+ val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null)
+
+ assert(!manager.renewalEnabled)
+ }
+
test("individual provider can be disabled via per-service config") {
val sparkConf = baseConf
.set("spark.security.credentials.test-direct.enabled", "false")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
