From 9b6826a6f7116cc7269b53f33c103f2fcee7274d Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Wed, 15 Jul 2026 09:24:49 -0700 Subject: [PATCH 1/9] [SPARK-38954][CORE] Cloud Credential Refresh and Distribution Without Kerberos --- .../HadoopDelegationTokenManager.scala | 93 ++++++++---- .../spark/internal/config/package.scala | 11 ++ .../scheduler/SupportsDelegationToken.scala | 13 +- .../CoarseGrainedSchedulerBackend.scala | 4 + .../local/LocalSchedulerBackend.scala | 5 + ...ark.security.HadoopDelegationTokenProvider | 3 + .../NonKerberosCredentialsSuite.scala | 141 ++++++++++++++++++ 7 files changed, 240 insertions(+), 30 deletions(-) create mode 100644 core/src/test/scala/org/apache/spark/deploy/security/NonKerberosCredentialsSuite.scala 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 58bbd7477bd07..ee6170c843d95 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 @@ -34,6 +34,7 @@ import org.apache.spark.deploy.SparkHadoopUtil import org.apache.spark.internal.Logging import org.apache.spark.internal.LogKeys import org.apache.spark.internal.config._ +import org.apache.spark.internal.config.Network.NETWORK_CRYPTO_ENABLED import org.apache.spark.rpc.RpcEndpointRef import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.UpdateDelegationTokens import org.apache.spark.security.HadoopDelegationTokenProvider @@ -76,17 +77,33 @@ private[spark] class HadoopDelegationTokenManager( require((principal == null) == (keytab == null), "Both principal and keytab must be defined, or neither.") + if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { + require(sparkConf.get(NETWORK_CRYPTO_ENABLED), + "spark.network.crypto.enabled must be true when " + + "spark.security.credentials.directProviders.enabled is true. " + + "Credential tokens must not be transmitted over unencrypted RPC channels.") + } + private val delegationTokenProviders = loadProviders() + if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) && delegationTokenProviders.isEmpty) { + logWarning("spark.security.credentials.directProviders.enabled is true but " + + "no HadoopDelegationTokenProvider implementations were discovered. " + + "Ensure provider JARs are on the classpath with META-INF/services registration.") + } logDebug("Using the following builtin delegation token providers: " + s"${delegationTokenProviders.keys.mkString(", ")}.") private var renewalExecutor: ScheduledExecutorService = _ /** @return Whether delegation token renewal is enabled. */ - def renewalEnabled: Boolean = sparkConf.get(KERBEROS_RENEWAL_CREDENTIALS) match { - case "keytab" => principal != null - case "ccache" => UserGroupInformation.getCurrentUser().hasKerberosCredentials() - case _ => false + def renewalEnabled: Boolean = { + val hasKerberos = sparkConf.get(KERBEROS_RENEWAL_CREDENTIALS) match { + case "keytab" => principal != null + case "ccache" => UserGroupInformation.getCurrentUser().hasKerberosCredentials() + case _ => false + } + + hasKerberos || sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) } /** @@ -141,8 +158,6 @@ private[spark] class HadoopDelegationTokenManager( val hasKerberosCreds = principal != null || Option(currentUser.getRealUser()).getOrElse(currentUser).hasKerberosCredentials() - // Delegation tokens can only be obtained if the real user has Kerberos credentials, so - // skip creation when those are not available. if (hasKerberosCreds) { val freshUGI = doLogin() freshUGI.doAs(new PrivilegedExceptionAction[Unit]() { @@ -154,6 +169,9 @@ private[spark] class HadoopDelegationTokenManager( if (!currentUser.equals(freshUGI)) { FileSystem.closeAllForUGI(freshUGI) } + } else if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { + val (newTokens, _) = obtainDelegationTokens() + creds.addAll(newTokens) } } @@ -166,7 +184,14 @@ private[spark] class HadoopDelegationTokenManager( val creds = new Credentials() val nextRenewal = delegationTokenProviders.values.flatMap { provider => if (provider.delegationTokensRequired(sparkConf, hadoopConf)) { - provider.obtainDelegationTokens(hadoopConf, sparkConf, creds) + 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) + None + } } else { logDebug(s"Service ${provider.serviceName} does not require a token." + s" Check your configuration to see if security is disabled or not.") @@ -199,8 +224,7 @@ private[spark] class HadoopDelegationTokenManager( */ private def updateTokensTask(): Array[Byte] = { try { - val freshUGI = doLogin() - val creds = obtainTokensAndScheduleRenewal(freshUGI) + val creds = obtainTokensAndScheduleRenewal() val tokens = SparkHadoopUtil.get.serialize(creds) logInfo("Updating delegation tokens.") @@ -208,7 +232,6 @@ private[spark] class HadoopDelegationTokenManager( tokens } catch { case _: InterruptedException => - // Ignore, may happen if shutting down. null case e: Exception => val delay = TimeUnit.SECONDS.toMillis(sparkConf.get(CREDENTIALS_RENEWAL_RETRY_WAIT)) @@ -226,24 +249,40 @@ private[spark] class HadoopDelegationTokenManager( * * @return Credentials containing the new tokens. */ - private def obtainTokensAndScheduleRenewal(ugi: UserGroupInformation): Credentials = { - ugi.doAs(new PrivilegedExceptionAction[Credentials]() { - override def run(): Credentials = { - val (creds, nextRenewal) = obtainDelegationTokens() - - // Calculate the time when new credentials should be created, based on the configured - // ratio. - val now = System.currentTimeMillis - val ratio = sparkConf.get(CREDENTIALS_RENEWAL_INTERVAL_RATIO) - val delay = (ratio * (nextRenewal - now)).toLong - logInfo(log"Calculated delay on renewal is ${MDC(LogKeys.DELAY, delay)}," + - log" based on next renewal ${MDC(LogKeys.NEXT_RENEWAL_TIME, nextRenewal)}" + - log" and the ratio ${MDC(LogKeys.CREDENTIALS_RENEWAL_INTERVAL_RATIO, ratio)}," + - log" and current time ${MDC(LogKeys.CURRENT_TIME, now)}") - scheduleRenewal(delay) - creds + private def obtainTokensAndScheduleRenewal(): Credentials = { + val hasKerberosConfig = sparkConf.get(KERBEROS_RENEWAL_CREDENTIALS) match { + case "keytab" => principal != null + case "ccache" => true + case _ => false + } + + val (creds, nextRenewal) = if (hasKerberosConfig) { + val freshUGI = doLogin() + val currentUser = UserGroupInformation.getCurrentUser() + val result = freshUGI.doAs(new PrivilegedExceptionAction[(Credentials, Long)]() { + override def run(): (Credentials, Long) = { + obtainDelegationTokens() + } + }) + if (!currentUser.equals(freshUGI)) { + FileSystem.closeAllForUGI(freshUGI) } - }) + result + } else if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { + obtainDelegationTokens() + } else { + (new Credentials(), Long.MaxValue) + } + + val now = System.currentTimeMillis + val ratio = sparkConf.get(CREDENTIALS_RENEWAL_INTERVAL_RATIO) + val delay = (ratio * (nextRenewal - now)).toLong + logInfo(log"Calculated delay on renewal is ${MDC(LogKeys.DELAY, delay)}," + + log" based on next renewal ${MDC(LogKeys.NEXT_RENEWAL_TIME, nextRenewal)}" + + log" and the ratio ${MDC(LogKeys.CREDENTIALS_RENEWAL_INTERVAL_RATIO, ratio)}," + + log" and current time ${MDC(LogKeys.CURRENT_TIME, now)}") + scheduleRenewal(delay) + creds } private def doLogin(): UserGroupInformation = { diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index 02e0fc9f00895..873a7c243c1b4 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -1730,6 +1730,17 @@ package object config { .checkValue(_ > 0, "The minimum renewal interval must be a positive time value.") .createWithDefaultString("30s") + private[spark] val CREDENTIALS_DIRECT_PROVIDERS_ENABLED = + ConfigBuilder("spark.security.credentials.directProviders.enabled") + .doc( + "When true, enables delegation token collection and renewal without Kerberos. " + + "Providers are called directly (without doLogin/doAs) and participate in the " + + "same renewal and distribution lifecycle as Kerberos delegation token providers. " + + "Providers that require Kerberos self-gate via delegationTokensRequired.") + .version("5.0.0") + .booleanConf + .createWithDefault(false) + private[spark] val SHUFFLE_SORT_INIT_BUFFER_SIZE = ConfigBuilder("spark.shuffle.sort.initialBufferSize") .internal() 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 cbc32f900abfb..d201d9f337735 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala @@ -21,11 +21,12 @@ import org.apache.hadoop.security.UserGroupInformation import org.apache.spark.deploy.SparkHadoopUtil import org.apache.spark.deploy.security.HadoopDelegationTokenManager +import org.apache.spark.internal.Logging /** * A mix-in trait for SchedulerBackend that supports delegation tokens. */ -private[spark] trait SupportsDelegationToken { +private[spark] trait SupportsDelegationToken extends Logging { // The token manager used to create security tokens. protected var delegationTokenManager: Option[HadoopDelegationTokenManager] = None @@ -33,7 +34,7 @@ private[spark] trait SupportsDelegationToken { /** * Create the delegation token manager to be used for the application. This method is called * once during the start of the scheduler backend (so after the object has already been - * fully constructed), only if security is enabled in the Hadoop configuration. + * fully constructed), when security is enabled or direct credential providers are configured. */ protected def createTokenManager(): Option[HadoopDelegationTokenManager] = None @@ -42,8 +43,14 @@ 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. + */ + protected def tokenManagerRequired(): Boolean = UserGroupInformation.isSecurityEnabled + protected def setupTokenManager(): Unit = { - if (UserGroupInformation.isSecurityEnabled) { + if (tokenManagerRequired()) { delegationTokenManager = createTokenManager() delegationTokenManager.foreach { dtm => val ugi = UserGroupInformation.getCurrentUser diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala index e4bdcd9fa7a37..ea45bb792052b 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala @@ -620,6 +620,10 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp setupTokenManager() } + override protected def tokenManagerRequired(): Boolean = { + super.tokenManagerRequired() || conf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) + } + protected def createDriverEndpoint(): DriverEndpoint = new DriverEndpoint() def stopExecutors(): Unit = { diff --git a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala index c3a1a60a0384f..3bb83fc72139c 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala @@ -27,6 +27,7 @@ import org.apache.spark.deploy.SparkHadoopUtil import org.apache.spark.deploy.security.HadoopDelegationTokenManager import org.apache.spark.executor.{Executor, ExecutorBackend} import org.apache.spark.internal.{config, Logging, LogKeys} +import org.apache.spark.internal.config._ import org.apache.spark.launcher.{LauncherBackend, SparkAppHandle} import org.apache.spark.resource.{CpuAmount, ResourceInformation, ResourceProfile} import org.apache.spark.rpc.{RpcCallContext, RpcEndpointRef, RpcEnv, ThreadSafeRpcEndpoint} @@ -129,6 +130,10 @@ private[spark] class LocalSchedulerBackend( Some(new HadoopDelegationTokenManager(conf, scheduler.sc.hadoopConfiguration, localEndpoint)) } + override protected def tokenManagerRequired(): Boolean = { + super.tokenManagerRequired() || conf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) + } + override protected def updateDelegationTokens(tokens: Array[Byte]): Unit = { SparkHadoopUtil.get.addDelegationTokens(tokens, conf) } 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 ed3908e95e4cb..830729beaa19a 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 @@ -16,3 +16,6 @@ # org.apache.spark.deploy.security.ExceptionThrowingDelegationTokenProvider +org.apache.spark.deploy.security.TestNonKerberosTokenProvider +org.apache.spark.deploy.security.TestDisabledProvider +org.apache.spark.deploy.security.TestFailingProvider 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 new file mode 100644 index 0000000000000..0d8d278e2db79 --- /dev/null +++ b/core/src/test/scala/org/apache/spark/deploy/security/NonKerberosCredentialsSuite.scala @@ -0,0 +1,141 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.deploy.security + +import org.apache.hadoop.conf.Configuration +import org.apache.hadoop.io.Text +import org.apache.hadoop.security.Credentials + +import org.apache.spark.{SparkConf, SparkFunSuite} +import org.apache.spark.internal.config._ +import org.apache.spark.internal.config.Network.NETWORK_CRYPTO_ENABLED +import org.apache.spark.security.HadoopDelegationTokenProvider + +private class TestNonKerberosTokenProvider extends HadoopDelegationTokenProvider { + override def serviceName: String = "test-direct" + + override def delegationTokensRequired( + sparkConf: SparkConf, hadoopConf: Configuration): Boolean = true + + override def obtainDelegationTokens( + hadoopConf: Configuration, + sparkConf: SparkConf, + creds: Credentials): Option[Long] = { + creds.addSecretKey(new Text("test.direct.credential"), "test-token".getBytes) + Some(System.currentTimeMillis() + 3600000L) + } +} + +private class TestDisabledProvider extends HadoopDelegationTokenProvider { + override def serviceName: String = "test-disabled" + + override def delegationTokensRequired( + sparkConf: SparkConf, hadoopConf: Configuration): Boolean = false + + override def obtainDelegationTokens( + hadoopConf: Configuration, + sparkConf: SparkConf, + creds: Credentials): Option[Long] = { + // scalastyle:off throwerror + throw new AssertionError("Should not be called when delegationTokensRequired is false") + // scalastyle:on throwerror + } +} + +private class TestFailingProvider extends HadoopDelegationTokenProvider { + override def serviceName: String = "test-failing" + + override def delegationTokensRequired( + sparkConf: SparkConf, hadoopConf: Configuration): Boolean = true + + override def obtainDelegationTokens( + hadoopConf: Configuration, + sparkConf: SparkConf, + creds: Credentials): Option[Long] = { + throw new RuntimeException("Simulated provider failure") + } +} + +class NonKerberosCredentialsSuite extends SparkFunSuite { + private val hadoopConf = new Configuration() + + private def baseConf: SparkConf = new SparkConf(false) + .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) + .set(NETWORK_CRYPTO_ENABLED, true) + + test("renewalEnabled returns true when config is enabled") { + val manager = new HadoopDelegationTokenManager(baseConf, hadoopConf, null) + assert(manager.renewalEnabled) + } + + test("renewalEnabled returns false when config is disabled and no Kerberos") { + val sparkConf = new SparkConf(false) + .set(NETWORK_CRYPTO_ENABLED, true) + val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) + assert(!manager.renewalEnabled) + } + + test("providers are called without Kerberos when config is enabled") { + val manager = new HadoopDelegationTokenManager(baseConf, hadoopConf, null) + + val creds = new Credentials() + manager.obtainDelegationTokens(creds) + + assert(creds.getSecretKey(new Text("test.direct.credential")) != null) + assert(new String(creds.getSecretKey(new Text("test.direct.credential"))) === "test-token") + } + + test("providers with delegationTokensRequired=false are not called") { + val manager = new HadoopDelegationTokenManager(baseConf, hadoopConf, null) + + val creds = new Credentials() + manager.obtainDelegationTokens(creds) + + assert(creds.getSecretKey(new Text("test.direct.credential")) != null) + } + + test("provider failure does not prevent other providers from running") { + val sparkConf = baseConf + .set("spark.security.credentials.test-failing.enabled", "true") + val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) + + val creds = new Credentials() + manager.obtainDelegationTokens(creds) + + assert(creds.getSecretKey(new Text("test.direct.credential")) != null) + assert(new String(creds.getSecretKey(new Text("test.direct.credential"))) === "test-token") + } + + test("individual provider can be disabled via per-service config") { + val sparkConf = baseConf + .set("spark.security.credentials.test-direct.enabled", "false") + val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) + + assert(!manager.isProviderLoaded("test-direct")) + } + + test("fails if network crypto is not enabled") { + val sparkConf = new SparkConf(false) + .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) + + val e = intercept[IllegalArgumentException] { + new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) + } + assert(e.getMessage.contains("spark.network.crypto.enabled must be true")) + } +} From e99078d0f9e70383bd3215fd065bcf75fa080ad8 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Wed, 15 Jul 2026 12:52:21 -0700 Subject: [PATCH 2/9] Add comment --- .../spark/deploy/security/HadoopDelegationTokenManager.scala | 1 + 1 file changed, 1 insertion(+) 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 ee6170c843d95..b6cbda363616f 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 @@ -274,6 +274,7 @@ private[spark] class HadoopDelegationTokenManager( (new Credentials(), Long.MaxValue) } + // Scheduling does not require a privileged context (only token acquisition does). val now = System.currentTimeMillis val ratio = sparkConf.get(CREDENTIALS_RENEWAL_INTERVAL_RATIO) val delay = (ratio * (nextRenewal - now)).toLong From d8eff040290a47758761ce139ba98e357f654af5 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Sat, 18 Jul 2026 15:21:55 -0700 Subject: [PATCH 3/9] [SPARK-38954][CORE] Add ConfigBindingPolicy.NOT_APPLICABLE to new config --- .../main/scala/org/apache/spark/internal/config/package.scala | 1 + 1 file changed, 1 insertion(+) diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index 873a7c243c1b4..868bab1fe7d00 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -1738,6 +1738,7 @@ package object config { "same renewal and distribution lifecycle as Kerberos delegation token providers. " + "Providers that require Kerberos self-gate via delegationTokensRequired.") .version("5.0.0") + .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) .booleanConf .createWithDefault(false) From db3deefbcced5c543d1a5f0a6a56f15ca2518e53 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Wed, 22 Jul 2026 16:37:14 -0700 Subject: [PATCH 4/9] address review comments --- .../HadoopDelegationTokenManager.scala | 52 +++++++++++-------- .../NonKerberosCredentialsSuite.scala | 45 +++++++++++++++- 2 files changed, 72 insertions(+), 25 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 b6cbda363616f..bdc59f1080017 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 @@ -78,10 +78,14 @@ private[spark] class HadoopDelegationTokenManager( "Both principal and keytab must be defined, or neither.") if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { - require(sparkConf.get(NETWORK_CRYPTO_ENABLED), - "spark.network.crypto.enabled must be true when " + + val hasEncryption = sparkConf.get(NETWORK_CRYPTO_ENABLED) || + sparkConf.get(SASL_ENCRYPTION_ENABLED) || + sparkConf.getBoolean("spark.ssl.rpc.enabled", false) + require(hasEncryption, + "RPC channel encryption (spark.network.crypto.enabled, " + + "spark.authenticate.enableSaslEncryption, or spark.ssl.rpc.enabled) must be enabled when " + "spark.security.credentials.directProviders.enabled is true. " + - "Credential tokens must not be transmitted over unencrypted RPC channels.") + "Credential tokens must not be transmitted over unencrypted channels.") } private val delegationTokenProviders = loadProviders() @@ -95,15 +99,16 @@ private[spark] class HadoopDelegationTokenManager( private var renewalExecutor: ScheduledExecutorService = _ - /** @return Whether delegation token renewal is enabled. */ - def renewalEnabled: Boolean = { - val hasKerberos = sparkConf.get(KERBEROS_RENEWAL_CREDENTIALS) match { + private def hasKerberosCredentials: Boolean = + sparkConf.get(KERBEROS_RENEWAL_CREDENTIALS) match { case "keytab" => principal != null case "ccache" => UserGroupInformation.getCurrentUser().hasKerberosCredentials() case _ => false } - hasKerberos || sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) + /** @return Whether delegation token renewal is enabled. */ + def renewalEnabled: Boolean = { + hasKerberosCredentials || sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) } /** @@ -170,7 +175,7 @@ private[spark] class HadoopDelegationTokenManager( FileSystem.closeAllForUGI(freshUGI) } } else if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { - val (newTokens, _) = obtainDelegationTokens() + val (newTokens, _) = obtainDelegationTokens(isolateFailures = true) creds.addAll(newTokens) } } @@ -178,19 +183,26 @@ private[spark] class HadoopDelegationTokenManager( /** * Fetch new delegation tokens for configured services. * + * @param isolateFailures When true, catch per-provider exceptions and continue with remaining + * providers. When false, exceptions propagate to the caller. * @return 2-tuple (credentials with new tokens, time by which the tokens must be renewed) */ - private def obtainDelegationTokens(): (Credentials, Long) = { + private def obtainDelegationTokens( + isolateFailures: Boolean = false): (Credentials, Long) = { val creds = new Credentials() val nextRenewal = delegationTokenProviders.values.flatMap { provider => if (provider.delegationTokensRequired(sparkConf, hadoopConf)) { - try { + 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) + None + } + } else { provider.obtainDelegationTokens(hadoopConf, sparkConf, creds) - } catch { - case e: Exception => - logWarning(log"Failed to obtain credentials from " + - log"${MDC(LogKeys.SERVICE_NAME, provider.serviceName)}.", e) - None } } else { logDebug(s"Service ${provider.serviceName} does not require a token." + @@ -250,13 +262,7 @@ private[spark] class HadoopDelegationTokenManager( * @return Credentials containing the new tokens. */ private def obtainTokensAndScheduleRenewal(): Credentials = { - val hasKerberosConfig = sparkConf.get(KERBEROS_RENEWAL_CREDENTIALS) match { - case "keytab" => principal != null - case "ccache" => true - case _ => false - } - - val (creds, nextRenewal) = if (hasKerberosConfig) { + val (creds, nextRenewal) = if (hasKerberosCredentials) { val freshUGI = doLogin() val currentUser = UserGroupInformation.getCurrentUser() val result = freshUGI.doAs(new PrivilegedExceptionAction[(Credentials, Long)]() { @@ -269,7 +275,7 @@ private[spark] class HadoopDelegationTokenManager( } result } else if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { - obtainDelegationTokens() + obtainDelegationTokens(isolateFailures = true) } else { (new Credentials(), Long.MaxValue) } 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 0d8d278e2db79..a6d036b7e164e 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 @@ -20,10 +20,15 @@ package org.apache.spark.deploy.security import org.apache.hadoop.conf.Configuration import org.apache.hadoop.io.Text import org.apache.hadoop.security.Credentials +import org.mockito.ArgumentCaptor +import org.mockito.Mockito.{mock, verify} import org.apache.spark.{SparkConf, SparkFunSuite} +import org.apache.spark.deploy.SparkHadoopUtil import org.apache.spark.internal.config._ import org.apache.spark.internal.config.Network.NETWORK_CRYPTO_ENABLED +import org.apache.spark.rpc.RpcEndpointRef +import org.apache.spark.scheduler.cluster.CoarseGrainedClusterMessages.UpdateDelegationTokens import org.apache.spark.security.HadoopDelegationTokenProvider private class TestNonKerberosTokenProvider extends HadoopDelegationTokenProvider { @@ -129,13 +134,49 @@ class NonKerberosCredentialsSuite extends SparkFunSuite { assert(!manager.isProviderLoaded("test-direct")) } - test("fails if network crypto is not enabled") { + test("fails if no RPC encryption is enabled") { val sparkConf = new SparkConf(false) .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) val e = intercept[IllegalArgumentException] { new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) } - assert(e.getMessage.contains("spark.network.crypto.enabled must be true")) + assert(e.getMessage.contains("RPC channel encryption")) + } + + test("accepts SASL encryption as sufficient") { + val sparkConf = new SparkConf(false) + .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) + .set(SASL_ENCRYPTION_ENABLED, true) + val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) + assert(manager.renewalEnabled) + } + + test("accepts SSL RPC encryption as sufficient") { + val sparkConf = new SparkConf(false) + .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) + .set("spark.ssl.rpc.enabled", "true") + val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) + assert(manager.renewalEnabled) + } + + test("start() obtains tokens and sends UpdateDelegationTokens to schedulerRef") { + val mockRef = mock(classOf[RpcEndpointRef]) + val manager = new HadoopDelegationTokenManager(baseConf, hadoopConf, mockRef) + + try { + val tokens = manager.start() + assert(tokens != null) + + val captor = ArgumentCaptor.forClass(classOf[Any]) + verify(mockRef).send(captor.capture()) + val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens] + + val creds = SparkHadoopUtil.get.deserialize(msg.tokens) + assert(creds.getSecretKey(new Text("test.direct.credential")) != null) + assert(new String(creds.getSecretKey(new Text("test.direct.credential"))) === "test-token") + } finally { + manager.stop() + } } } From 1528b41b8fd7dfc29b8f2234dc9cc04dff7af5d0 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Thu, 23 Jul 2026 11:28:27 -0700 Subject: [PATCH 5/9] address more review comments --- .../HadoopDelegationTokenManager.scala | 48 +++++++++++-------- .../spark/internal/config/package.scala | 2 +- .../NonKerberosCredentialsSuite.scala | 6 ++- docs/security.md | 16 +++++++ 4 files changed, 50 insertions(+), 22 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 bdc59f1080017..84db868f627f1 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 @@ -112,14 +112,15 @@ private[spark] class HadoopDelegationTokenManager( } /** - * Start the token renewer. Requires a principal and keytab. Upon start, the renewer will - * obtain delegation tokens for all configured services and send them to the driver, and - * set up tasks to periodically get fresh tokens as needed. + * Start the token renewer. Upon start, the renewer will obtain delegation tokens for all + * configured services and send them to the driver, and set up tasks to periodically get + * fresh tokens as needed. * - * This method requires that a keytab has been provided to Spark, and will try to keep the - * logged in user's TGT valid while this manager is active. + * When Kerberos credentials are available, the manager keeps the TGT renewed and calls + * providers inside a privileged context. When only direct providers are configured, providers + * are called without Kerberos authentication. * - * @return New set of delegation tokens created for the configured principal. + * @return New set of delegation tokens created for the configured services. */ def start(): Array[Byte] = { require(renewalEnabled, "Token renewal must be enabled to start the renewer.") @@ -167,7 +168,7 @@ private[spark] class HadoopDelegationTokenManager( val freshUGI = doLogin() freshUGI.doAs(new PrivilegedExceptionAction[Unit]() { override def run(): Unit = { - val (newTokens, _) = obtainDelegationTokens() + val (newTokens, _, _) = obtainDelegationTokens() creds.addAll(newTokens) } }) @@ -175,7 +176,7 @@ private[spark] class HadoopDelegationTokenManager( FileSystem.closeAllForUGI(freshUGI) } } else if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { - val (newTokens, _) = obtainDelegationTokens(isolateFailures = true) + val (newTokens, _, _) = obtainDelegationTokens(isolateFailures = true) creds.addAll(newTokens) } } @@ -185,11 +186,14 @@ private[spark] class HadoopDelegationTokenManager( * * @param isolateFailures When true, catch per-provider exceptions and continue with remaining * providers. When false, exceptions propagate to the caller. - * @return 2-tuple (credentials with new tokens, time by which the tokens must be renewed) + * @return 3-tuple (credentials with new tokens, time by which the tokens must be renewed, + * number of providers that failed). The failure count is always 0 when + * isolateFailures is false (failures propagate instead). */ private def obtainDelegationTokens( - isolateFailures: Boolean = false): (Credentials, Long) = { + isolateFailures: Boolean = false): (Credentials, Long, Int) = { val creds = new Credentials() + var failureCount = 0 val nextRenewal = delegationTokenProviders.values.flatMap { provider => if (provider.delegationTokensRequired(sparkConf, hadoopConf)) { if (isolateFailures) { @@ -199,6 +203,7 @@ private[spark] class HadoopDelegationTokenManager( case e: Exception => logWarning(log"Failed to obtain credentials from " + log"${MDC(LogKeys.SERVICE_NAME, provider.serviceName)}.", e) + failureCount += 1 None } } else { @@ -210,7 +215,7 @@ private[spark] class HadoopDelegationTokenManager( None } }.foldLeft(Long.MaxValue)(math.min) - (creds, nextRenewal) + (creds, nextRenewal, failureCount) } // Visible for testing. @@ -262,14 +267,15 @@ private[spark] class HadoopDelegationTokenManager( * @return Credentials containing the new tokens. */ private def obtainTokensAndScheduleRenewal(): Credentials = { - val (creds, nextRenewal) = if (hasKerberosCredentials) { + val (creds, nextRenewal, failureCount) = if (hasKerberosCredentials) { val freshUGI = doLogin() val currentUser = UserGroupInformation.getCurrentUser() - val result = freshUGI.doAs(new PrivilegedExceptionAction[(Credentials, Long)]() { - override def run(): (Credentials, Long) = { - obtainDelegationTokens() - } - }) + val result = freshUGI.doAs( + new PrivilegedExceptionAction[(Credentials, Long, Int)]() { + override def run(): (Credentials, Long, Int) = { + obtainDelegationTokens() + } + }) if (!currentUser.equals(freshUGI)) { FileSystem.closeAllForUGI(freshUGI) } @@ -277,13 +283,17 @@ private[spark] class HadoopDelegationTokenManager( } else if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { obtainDelegationTokens(isolateFailures = true) } else { - (new Credentials(), Long.MaxValue) + (new Credentials(), Long.MaxValue, 0) } // Scheduling does not require a privileged context (only token acquisition does). val now = System.currentTimeMillis val ratio = sparkConf.get(CREDENTIALS_RENEWAL_INTERVAL_RATIO) - val delay = (ratio * (nextRenewal - now)).toLong + val delay = if (failureCount > 0 && nextRenewal == Long.MaxValue) { + TimeUnit.SECONDS.toMillis(sparkConf.get(CREDENTIALS_RENEWAL_RETRY_WAIT)) + } else { + (ratio * (nextRenewal - now)).toLong + } logInfo(log"Calculated delay on renewal is ${MDC(LogKeys.DELAY, delay)}," + log" based on next renewal ${MDC(LogKeys.NEXT_RENEWAL_TIME, nextRenewal)}" + log" and the ratio ${MDC(LogKeys.CREDENTIALS_RENEWAL_INTERVAL_RATIO, ratio)}," + diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index 868bab1fe7d00..e01599e7d63f1 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -1737,7 +1737,7 @@ package object config { "Providers are called directly (without doLogin/doAs) and participate in the " + "same renewal and distribution lifecycle as Kerberos delegation token providers. " + "Providers that require Kerberos self-gate via delegationTokensRequired.") - .version("5.0.0") + .version("4.3.0") .withBindingPolicy(ConfigBindingPolicy.NOT_APPLICABLE) .booleanConf .createWithDefault(false) 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 a6d036b7e164e..e3a1a0949241f 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 @@ -35,7 +35,8 @@ private class TestNonKerberosTokenProvider extends HadoopDelegationTokenProvider override def serviceName: String = "test-direct" override def delegationTokensRequired( - sparkConf: SparkConf, hadoopConf: Configuration): Boolean = true + sparkConf: SparkConf, hadoopConf: Configuration): Boolean = + sparkConf.getBoolean("spark.security.credentials.directProviders.enabled", false) override def obtainDelegationTokens( hadoopConf: Configuration, @@ -66,7 +67,8 @@ private class TestFailingProvider extends HadoopDelegationTokenProvider { override def serviceName: String = "test-failing" override def delegationTokensRequired( - sparkConf: SparkConf, hadoopConf: Configuration): Boolean = true + sparkConf: SparkConf, hadoopConf: Configuration): Boolean = + sparkConf.getBoolean("spark.security.credentials.directProviders.enabled", false) override def obtainDelegationTokens( hadoopConf: Configuration, diff --git a/docs/security.md b/docs/security.md index 7c0d1b08875e1..6f6324e04b7a5 100644 --- a/docs/security.md +++ b/docs/security.md @@ -981,6 +981,22 @@ The following options provides finer-grained control for this feature: 2.3.0 + + spark.security.credentials.directProviders.enabled + false + + When true, enables credential collection and renewal without Kerberos. Providers + registered via HadoopDelegationTokenProvider are called directly (without + doLogin/doAs) and participate in the same renewal and + distribution lifecycle as Kerberos delegation token providers. Providers that require + Kerberos self-gate via their delegationTokensRequired method. + Requires RPC channel encryption: at least one of + spark.network.crypto.enabled, + spark.authenticate.enableSaslEncryption, or + spark.ssl.rpc.enabled must be true. + + 4.3.0 + spark.kerberos.access.hadoopFileSystems (none) From dec600289de04bac1a4c26170275f90277969e36 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Fri, 24 Jul 2026 10:59:51 -0700 Subject: [PATCH 6/9] address additional review comments --- .../HadoopDelegationTokenManager.scala | 23 ++++++++++++------- .../scheduler/SupportsDelegationToken.scala | 3 +-- .../CoarseGrainedSchedulerBackend.scala | 5 ++++ .../local/LocalSchedulerBackend.scala | 3 +-- .../NonKerberosCredentialsSuite.scala | 2 ++ docs/security.md | 9 ++++---- 6 files changed, 29 insertions(+), 16 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 84db868f627f1..fb7a62e48389c 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 @@ -78,13 +78,14 @@ private[spark] class HadoopDelegationTokenManager( "Both principal and keytab must be defined, or neither.") if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { - val hasEncryption = sparkConf.get(NETWORK_CRYPTO_ENABLED) || - sparkConf.get(SASL_ENCRYPTION_ENABLED) || - sparkConf.getBoolean("spark.ssl.rpc.enabled", false) + val hasEncryption = sparkConf.getBoolean("spark.ssl.rpc.enabled", false) || + (sparkConf.get(NETWORK_AUTH_ENABLED) && + (sparkConf.get(NETWORK_CRYPTO_ENABLED) || sparkConf.get(SASL_ENCRYPTION_ENABLED))) require(hasEncryption, - "RPC channel encryption (spark.network.crypto.enabled, " + - "spark.authenticate.enableSaslEncryption, or spark.ssl.rpc.enabled) must be enabled when " + - "spark.security.credentials.directProviders.enabled is true. " + + "RPC channel encryption must be enabled when " + + "spark.security.credentials.directProviders.enabled is true: either " + + "spark.ssl.rpc.enabled=true, or spark.authenticate=true together with " + + "spark.network.crypto.enabled or spark.authenticate.enableSaslEncryption. " + "Credential tokens must not be transmitted over unencrypted channels.") } @@ -108,7 +109,8 @@ private[spark] class HadoopDelegationTokenManager( /** @return Whether delegation token renewal is enabled. */ def renewalEnabled: Boolean = { - hasKerberosCredentials || sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) + hasKerberosCredentials || + (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) && delegationTokenProviders.nonEmpty) } /** @@ -249,6 +251,7 @@ private[spark] class HadoopDelegationTokenManager( tokens } catch { case _: InterruptedException => + // Ignore, may happen if shutting down. null case e: Exception => val delay = TimeUnit.SECONDS.toMillis(sparkConf.get(CREDENTIALS_RENEWAL_RETRY_WAIT)) @@ -283,7 +286,11 @@ private[spark] class HadoopDelegationTokenManager( } else if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { obtainDelegationTokens(isolateFailures = true) } else { - (new Credentials(), Long.MaxValue, 0) + // Reachable when ccache-based Kerberos credentials expired after start(). + // Throw so that updateTokensTask() reschedules a retry instead of silently + // distributing empty credentials and never renewing again. + throw new IllegalStateException("Cannot obtain delegation tokens: no Kerberos " + + "credentials are available and direct credential providers are not enabled.") } // Scheduling does not require a privileged context (only token acquisition does). 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 d201d9f337735..3fdbe4c25017a 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/SupportsDelegationToken.scala @@ -21,12 +21,11 @@ import org.apache.hadoop.security.UserGroupInformation import org.apache.spark.deploy.SparkHadoopUtil import org.apache.spark.deploy.security.HadoopDelegationTokenManager -import org.apache.spark.internal.Logging /** * A mix-in trait for SchedulerBackend that supports delegation tokens. */ -private[spark] trait SupportsDelegationToken extends Logging { +private[spark] trait SupportsDelegationToken { // The token manager used to create security tokens. protected var delegationTokenManager: Option[HadoopDelegationTokenManager] = None diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala index ea45bb792052b..1665e5a7a6a47 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala @@ -618,6 +618,11 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp override def start(): Unit = { setupTokenManager() + if (conf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) && delegationTokenManager.isEmpty) { + logWarning("spark.security.credentials.directProviders.enabled is set but " + + "this cluster manager does not support credential distribution. " + + "No tokens will be collected or renewed.") + } } override protected def tokenManagerRequired(): Boolean = { diff --git a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala index 3bb83fc72139c..88f628f367288 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala @@ -27,7 +27,6 @@ import org.apache.spark.deploy.SparkHadoopUtil import org.apache.spark.deploy.security.HadoopDelegationTokenManager import org.apache.spark.executor.{Executor, ExecutorBackend} import org.apache.spark.internal.{config, Logging, LogKeys} -import org.apache.spark.internal.config._ import org.apache.spark.launcher.{LauncherBackend, SparkAppHandle} import org.apache.spark.resource.{CpuAmount, ResourceInformation, ResourceProfile} import org.apache.spark.rpc.{RpcCallContext, RpcEndpointRef, RpcEnv, ThreadSafeRpcEndpoint} @@ -131,7 +130,7 @@ private[spark] class LocalSchedulerBackend( } override protected def tokenManagerRequired(): Boolean = { - super.tokenManagerRequired() || conf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) + super.tokenManagerRequired() || conf.get(config.CREDENTIALS_DIRECT_PROVIDERS_ENABLED) } override protected def updateDelegationTokens(tokens: Array[Byte]): Unit = { 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 e3a1a0949241f..64914340d4eee 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 @@ -83,6 +83,7 @@ class NonKerberosCredentialsSuite extends SparkFunSuite { private def baseConf: SparkConf = new SparkConf(false) .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) + .set(NETWORK_AUTH_ENABLED, true) .set(NETWORK_CRYPTO_ENABLED, true) test("renewalEnabled returns true when config is enabled") { @@ -149,6 +150,7 @@ class NonKerberosCredentialsSuite extends SparkFunSuite { test("accepts SASL encryption as sufficient") { val sparkConf = new SparkConf(false) .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) + .set(NETWORK_AUTH_ENABLED, true) .set(SASL_ENCRYPTION_ENABLED, true) val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) assert(manager.renewalEnabled) diff --git a/docs/security.md b/docs/security.md index 6f6324e04b7a5..3915f09f4e8d7 100644 --- a/docs/security.md +++ b/docs/security.md @@ -990,10 +990,11 @@ The following options provides finer-grained control for this feature: doLogin/doAs) and participate in the same renewal and distribution lifecycle as Kerberos delegation token providers. Providers that require Kerberos self-gate via their delegationTokensRequired method. - Requires RPC channel encryption: at least one of - spark.network.crypto.enabled, - spark.authenticate.enableSaslEncryption, or - spark.ssl.rpc.enabled must be true. + Requires RPC channel encryption: either spark.ssl.rpc.enabled=true, + or spark.authenticate=true together with + spark.network.crypto.enabled or + spark.authenticate.enableSaslEncryption. + Supported on YARN, Kubernetes, and local mode. Not supported on standalone clusters. 4.3.0 From 2db76ba37b93a4427e9f28b9931f1affdf00f3b3 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Wed, 29 Jul 2026 14:19:09 -0700 Subject: [PATCH 7/9] address still more review comments --- .../HadoopDelegationTokenManager.scala | 36 +++++++++---------- .../spark/internal/config/package.scala | 4 +-- .../CoarseGrainedSchedulerBackend.scala | 6 ++-- .../local/LocalSchedulerBackend.scala | 2 +- .../NonKerberosCredentialsSuite.scala | 35 ++++++++++++++---- docs/security.md | 2 +- 6 files changed, 53 insertions(+), 32 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 fb7a62e48389c..1d2afe48d316e 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 @@ -77,21 +77,21 @@ private[spark] class HadoopDelegationTokenManager( require((principal == null) == (keytab == null), "Both principal and keytab must be defined, or neither.") - if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { + if (sparkConf.get(DIRECT_CREDENTIAL_PROVIDERS_ENABLED)) { val hasEncryption = sparkConf.getBoolean("spark.ssl.rpc.enabled", false) || (sparkConf.get(NETWORK_AUTH_ENABLED) && (sparkConf.get(NETWORK_CRYPTO_ENABLED) || sparkConf.get(SASL_ENCRYPTION_ENABLED))) require(hasEncryption, "RPC channel encryption must be enabled when " + - "spark.security.credentials.directProviders.enabled is true: either " + + "spark.security.directCredentialProviders.enabled is true: either " + "spark.ssl.rpc.enabled=true, or spark.authenticate=true together with " + "spark.network.crypto.enabled or spark.authenticate.enableSaslEncryption. " + "Credential tokens must not be transmitted over unencrypted channels.") } private val delegationTokenProviders = loadProviders() - if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) && delegationTokenProviders.isEmpty) { - logWarning("spark.security.credentials.directProviders.enabled is true but " + + if (sparkConf.get(DIRECT_CREDENTIAL_PROVIDERS_ENABLED) && delegationTokenProviders.isEmpty) { + logWarning("spark.security.directCredentialProviders.enabled is true but " + "no HadoopDelegationTokenProvider implementations were discovered. " + "Ensure provider JARs are on the classpath with META-INF/services registration.") } @@ -110,7 +110,7 @@ private[spark] class HadoopDelegationTokenManager( /** @return Whether delegation token renewal is enabled. */ def renewalEnabled: Boolean = { hasKerberosCredentials || - (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) && delegationTokenProviders.nonEmpty) + (sparkConf.get(DIRECT_CREDENTIAL_PROVIDERS_ENABLED) && delegationTokenProviders.nonEmpty) } /** @@ -177,7 +177,7 @@ private[spark] class HadoopDelegationTokenManager( if (!currentUser.equals(freshUGI)) { FileSystem.closeAllForUGI(freshUGI) } - } else if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { + } else if (sparkConf.get(DIRECT_CREDENTIAL_PROVIDERS_ENABLED)) { val (newTokens, _, _) = obtainDelegationTokens(isolateFailures = true) creds.addAll(newTokens) } @@ -272,18 +272,13 @@ private[spark] class HadoopDelegationTokenManager( private def obtainTokensAndScheduleRenewal(): Credentials = { val (creds, nextRenewal, failureCount) = if (hasKerberosCredentials) { val freshUGI = doLogin() - val currentUser = UserGroupInformation.getCurrentUser() - val result = freshUGI.doAs( + freshUGI.doAs( new PrivilegedExceptionAction[(Credentials, Long, Int)]() { override def run(): (Credentials, Long, Int) = { obtainDelegationTokens() } }) - if (!currentUser.equals(freshUGI)) { - FileSystem.closeAllForUGI(freshUGI) - } - result - } else if (sparkConf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED)) { + } else if (sparkConf.get(DIRECT_CREDENTIAL_PROVIDERS_ENABLED)) { obtainDelegationTokens(isolateFailures = true) } else { // Reachable when ccache-based Kerberos credentials expired after start(). @@ -293,14 +288,19 @@ private[spark] class HadoopDelegationTokenManager( "credentials are available and direct credential providers are not enabled.") } + // If every direct provider failed and no tokens were obtained, throw so that + // updateTokensTask() skips distributing empty credentials and schedules a retry. + // This mirrors the Kerberos path, where a provider failure propagates instead of + // sending empty tokens to executors. + if (failureCount > 0 && nextRenewal == Long.MaxValue) { + throw new IllegalStateException( + "All direct credential providers failed to obtain delegation tokens.") + } + // Scheduling does not require a privileged context (only token acquisition does). val now = System.currentTimeMillis val ratio = sparkConf.get(CREDENTIALS_RENEWAL_INTERVAL_RATIO) - val delay = if (failureCount > 0 && nextRenewal == Long.MaxValue) { - TimeUnit.SECONDS.toMillis(sparkConf.get(CREDENTIALS_RENEWAL_RETRY_WAIT)) - } else { - (ratio * (nextRenewal - now)).toLong - } + val delay = (ratio * (nextRenewal - now)).toLong logInfo(log"Calculated delay on renewal is ${MDC(LogKeys.DELAY, delay)}," + log" based on next renewal ${MDC(LogKeys.NEXT_RENEWAL_TIME, nextRenewal)}" + log" and the ratio ${MDC(LogKeys.CREDENTIALS_RENEWAL_INTERVAL_RATIO, ratio)}," + diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala b/core/src/main/scala/org/apache/spark/internal/config/package.scala index e01599e7d63f1..8038307afc65a 100644 --- a/core/src/main/scala/org/apache/spark/internal/config/package.scala +++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala @@ -1730,8 +1730,8 @@ package object config { .checkValue(_ > 0, "The minimum renewal interval must be a positive time value.") .createWithDefaultString("30s") - private[spark] val CREDENTIALS_DIRECT_PROVIDERS_ENABLED = - ConfigBuilder("spark.security.credentials.directProviders.enabled") + private[spark] val DIRECT_CREDENTIAL_PROVIDERS_ENABLED = + ConfigBuilder("spark.security.directCredentialProviders.enabled") .doc( "When true, enables delegation token collection and renewal without Kerberos. " + "Providers are called directly (without doLogin/doAs) and participate in the " + diff --git a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala index 1665e5a7a6a47..99fdc534a6fb9 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/cluster/CoarseGrainedSchedulerBackend.scala @@ -618,15 +618,15 @@ class CoarseGrainedSchedulerBackend(scheduler: TaskSchedulerImpl, val rpcEnv: Rp override def start(): Unit = { setupTokenManager() - if (conf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) && delegationTokenManager.isEmpty) { - logWarning("spark.security.credentials.directProviders.enabled is set but " + + if (conf.get(DIRECT_CREDENTIAL_PROVIDERS_ENABLED) && delegationTokenManager.isEmpty) { + logWarning("spark.security.directCredentialProviders.enabled is set but " + "this cluster manager does not support credential distribution. " + "No tokens will be collected or renewed.") } } override protected def tokenManagerRequired(): Boolean = { - super.tokenManagerRequired() || conf.get(CREDENTIALS_DIRECT_PROVIDERS_ENABLED) + super.tokenManagerRequired() || conf.get(DIRECT_CREDENTIAL_PROVIDERS_ENABLED) } protected def createDriverEndpoint(): DriverEndpoint = new DriverEndpoint() diff --git a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala index 88f628f367288..b6a90565aaff4 100644 --- a/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala +++ b/core/src/main/scala/org/apache/spark/scheduler/local/LocalSchedulerBackend.scala @@ -130,7 +130,7 @@ private[spark] class LocalSchedulerBackend( } override protected def tokenManagerRequired(): Boolean = { - super.tokenManagerRequired() || conf.get(config.CREDENTIALS_DIRECT_PROVIDERS_ENABLED) + super.tokenManagerRequired() || conf.get(config.DIRECT_CREDENTIAL_PROVIDERS_ENABLED) } override protected def updateDelegationTokens(tokens: Array[Byte]): Unit = { 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 64914340d4eee..1dc0fa8f94222 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 @@ -21,7 +21,8 @@ import org.apache.hadoop.conf.Configuration import org.apache.hadoop.io.Text import org.apache.hadoop.security.Credentials import org.mockito.ArgumentCaptor -import org.mockito.Mockito.{mock, verify} +import org.mockito.ArgumentMatchers.any +import org.mockito.Mockito.{mock, never, verify} import org.apache.spark.{SparkConf, SparkFunSuite} import org.apache.spark.deploy.SparkHadoopUtil @@ -36,7 +37,7 @@ private class TestNonKerberosTokenProvider extends HadoopDelegationTokenProvider override def delegationTokensRequired( sparkConf: SparkConf, hadoopConf: Configuration): Boolean = - sparkConf.getBoolean("spark.security.credentials.directProviders.enabled", false) + sparkConf.getBoolean("spark.security.directCredentialProviders.enabled", false) override def obtainDelegationTokens( hadoopConf: Configuration, @@ -68,7 +69,7 @@ private class TestFailingProvider extends HadoopDelegationTokenProvider { override def delegationTokensRequired( sparkConf: SparkConf, hadoopConf: Configuration): Boolean = - sparkConf.getBoolean("spark.security.credentials.directProviders.enabled", false) + sparkConf.getBoolean("spark.security.directCredentialProviders.enabled", false) override def obtainDelegationTokens( hadoopConf: Configuration, @@ -82,7 +83,7 @@ class NonKerberosCredentialsSuite extends SparkFunSuite { private val hadoopConf = new Configuration() private def baseConf: SparkConf = new SparkConf(false) - .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) + .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true) .set(NETWORK_AUTH_ENABLED, true) .set(NETWORK_CRYPTO_ENABLED, true) @@ -139,7 +140,7 @@ class NonKerberosCredentialsSuite extends SparkFunSuite { test("fails if no RPC encryption is enabled") { val sparkConf = new SparkConf(false) - .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) + .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true) val e = intercept[IllegalArgumentException] { new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) @@ -149,7 +150,7 @@ class NonKerberosCredentialsSuite extends SparkFunSuite { test("accepts SASL encryption as sufficient") { val sparkConf = new SparkConf(false) - .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) + .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true) .set(NETWORK_AUTH_ENABLED, true) .set(SASL_ENCRYPTION_ENABLED, true) val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) @@ -158,7 +159,7 @@ class NonKerberosCredentialsSuite extends SparkFunSuite { test("accepts SSL RPC encryption as sufficient") { val sparkConf = new SparkConf(false) - .set(CREDENTIALS_DIRECT_PROVIDERS_ENABLED, true) + .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true) .set("spark.ssl.rpc.enabled", "true") val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, null) assert(manager.renewalEnabled) @@ -183,4 +184,24 @@ class NonKerberosCredentialsSuite extends SparkFunSuite { manager.stop() } } + + test("start() does not send empty credentials when all direct providers fail") { + val mockRef = mock(classOf[RpcEndpointRef]) + // Disable the only succeeding provider so the failing provider is the sole active one, + // producing a total failure with no tokens obtained. + val sparkConf = baseConf + .set("spark.security.credentials.test-direct.enabled", "false") + val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, mockRef) + + try { + // On total failure, obtainTokensAndScheduleRenewal throws so updateTokensTask() skips + // distributing empty credentials and schedules a retry instead. start() returns null + // and no UpdateDelegationTokens message is ever sent to the executors. + val tokens = manager.start() + assert(tokens == null) + verify(mockRef, never()).send(any()) + } finally { + manager.stop() + } + } } diff --git a/docs/security.md b/docs/security.md index 3915f09f4e8d7..d802502560664 100644 --- a/docs/security.md +++ b/docs/security.md @@ -982,7 +982,7 @@ The following options provides finer-grained control for this feature: 2.3.0 - spark.security.credentials.directProviders.enabled + spark.security.directCredentialProviders.enabled false When true, enables credential collection and renewal without Kerberos. Providers From 1724972ff371b3bf4831c47910f989379f702387 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Wed, 29 Jul 2026 15:12:01 -0700 Subject: [PATCH 8/9] fix regression --- .../HadoopDelegationTokenManager.scala | 12 +++-- ...ark.security.HadoopDelegationTokenProvider | 1 + .../NonKerberosCredentialsSuite.scala | 51 ++++++++++++++++++- 3 files changed, 57 insertions(+), 7 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 1d2afe48d316e..1622d8501e03c 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 @@ -288,11 +288,13 @@ private[spark] class HadoopDelegationTokenManager( "credentials are available and direct credential providers are not enabled.") } - // If every direct provider failed and no tokens were obtained, throw so that - // updateTokensTask() skips distributing empty credentials and schedules a retry. - // This mirrors the Kerberos path, where a provider failure propagates instead of - // sending empty tokens to executors. - if (failureCount > 0 && nextRenewal == Long.MaxValue) { + // If at least one provider failed and no credentials were obtained at all, throw so that + // updateTokensTask() skips distributing empty credentials and schedules a retry. This + // mirrors the Kerberos path, where a provider failure propagates instead of sending empty + // tokens to executors. The check keys off the actual credential content rather than + // nextRenewal, since a provider may legitimately add tokens and still report no expiry + // (nextRenewal == Long.MaxValue), which is not a failure. + if (failureCount > 0 && creds.numberOfTokens() == 0 && creds.numberOfSecretKeys() == 0) { throw new IllegalStateException( "All direct credential providers failed to obtain delegation tokens.") } 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 830729beaa19a..8621d66256e04 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,3 +19,4 @@ 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.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 1dc0fa8f94222..7f6b1d611fe03 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,25 @@ private class TestFailingProvider extends HadoopDelegationTokenProvider { } } +// 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. +private class TestNoExpiryProvider extends HadoopDelegationTokenProvider { + override def serviceName: String = "test-noexpiry" + + override def delegationTokensRequired( + sparkConf: SparkConf, hadoopConf: Configuration): Boolean = + sparkConf.getBoolean("spark.security.directCredentialProviders.enabled", false) + + override def obtainDelegationTokens( + hadoopConf: Configuration, + sparkConf: SparkConf, + creds: Credentials): Option[Long] = { + creds.addSecretKey(new Text("test.noexpiry.credential"), "noexpiry-token".getBytes) + None + } +} + class NonKerberosCredentialsSuite extends SparkFunSuite { private val hadoopConf = new Configuration() @@ -187,10 +206,11 @@ class NonKerberosCredentialsSuite extends SparkFunSuite { test("start() does not send empty credentials when all direct providers fail") { val mockRef = mock(classOf[RpcEndpointRef]) - // Disable the only succeeding provider so the failing provider is the sole active one, - // producing a total failure with no tokens obtained. + // Disable both succeeding providers so the failing provider is the sole active one, + // producing a total failure with no credentials obtained. val sparkConf = baseConf .set("spark.security.credentials.test-direct.enabled", "false") + .set("spark.security.credentials.test-noexpiry.enabled", "false") val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, mockRef) try { @@ -204,4 +224,31 @@ class NonKerberosCredentialsSuite extends SparkFunSuite { manager.stop() } } + + test("start() sends partial credentials when some providers succeed without expiry") { + val mockRef = mock(classOf[RpcEndpointRef]) + // Disable the provider that reports an expiry, leaving a provider that adds a credential + // but returns None (nextRenewal stays Long.MaxValue) alongside the failing one. Even + // though a provider failed and no expiry was reported, the obtained credential must still + // be distributed rather than discarded as a spurious total failure. + val sparkConf = baseConf + .set("spark.security.credentials.test-direct.enabled", "false") + val manager = new HadoopDelegationTokenManager(sparkConf, hadoopConf, mockRef) + + try { + val tokens = manager.start() + assert(tokens != null) + + val captor = ArgumentCaptor.forClass(classOf[Any]) + verify(mockRef).send(captor.capture()) + val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens] + + val creds = SparkHadoopUtil.get.deserialize(msg.tokens) + assert(creds.getSecretKey(new Text("test.noexpiry.credential")) != null) + assert(new String(creds.getSecretKey(new Text("test.noexpiry.credential"))) + === "noexpiry-token") + } finally { + manager.stop() + } + } } From c429717c6d5aff1cc17c8d452f58f53c0fac66c3 Mon Sep 17 00:00:00 2001 From: Parth Chandra Date: Wed, 29 Jul 2026 15:40:24 -0700 Subject: [PATCH 9/9] fixes --- .../deploy/security/HadoopDelegationTokenManager.scala | 3 ++- .../spark/deploy/security/NonKerberosCredentialsSuite.scala | 6 +++--- 2 files changed, 5 insertions(+), 4 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 1622d8501e03c..ca202bd9bbfb7 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 @@ -110,7 +110,8 @@ private[spark] class HadoopDelegationTokenManager( /** @return Whether delegation token renewal is enabled. */ def renewalEnabled: Boolean = { hasKerberosCredentials || - (sparkConf.get(DIRECT_CREDENTIAL_PROVIDERS_ENABLED) && delegationTokenProviders.nonEmpty) + (sparkConf.get(DIRECT_CREDENTIAL_PROVIDERS_ENABLED) && + delegationTokenProviders.values.exists(_.delegationTokensRequired(sparkConf, hadoopConf))) } /** 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 7f6b1d611fe03..d6ecc954cfe0b 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 @@ -37,7 +37,7 @@ private class TestNonKerberosTokenProvider extends HadoopDelegationTokenProvider override def delegationTokensRequired( sparkConf: SparkConf, hadoopConf: Configuration): Boolean = - sparkConf.getBoolean("spark.security.directCredentialProviders.enabled", false) + sparkConf.get(DIRECT_CREDENTIAL_PROVIDERS_ENABLED) override def obtainDelegationTokens( hadoopConf: Configuration, @@ -69,7 +69,7 @@ private class TestFailingProvider extends HadoopDelegationTokenProvider { override def delegationTokensRequired( sparkConf: SparkConf, hadoopConf: Configuration): Boolean = - sparkConf.getBoolean("spark.security.directCredentialProviders.enabled", false) + sparkConf.get(DIRECT_CREDENTIAL_PROVIDERS_ENABLED) override def obtainDelegationTokens( hadoopConf: Configuration, @@ -87,7 +87,7 @@ private class TestNoExpiryProvider extends HadoopDelegationTokenProvider { override def delegationTokensRequired( sparkConf: SparkConf, hadoopConf: Configuration): Boolean = - sparkConf.getBoolean("spark.security.directCredentialProviders.enabled", false) + sparkConf.get(DIRECT_CREDENTIAL_PROVIDERS_ENABLED) override def obtainDelegationTokens( hadoopConf: Configuration,