You are viewing a plain text version of this content. The canonical link for it is here.
Posted to reviews@spark.apache.org by "shrirangmhalgi (via GitHub)" <gi...@apache.org> on 2026/08/08 06:28:37 UTC
[PR] [SPARK-57896][CORE][TESTS] Add Kerberos coexistence and per-user token integration tests [spark]
shrirangmhalgi opened a new pull request, #57866:
URL: https://github.com/apache/spark/pull/57866
### What changes were proposed in this pull request?
Add `OidcCredentialIntegrationSuite` with 13 integration tests that verify the coexistence of the new OIDC credential propagation framework (`UserCredentialManager`) with the existing Kerberos delegation token system (`HadoopDelegationTokenManager`).
The tests cover:
- OIDC credential delivery via the `UserCredentialManager` callback using the existing `FakeCredentialProvider`
- Credential refresh on token expiry with version monotonicity guarantees
- Per-user identity tokens produce credentials identically to workload-level tokens
- Both `UserCredentialManager` and `HadoopDelegationTokenManager` active simultaneously without interference
- Failure isolation: OIDC failure does not affect delegation tokens, and vice versa
- `UserCredentials` serialization roundtrip (executor receive path)
- Clean shutdown lifecycle
- Feature-disabled state does not interfere with the other credential system
### Why are the changes needed?
SPARK-57896 is a sub-task of the OIDC Credential Propagation SPIP (SPARK-57703). The SPIP's Q8 mid-term success criteria require integration tests proving that:
- A Fake `CredentialProvider` delivers `ServiceCredential` to executors and refreshes on expiry
- The same test passes with per-user identity tokens
- Both credential systems coexist without interference
- Existing Kerberos delegation token tests pass without modification
- These tests did not previously exist. All core SPI components are now merged on master, making these tests implementable.
### Does this PR introduce _any_ user-facing change?
No. This PR adds only test code.
### How was this patch tested?
All 13 new tests pass in `OidcCredentialIntegrationSuite`. The existing credential suites pass without modification:
- `UserCredentialManagerSuite`
- `NonKerberosCredentialsSuite`
- `HadoopDelegationTokenManagerSuite`
- `HadoopFSDelegationTokenProviderSuite`
### Was this patch authored or co-authored using generative AI tooling?
Yes. Co-Authored using Claude Opus 4.6
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57896][CORE][TESTS] Add Kerberos coexistence and per-user token integration tests [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak closed pull request #57866: [SPARK-57896][CORE][TESTS] Add Kerberos coexistence and per-user token integration tests
URL: https://github.com/apache/spark/pull/57866
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57896][CORE][TESTS] Add Kerberos coexistence and per-user token integration tests [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #57866:
URL: https://github.com/apache/spark/pull/57866#discussion_r3742413571
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,591 @@
+/*
+ * 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 java.io.{File, PrintWriter}
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ if (tokenFile != null) {
+ tokenFile.delete()
+ }
+ super.afterEach()
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ val pw = new PrintWriter(tokenFile)
+ try {
+ pw.print(content)
+ } finally {
+ pw.close()
+ }
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ // Verify callback was invoked with initial credentials
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ // Deserialize and verify the credential content
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ // FakeCredentialProvider resolves for scheme "fake"
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L) // 2s before expiry
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L) // 500ms min interval
+
+ // Create a short-lived context (expires in 3 seconds)
+ val ctx = createUserContext(expiresInSeconds = 3)
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, _) => {
+ latestVersion.set(version)
+ updateCount.incrementAndGet()
+ })
+
+ try {
+ manager.start()
+ assert(updateCount.get() == 1, "Should have exactly 1 update after start()")
+
+ // Wait for renewal to trigger (safety margin is 2s, credential expires in 3s,
+ // so renewal should happen around t=1s)
+ eventually(timeout(10.seconds)) {
+ assert(updateCount.get() >= 2,
+ s"Expected at least 2 updates (got ${updateCount.get()}), " +
+ "indicating credential renewal occurred")
+ }
+
+ // Verify version is monotonically increasing
+ assert(latestVersion.get() >= 2L,
+ "Version should be at least 2 after renewal")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: per-user identity token produces valid credentials") {
+ val conf = createOidcConf()
+
+ // Create a per-user context (different principal, same mechanism)
+ val userCtx = createUserContext(principal = "alice@corp.example.com")
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(userCtx),
+ (_, bytes) => callbackRef.set(bytes))
+
+ try {
+ val (_, initialBytes) = manager.start()
+
+ // Verify per-user token produces the same credential structure as workload token
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent,
+ "Per-user token should produce credentials for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Per-user credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Per-user credential should not be expired")
+
+ // Verify the credential is identical in structure to workload token output
+ // (FakeCredentialProvider doesn't distinguish -- same ServiceCredential for any UserContext)
+ val workloadCtx = createUserContext(principal = "workload-identity")
+ val workloadManager = new UserCredentialManager(
+ conf,
+ createIngestor(workloadCtx),
+ (_, _) => ())
+ try {
+ val (_, workloadBytes) = workloadManager.start()
+ val workloadCreds = UserCredentialManager.deserializeUserCredentials(workloadBytes)
+ val workloadFake = workloadCreds.forScheme("fake")
+ assert(workloadFake.isPresent)
+ // Both produce the same credential properties (provider=fake)
+ assert(workloadFake.get().getProperties == fakeCred.get().getProperties,
+ "Per-user and workload tokens should produce identical credential properties")
+ } finally {
+ workloadManager.stop()
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: UserCredentialManager and HadoopDelegationTokenManager coexist") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // Configure for BOTH OIDC and direct credential providers (non-Kerberos DT path)
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+ val oidcVersion = new AtomicReference[Long](0L)
+
+ // Start UserCredentialManager (OIDC)
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ oidcVersion.set(version)
+ oidcCallbackRef.set(bytes)
+ })
+
+ // Start HadoopDelegationTokenManager (Kerberos/direct)
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // Start OIDC manager
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L)
+ assert(oidcBytes != null)
+
+ // Start DT manager (uses TestNonKerberosTokenProvider from NonKerberosCredentialsSuite)
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should produce tokens")
+
+ // Verify OIDC credentials are valid
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should contain 'fake' scheme")
+
+ // Verify DT credentials were sent via RPC
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ eventually(timeout(5.seconds)) {
+ verify(mockRef).send(captor.capture())
Review Comment:
Ditto.
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,591 @@
+/*
+ * 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 java.io.{File, PrintWriter}
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
Review Comment:
Currently `CredentialProviderLoader.resetForTesting` is only called in `beforeEach`. If a test fails mid-execution, the next test's `beforeEach` reset handles it, but adding to `afterEach` is defensive and aligns with Spark test conventions:
```scala
override def afterEach(): Unit = {
try {
if (tokenFile != null) tokenFile.delete()
} finally {
CredentialProviderLoader.resetForTesting()
super.afterEach()
}
}
```
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,591 @@
+/*
+ * 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 java.io.{File, PrintWriter}
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ if (tokenFile != null) {
+ tokenFile.delete()
+ }
+ super.afterEach()
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ val pw = new PrintWriter(tokenFile)
+ try {
+ pw.print(content)
+ } finally {
+ pw.close()
+ }
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ // Verify callback was invoked with initial credentials
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ // Deserialize and verify the credential content
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ // FakeCredentialProvider resolves for scheme "fake"
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L) // 2s before expiry
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L) // 500ms min interval
+
+ // Create a short-lived context (expires in 3 seconds)
+ val ctx = createUserContext(expiresInSeconds = 3)
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, _) => {
+ latestVersion.set(version)
+ updateCount.incrementAndGet()
+ })
+
+ try {
+ manager.start()
+ assert(updateCount.get() == 1, "Should have exactly 1 update after start()")
+
+ // Wait for renewal to trigger (safety margin is 2s, credential expires in 3s,
+ // so renewal should happen around t=1s)
+ eventually(timeout(10.seconds)) {
+ assert(updateCount.get() >= 2,
+ s"Expected at least 2 updates (got ${updateCount.get()}), " +
+ "indicating credential renewal occurred")
+ }
+
+ // Verify version is monotonically increasing
+ assert(latestVersion.get() >= 2L,
+ "Version should be at least 2 after renewal")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: per-user identity token produces valid credentials") {
+ val conf = createOidcConf()
+
+ // Create a per-user context (different principal, same mechanism)
+ val userCtx = createUserContext(principal = "alice@corp.example.com")
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(userCtx),
+ (_, bytes) => callbackRef.set(bytes))
+
+ try {
+ val (_, initialBytes) = manager.start()
+
+ // Verify per-user token produces the same credential structure as workload token
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent,
+ "Per-user token should produce credentials for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Per-user credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Per-user credential should not be expired")
+
+ // Verify the credential is identical in structure to workload token output
+ // (FakeCredentialProvider doesn't distinguish -- same ServiceCredential for any UserContext)
+ val workloadCtx = createUserContext(principal = "workload-identity")
+ val workloadManager = new UserCredentialManager(
+ conf,
+ createIngestor(workloadCtx),
+ (_, _) => ())
+ try {
+ val (_, workloadBytes) = workloadManager.start()
+ val workloadCreds = UserCredentialManager.deserializeUserCredentials(workloadBytes)
+ val workloadFake = workloadCreds.forScheme("fake")
+ assert(workloadFake.isPresent)
+ // Both produce the same credential properties (provider=fake)
+ assert(workloadFake.get().getProperties == fakeCred.get().getProperties,
+ "Per-user and workload tokens should produce identical credential properties")
+ } finally {
+ workloadManager.stop()
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: UserCredentialManager and HadoopDelegationTokenManager coexist") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // Configure for BOTH OIDC and direct credential providers (non-Kerberos DT path)
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+ val oidcVersion = new AtomicReference[Long](0L)
Review Comment:
`AtomicReference[Long]` causes Scala `Long` -> Java `java.lang.Long` autoboxing. Use `. `AtomicLong` for correct atomic semantics without boxing overhead.
There are similar code in this PR.
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,591 @@
+/*
+ * 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 java.io.{File, PrintWriter}
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ if (tokenFile != null) {
+ tokenFile.delete()
+ }
+ super.afterEach()
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ val pw = new PrintWriter(tokenFile)
+ try {
+ pw.print(content)
+ } finally {
+ pw.close()
+ }
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ // Verify callback was invoked with initial credentials
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ // Deserialize and verify the credential content
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ // FakeCredentialProvider resolves for scheme "fake"
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L) // 2s before expiry
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L) // 500ms min interval
+
+ // Create a short-lived context (expires in 3 seconds)
+ val ctx = createUserContext(expiresInSeconds = 3)
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, _) => {
+ latestVersion.set(version)
+ updateCount.incrementAndGet()
+ })
+
+ try {
+ manager.start()
+ assert(updateCount.get() == 1, "Should have exactly 1 update after start()")
+
+ // Wait for renewal to trigger (safety margin is 2s, credential expires in 3s,
+ // so renewal should happen around t=1s)
+ eventually(timeout(10.seconds)) {
+ assert(updateCount.get() >= 2,
+ s"Expected at least 2 updates (got ${updateCount.get()}), " +
+ "indicating credential renewal occurred")
+ }
+
+ // Verify version is monotonically increasing
+ assert(latestVersion.get() >= 2L,
+ "Version should be at least 2 after renewal")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: per-user identity token produces valid credentials") {
+ val conf = createOidcConf()
+
+ // Create a per-user context (different principal, same mechanism)
+ val userCtx = createUserContext(principal = "alice@corp.example.com")
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(userCtx),
+ (_, bytes) => callbackRef.set(bytes))
+
+ try {
+ val (_, initialBytes) = manager.start()
+
+ // Verify per-user token produces the same credential structure as workload token
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent,
+ "Per-user token should produce credentials for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Per-user credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Per-user credential should not be expired")
+
+ // Verify the credential is identical in structure to workload token output
+ // (FakeCredentialProvider doesn't distinguish -- same ServiceCredential for any UserContext)
+ val workloadCtx = createUserContext(principal = "workload-identity")
+ val workloadManager = new UserCredentialManager(
+ conf,
+ createIngestor(workloadCtx),
+ (_, _) => ())
+ try {
+ val (_, workloadBytes) = workloadManager.start()
+ val workloadCreds = UserCredentialManager.deserializeUserCredentials(workloadBytes)
+ val workloadFake = workloadCreds.forScheme("fake")
+ assert(workloadFake.isPresent)
+ // Both produce the same credential properties (provider=fake)
+ assert(workloadFake.get().getProperties == fakeCred.get().getProperties,
+ "Per-user and workload tokens should produce identical credential properties")
+ } finally {
+ workloadManager.stop()
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: UserCredentialManager and HadoopDelegationTokenManager coexist") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // Configure for BOTH OIDC and direct credential providers (non-Kerberos DT path)
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+ val oidcVersion = new AtomicReference[Long](0L)
+
+ // Start UserCredentialManager (OIDC)
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ oidcVersion.set(version)
+ oidcCallbackRef.set(bytes)
+ })
+
+ // Start HadoopDelegationTokenManager (Kerberos/direct)
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // Start OIDC manager
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L)
+ assert(oidcBytes != null)
+
+ // Start DT manager (uses TestNonKerberosTokenProvider from NonKerberosCredentialsSuite)
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should produce tokens")
+
+ // Verify OIDC credentials are valid
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should contain 'fake' scheme")
+
+ // Verify DT credentials were sent via RPC
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ eventually(timeout(5.seconds)) {
+ verify(mockRef).send(captor.capture())
+ }
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should contain test.direct.credential")
+ assert(new String(dtCreds.getSecretKey(new Text("test.direct.credential"))) === "test-token",
+ "DT credential value should match")
+
+ // Both systems produced credentials independently
+ assert(oidcVersion.get() == 1L, "OIDC version should remain at 1")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC failure does not affect HadoopDelegationTokenManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ // OIDC manager with a FAILING ingestor (simulates missing/corrupt token file)
+ val failingOidcManager = new UserCredentialManager(
+ conf,
+ createFailingIngestor(),
+ (_, _) => ())
+
+ // DT manager should work independently
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // OIDC start should fail (empty token)
+ val oidcException = intercept[Exception] {
+ failingOidcManager.start()
+ }
+ assert(oidcException != null, "OIDC manager should fail with empty token")
+
+ // DT manager should still work perfectly despite OIDC failure
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should succeed despite OIDC failure")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should be unaffected by OIDC failure")
+ } finally {
+ failingOidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: DT provider failure does not affect UserCredentialManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // Enable direct providers but disable all succeeding providers
+ // so only the failing one runs (forces DT total failure)
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+ .set("spark.security.credentials.test-direct.enabled", "false")
+ .set("spark.security.credentials.test-noexpiry.enabled", "false")
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => oidcCallbackRef.set(bytes))
+
+ // DT manager with only failing provider active
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // DT manager start returns null when all providers fail
+ val dtTokens = dtManager.start()
+ assert(dtTokens == null, "DT manager should return null when all providers fail")
+
+ // OIDC manager should still work perfectly
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L, "OIDC version should be 1")
+ assert(oidcBytes != null, "OIDC should produce credentials despite DT failure")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should be unaffected by DT failure")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+
+ test("SPARK-57896: UserCredentials serialization roundtrip preserves all schemes") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ // Simulate what the executor does: deserialize the byte array
+ val bytes = serializedRef.get()
+ assert(bytes != null && bytes.length > 0, "Serialized credentials should be non-empty")
+
+ // First deserialization
+ val creds1 = UserCredentialManager.deserializeUserCredentials(bytes)
+ // Second deserialization of the same bytes (idempotency)
+ val creds2 = UserCredentialManager.deserializeUserCredentials(bytes)
+
+ // Both produce identical results
+ assert(creds1.forScheme("fake").isPresent)
+ assert(creds2.forScheme("fake").isPresent)
+ assert(creds1.forScheme("fake").get().getProperties ==
+ creds2.forScheme("fake").get().getProperties,
+ "Multiple deserializations of same bytes should produce identical credentials")
+
+ // Verify the credential is well-formed
+ val cred = creds1.forScheme("fake").get()
+ assert(cred.getProperties.containsKey("provider"))
+ assert(cred.getExpiresAt != null, "Credential should have an expiry set")
+ assert(!cred.isExpired(Instant.now()), "Freshly resolved credential should not be expired")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: multiple URI schemes resolved in single credential bundle") {
+ // FakeCredentialProvider supports both "fake" and "shared" schemes.
+ // When multiple target URIs are configured, all are resolved.
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ val creds = UserCredentialManager.deserializeUserCredentials(serializedRef.get())
+
+ // FakeCredentialProvider declares supportedSchemes = Set("fake", "shared")
+ // but "shared" is ambiguous (AnotherFakeCredentialProvider also claims it),
+ // so only "fake" auto-resolves without explicit config.
+ assert(creds.forScheme("fake").isPresent, "Should resolve 'fake' scheme")
+
+ // Verify case-insensitive lookup works
+ assert(creds.forScheme("FAKE").isPresent,
+ "Scheme lookup should be case-insensitive")
+ assert(creds.forScheme("Fake").isPresent,
+ "Scheme lookup should be case-insensitive")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: ServiceCredential.isExpired detects expiry correctly") {
+ // Unit-level verification that the executor can detect stale credentials
+ val now = Instant.now()
+
+ // Credential that expires in the future -- not expired
+ val validCred = new ServiceCredential(
+ java.util.Map.of("provider", "test"), now.plusSeconds(300))
+ assert(!validCred.isExpired(now), "Future-expiry credential should not be expired")
+
+ // Credential that already expired -- is expired
+ val expiredCred = new ServiceCredential(
+ java.util.Map.of("provider", "test"), now.minusSeconds(10))
+ assert(expiredCred.isExpired(now), "Past-expiry credential should be expired")
+
+ // Credential with null expiry -- never expires
+ val noExpiryCred = new ServiceCredential(
+ java.util.Map.of("provider", "test"), null)
+ assert(!noExpiryCred.isExpired(now),
+ "Null-expiry credential should never be considered expired")
+ }
+
+ test("SPARK-57896: stop() after start() completes cleanly without exceptions") {
+ val conf = createOidcConf()
+ val ctx = createUserContext(expiresInSeconds = 60)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, _) => ())
+
+ // start + immediate stop should not throw or leave dangling threads
+ manager.start()
+ manager.stop()
+
+ // Double stop should also be safe
+ manager.stop()
+ }
+
+ test("SPARK-57896: credential version is monotonically increasing across renewals") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val ctx = createUserContext(expiresInSeconds = 3)
+ val versions = new java.util.concurrent.CopyOnWriteArrayList[Long]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
Review Comment:
Ditto.
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,591 @@
+/*
+ * 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 java.io.{File, PrintWriter}
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ if (tokenFile != null) {
+ tokenFile.delete()
+ }
+ super.afterEach()
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ val pw = new PrintWriter(tokenFile)
+ try {
+ pw.print(content)
+ } finally {
+ pw.close()
+ }
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ // Verify callback was invoked with initial credentials
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ // Deserialize and verify the credential content
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ // FakeCredentialProvider resolves for scheme "fake"
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L) // 2s before expiry
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L) // 500ms min interval
+
+ // Create a short-lived context (expires in 3 seconds)
+ val ctx = createUserContext(expiresInSeconds = 3)
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
Review Comment:
**Problem**: `createIngestor(ctx)` returns a fixed `UserContext` with `expiresAt = Instant.now().plusSeconds(3)` (computed once at test setup). When `renewCredentialsTask()` calls `tokenIngestor.load()` during renewal, it gets the same context. By renewal time, this expiry is already in the past, so `computeRenewalDelay()` returns `minInterval` (500ms), causing immediate re-scheduling. It's effectively a spin-loop.
The tests pass because the spin-loop generates many callbacks quickly, satisfying "at least N updates". But this doesn't verify that renewal fires at the correct time relative to credential expiry.
**Fix**: Have the ingestor return a fresh `UserContext` on each `load()` call so that each renewal gets a genuinely new expiry:
```scala
val ingestor = new TokenIngestor {
override def load(): Optional[UserContext] =
Optional.of(createUserContext(expiresInSeconds = 3))
}
```
Since `createUserContext` computes `Instant.now().plusSeconds(3)` at call time, each renewal receives a context with a fresh expiry. This verifies that renewal fires at the correct time relative to credential expiry rather than spinning on an already-expired token.
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,591 @@
+/*
+ * 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 java.io.{File, PrintWriter}
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ if (tokenFile != null) {
+ tokenFile.delete()
+ }
+ super.afterEach()
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ val pw = new PrintWriter(tokenFile)
+ try {
+ pw.print(content)
+ } finally {
+ pw.close()
+ }
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ // Verify callback was invoked with initial credentials
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ // Deserialize and verify the credential content
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ // FakeCredentialProvider resolves for scheme "fake"
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L) // 2s before expiry
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L) // 500ms min interval
+
+ // Create a short-lived context (expires in 3 seconds)
+ val ctx = createUserContext(expiresInSeconds = 3)
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, _) => {
+ latestVersion.set(version)
+ updateCount.incrementAndGet()
+ })
+
+ try {
+ manager.start()
+ assert(updateCount.get() == 1, "Should have exactly 1 update after start()")
+
+ // Wait for renewal to trigger (safety margin is 2s, credential expires in 3s,
+ // so renewal should happen around t=1s)
+ eventually(timeout(10.seconds)) {
+ assert(updateCount.get() >= 2,
+ s"Expected at least 2 updates (got ${updateCount.get()}), " +
+ "indicating credential renewal occurred")
+ }
+
+ // Verify version is monotonically increasing
+ assert(latestVersion.get() >= 2L,
+ "Version should be at least 2 after renewal")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: per-user identity token produces valid credentials") {
+ val conf = createOidcConf()
+
+ // Create a per-user context (different principal, same mechanism)
+ val userCtx = createUserContext(principal = "alice@corp.example.com")
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(userCtx),
+ (_, bytes) => callbackRef.set(bytes))
+
+ try {
+ val (_, initialBytes) = manager.start()
+
+ // Verify per-user token produces the same credential structure as workload token
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent,
+ "Per-user token should produce credentials for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Per-user credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Per-user credential should not be expired")
+
+ // Verify the credential is identical in structure to workload token output
+ // (FakeCredentialProvider doesn't distinguish -- same ServiceCredential for any UserContext)
+ val workloadCtx = createUserContext(principal = "workload-identity")
+ val workloadManager = new UserCredentialManager(
+ conf,
+ createIngestor(workloadCtx),
+ (_, _) => ())
+ try {
+ val (_, workloadBytes) = workloadManager.start()
+ val workloadCreds = UserCredentialManager.deserializeUserCredentials(workloadBytes)
+ val workloadFake = workloadCreds.forScheme("fake")
+ assert(workloadFake.isPresent)
+ // Both produce the same credential properties (provider=fake)
+ assert(workloadFake.get().getProperties == fakeCred.get().getProperties,
+ "Per-user and workload tokens should produce identical credential properties")
+ } finally {
+ workloadManager.stop()
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: UserCredentialManager and HadoopDelegationTokenManager coexist") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // Configure for BOTH OIDC and direct credential providers (non-Kerberos DT path)
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+ val oidcVersion = new AtomicReference[Long](0L)
+
+ // Start UserCredentialManager (OIDC)
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ oidcVersion.set(version)
+ oidcCallbackRef.set(bytes)
+ })
+
+ // Start HadoopDelegationTokenManager (Kerberos/direct)
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // Start OIDC manager
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L)
+ assert(oidcBytes != null)
+
+ // Start DT manager (uses TestNonKerberosTokenProvider from NonKerberosCredentialsSuite)
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should produce tokens")
+
+ // Verify OIDC credentials are valid
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should contain 'fake' scheme")
+
+ // Verify DT credentials were sent via RPC
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ eventually(timeout(5.seconds)) {
+ verify(mockRef).send(captor.capture())
+ }
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should contain test.direct.credential")
+ assert(new String(dtCreds.getSecretKey(new Text("test.direct.credential"))) === "test-token",
+ "DT credential value should match")
+
+ // Both systems produced credentials independently
+ assert(oidcVersion.get() == 1L, "OIDC version should remain at 1")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC failure does not affect HadoopDelegationTokenManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ // OIDC manager with a FAILING ingestor (simulates missing/corrupt token file)
+ val failingOidcManager = new UserCredentialManager(
+ conf,
+ createFailingIngestor(),
+ (_, _) => ())
+
+ // DT manager should work independently
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // OIDC start should fail (empty token)
+ val oidcException = intercept[Exception] {
+ failingOidcManager.start()
+ }
+ assert(oidcException != null, "OIDC manager should fail with empty token")
Review Comment:
**Problem**:
1. `intercept[Exception]` catches unintended exceptions (NPE, ClassCastException, etc.), causing false passes.
2. `assert(oidcException != null)` is always true after a successful `intercept`. It's a no-op assertion.
**Fix**: `UserCredentialManager.start()` throws `IllegalStateException` when the token is missing:
```scala
val oidcException = intercept[IllegalStateException] {
failingOidcManager.start()
}
assert(oidcException.getMessage.contains("identity token file is missing or malformed"))
```
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,591 @@
+/*
+ * 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 java.io.{File, PrintWriter}
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ if (tokenFile != null) {
+ tokenFile.delete()
+ }
+ super.afterEach()
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ val pw = new PrintWriter(tokenFile)
+ try {
+ pw.print(content)
Review Comment:
`Files.writeString` might be better.
It's simpler and avoids `PrintWriter`'s silent error swallowing behavior.
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,591 @@
+/*
+ * 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 java.io.{File, PrintWriter}
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ if (tokenFile != null) {
+ tokenFile.delete()
+ }
+ super.afterEach()
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ val pw = new PrintWriter(tokenFile)
+ try {
+ pw.print(content)
+ } finally {
+ pw.close()
+ }
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ // Verify callback was invoked with initial credentials
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ // Deserialize and verify the credential content
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ // FakeCredentialProvider resolves for scheme "fake"
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L) // 2s before expiry
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L) // 500ms min interval
+
+ // Create a short-lived context (expires in 3 seconds)
+ val ctx = createUserContext(expiresInSeconds = 3)
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, _) => {
+ latestVersion.set(version)
+ updateCount.incrementAndGet()
+ })
+
+ try {
+ manager.start()
+ assert(updateCount.get() == 1, "Should have exactly 1 update after start()")
+
+ // Wait for renewal to trigger (safety margin is 2s, credential expires in 3s,
+ // so renewal should happen around t=1s)
+ eventually(timeout(10.seconds)) {
+ assert(updateCount.get() >= 2,
+ s"Expected at least 2 updates (got ${updateCount.get()}), " +
+ "indicating credential renewal occurred")
+ }
+
+ // Verify version is monotonically increasing
+ assert(latestVersion.get() >= 2L,
+ "Version should be at least 2 after renewal")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: per-user identity token produces valid credentials") {
+ val conf = createOidcConf()
+
+ // Create a per-user context (different principal, same mechanism)
+ val userCtx = createUserContext(principal = "alice@corp.example.com")
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(userCtx),
+ (_, bytes) => callbackRef.set(bytes))
+
+ try {
+ val (_, initialBytes) = manager.start()
+
+ // Verify per-user token produces the same credential structure as workload token
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent,
+ "Per-user token should produce credentials for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Per-user credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Per-user credential should not be expired")
+
+ // Verify the credential is identical in structure to workload token output
+ // (FakeCredentialProvider doesn't distinguish -- same ServiceCredential for any UserContext)
+ val workloadCtx = createUserContext(principal = "workload-identity")
+ val workloadManager = new UserCredentialManager(
+ conf,
+ createIngestor(workloadCtx),
+ (_, _) => ())
+ try {
+ val (_, workloadBytes) = workloadManager.start()
+ val workloadCreds = UserCredentialManager.deserializeUserCredentials(workloadBytes)
+ val workloadFake = workloadCreds.forScheme("fake")
+ assert(workloadFake.isPresent)
+ // Both produce the same credential properties (provider=fake)
+ assert(workloadFake.get().getProperties == fakeCred.get().getProperties,
+ "Per-user and workload tokens should produce identical credential properties")
+ } finally {
+ workloadManager.stop()
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: UserCredentialManager and HadoopDelegationTokenManager coexist") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // Configure for BOTH OIDC and direct credential providers (non-Kerberos DT path)
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+ val oidcVersion = new AtomicReference[Long](0L)
+
+ // Start UserCredentialManager (OIDC)
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ oidcVersion.set(version)
+ oidcCallbackRef.set(bytes)
+ })
+
+ // Start HadoopDelegationTokenManager (Kerberos/direct)
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // Start OIDC manager
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L)
+ assert(oidcBytes != null)
+
+ // Start DT manager (uses TestNonKerberosTokenProvider from NonKerberosCredentialsSuite)
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should produce tokens")
+
+ // Verify OIDC credentials are valid
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should contain 'fake' scheme")
+
+ // Verify DT credentials were sent via RPC
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ eventually(timeout(5.seconds)) {
+ verify(mockRef).send(captor.capture())
Review Comment:
**Problem**: `HadoopDelegationTokenManager.start()` calls `updateTokensTask()` synchronously, which calls `schedulerRef.send()` before `start()` returns. Since `mockRef` is a Mockito mock (returns instantly), `send()` has already been invoked by the time `start()` completes. The `eventually` wrapper is unnecessary.
Moreover, `eventually` + `verify` is a Mockito anti-pattern: on retry, `verify` sees the same invocation counted again and may throw `TooManyActualInvocations`.
**Fix**: Remove `eventually` and call `verify` directly, consistent with existing tests:
```scala
val captor = ArgumentCaptor.forClass(classOf[Any])
verify(mockRef).send(captor.capture())
```
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,591 @@
+/*
+ * 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 java.io.{File, PrintWriter}
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ if (tokenFile != null) {
+ tokenFile.delete()
+ }
+ super.afterEach()
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ val pw = new PrintWriter(tokenFile)
+ try {
+ pw.print(content)
+ } finally {
+ pw.close()
+ }
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ // Verify callback was invoked with initial credentials
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ // Deserialize and verify the credential content
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ // FakeCredentialProvider resolves for scheme "fake"
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L) // 2s before expiry
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L) // 500ms min interval
+
+ // Create a short-lived context (expires in 3 seconds)
+ val ctx = createUserContext(expiresInSeconds = 3)
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, _) => {
+ latestVersion.set(version)
+ updateCount.incrementAndGet()
+ })
+
+ try {
+ manager.start()
+ assert(updateCount.get() == 1, "Should have exactly 1 update after start()")
+
+ // Wait for renewal to trigger (safety margin is 2s, credential expires in 3s,
+ // so renewal should happen around t=1s)
+ eventually(timeout(10.seconds)) {
+ assert(updateCount.get() >= 2,
+ s"Expected at least 2 updates (got ${updateCount.get()}), " +
+ "indicating credential renewal occurred")
+ }
+
+ // Verify version is monotonically increasing
+ assert(latestVersion.get() >= 2L,
+ "Version should be at least 2 after renewal")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: per-user identity token produces valid credentials") {
+ val conf = createOidcConf()
+
+ // Create a per-user context (different principal, same mechanism)
+ val userCtx = createUserContext(principal = "alice@corp.example.com")
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(userCtx),
+ (_, bytes) => callbackRef.set(bytes))
+
+ try {
+ val (_, initialBytes) = manager.start()
+
+ // Verify per-user token produces the same credential structure as workload token
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent,
+ "Per-user token should produce credentials for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Per-user credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Per-user credential should not be expired")
+
+ // Verify the credential is identical in structure to workload token output
+ // (FakeCredentialProvider doesn't distinguish -- same ServiceCredential for any UserContext)
+ val workloadCtx = createUserContext(principal = "workload-identity")
+ val workloadManager = new UserCredentialManager(
+ conf,
+ createIngestor(workloadCtx),
+ (_, _) => ())
+ try {
+ val (_, workloadBytes) = workloadManager.start()
+ val workloadCreds = UserCredentialManager.deserializeUserCredentials(workloadBytes)
+ val workloadFake = workloadCreds.forScheme("fake")
+ assert(workloadFake.isPresent)
+ // Both produce the same credential properties (provider=fake)
+ assert(workloadFake.get().getProperties == fakeCred.get().getProperties,
+ "Per-user and workload tokens should produce identical credential properties")
+ } finally {
+ workloadManager.stop()
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: UserCredentialManager and HadoopDelegationTokenManager coexist") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // Configure for BOTH OIDC and direct credential providers (non-Kerberos DT path)
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+ val oidcVersion = new AtomicReference[Long](0L)
+
+ // Start UserCredentialManager (OIDC)
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ oidcVersion.set(version)
+ oidcCallbackRef.set(bytes)
+ })
+
+ // Start HadoopDelegationTokenManager (Kerberos/direct)
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // Start OIDC manager
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L)
+ assert(oidcBytes != null)
+
+ // Start DT manager (uses TestNonKerberosTokenProvider from NonKerberosCredentialsSuite)
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should produce tokens")
+
+ // Verify OIDC credentials are valid
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should contain 'fake' scheme")
+
+ // Verify DT credentials were sent via RPC
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ eventually(timeout(5.seconds)) {
+ verify(mockRef).send(captor.capture())
+ }
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should contain test.direct.credential")
+ assert(new String(dtCreds.getSecretKey(new Text("test.direct.credential"))) === "test-token",
+ "DT credential value should match")
+
+ // Both systems produced credentials independently
+ assert(oidcVersion.get() == 1L, "OIDC version should remain at 1")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC failure does not affect HadoopDelegationTokenManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ // OIDC manager with a FAILING ingestor (simulates missing/corrupt token file)
+ val failingOidcManager = new UserCredentialManager(
+ conf,
+ createFailingIngestor(),
+ (_, _) => ())
+
+ // DT manager should work independently
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // OIDC start should fail (empty token)
+ val oidcException = intercept[Exception] {
+ failingOidcManager.start()
+ }
+ assert(oidcException != null, "OIDC manager should fail with empty token")
+
+ // DT manager should still work perfectly despite OIDC failure
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should succeed despite OIDC failure")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should be unaffected by OIDC failure")
+ } finally {
+ failingOidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: DT provider failure does not affect UserCredentialManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // Enable direct providers but disable all succeeding providers
+ // so only the failing one runs (forces DT total failure)
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+ .set("spark.security.credentials.test-direct.enabled", "false")
+ .set("spark.security.credentials.test-noexpiry.enabled", "false")
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => oidcCallbackRef.set(bytes))
+
+ // DT manager with only failing provider active
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // DT manager start returns null when all providers fail
+ val dtTokens = dtManager.start()
+ assert(dtTokens == null, "DT manager should return null when all providers fail")
+
+ // OIDC manager should still work perfectly
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L, "OIDC version should be 1")
+ assert(oidcBytes != null, "OIDC should produce credentials despite DT failure")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should be unaffected by DT failure")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+
+ test("SPARK-57896: UserCredentials serialization roundtrip preserves all schemes") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ // Simulate what the executor does: deserialize the byte array
+ val bytes = serializedRef.get()
+ assert(bytes != null && bytes.length > 0, "Serialized credentials should be non-empty")
+
+ // First deserialization
+ val creds1 = UserCredentialManager.deserializeUserCredentials(bytes)
+ // Second deserialization of the same bytes (idempotency)
+ val creds2 = UserCredentialManager.deserializeUserCredentials(bytes)
+
+ // Both produce identical results
+ assert(creds1.forScheme("fake").isPresent)
+ assert(creds2.forScheme("fake").isPresent)
+ assert(creds1.forScheme("fake").get().getProperties ==
+ creds2.forScheme("fake").get().getProperties,
+ "Multiple deserializations of same bytes should produce identical credentials")
+
+ // Verify the credential is well-formed
+ val cred = creds1.forScheme("fake").get()
+ assert(cred.getProperties.containsKey("provider"))
+ assert(cred.getExpiresAt != null, "Credential should have an expiry set")
+ assert(!cred.isExpired(Instant.now()), "Freshly resolved credential should not be expired")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: multiple URI schemes resolved in single credential bundle") {
Review Comment:
The name of this test is `"multiple URI schemes resolved in single credential bundle"` but actually tests case-insensitive scheme lookup, not multiple scheme resolution. The comment even acknowledges "shared" is ambiguous and won't auto-resolve.
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,591 @@
+/*
+ * 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 java.io.{File, PrintWriter}
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ if (tokenFile != null) {
+ tokenFile.delete()
+ }
+ super.afterEach()
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ val pw = new PrintWriter(tokenFile)
+ try {
+ pw.print(content)
+ } finally {
+ pw.close()
+ }
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
Review Comment:
The two calls can differ by milliseconds. Use a single `val now = Instant.now()`:
```scala
private def createUserContext(...): UserContext = {
val now = Instant.now()
new UserContext(principal, "https://issuer.example.com", "fake.jwt.token",
now, now.plusSeconds(expiresInSeconds))
}
```
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,591 @@
+/*
+ * 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 java.io.{File, PrintWriter}
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ if (tokenFile != null) {
+ tokenFile.delete()
+ }
+ super.afterEach()
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ val pw = new PrintWriter(tokenFile)
+ try {
+ pw.print(content)
+ } finally {
+ pw.close()
+ }
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ // Verify callback was invoked with initial credentials
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ // Deserialize and verify the credential content
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ // FakeCredentialProvider resolves for scheme "fake"
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L) // 2s before expiry
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L) // 500ms min interval
+
+ // Create a short-lived context (expires in 3 seconds)
+ val ctx = createUserContext(expiresInSeconds = 3)
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, _) => {
+ latestVersion.set(version)
+ updateCount.incrementAndGet()
+ })
+
+ try {
+ manager.start()
+ assert(updateCount.get() == 1, "Should have exactly 1 update after start()")
+
+ // Wait for renewal to trigger (safety margin is 2s, credential expires in 3s,
+ // so renewal should happen around t=1s)
+ eventually(timeout(10.seconds)) {
+ assert(updateCount.get() >= 2,
+ s"Expected at least 2 updates (got ${updateCount.get()}), " +
+ "indicating credential renewal occurred")
+ }
+
+ // Verify version is monotonically increasing
+ assert(latestVersion.get() >= 2L,
+ "Version should be at least 2 after renewal")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: per-user identity token produces valid credentials") {
+ val conf = createOidcConf()
+
+ // Create a per-user context (different principal, same mechanism)
+ val userCtx = createUserContext(principal = "alice@corp.example.com")
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(userCtx),
+ (_, bytes) => callbackRef.set(bytes))
+
+ try {
+ val (_, initialBytes) = manager.start()
+
+ // Verify per-user token produces the same credential structure as workload token
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent,
+ "Per-user token should produce credentials for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Per-user credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Per-user credential should not be expired")
+
+ // Verify the credential is identical in structure to workload token output
+ // (FakeCredentialProvider doesn't distinguish -- same ServiceCredential for any UserContext)
+ val workloadCtx = createUserContext(principal = "workload-identity")
+ val workloadManager = new UserCredentialManager(
+ conf,
+ createIngestor(workloadCtx),
+ (_, _) => ())
+ try {
+ val (_, workloadBytes) = workloadManager.start()
+ val workloadCreds = UserCredentialManager.deserializeUserCredentials(workloadBytes)
+ val workloadFake = workloadCreds.forScheme("fake")
+ assert(workloadFake.isPresent)
+ // Both produce the same credential properties (provider=fake)
+ assert(workloadFake.get().getProperties == fakeCred.get().getProperties,
+ "Per-user and workload tokens should produce identical credential properties")
+ } finally {
+ workloadManager.stop()
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: UserCredentialManager and HadoopDelegationTokenManager coexist") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // Configure for BOTH OIDC and direct credential providers (non-Kerberos DT path)
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+ val oidcVersion = new AtomicReference[Long](0L)
+
+ // Start UserCredentialManager (OIDC)
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ oidcVersion.set(version)
+ oidcCallbackRef.set(bytes)
+ })
+
+ // Start HadoopDelegationTokenManager (Kerberos/direct)
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // Start OIDC manager
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L)
+ assert(oidcBytes != null)
+
+ // Start DT manager (uses TestNonKerberosTokenProvider from NonKerberosCredentialsSuite)
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should produce tokens")
+
+ // Verify OIDC credentials are valid
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should contain 'fake' scheme")
+
+ // Verify DT credentials were sent via RPC
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ eventually(timeout(5.seconds)) {
+ verify(mockRef).send(captor.capture())
+ }
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should contain test.direct.credential")
+ assert(new String(dtCreds.getSecretKey(new Text("test.direct.credential"))) === "test-token",
+ "DT credential value should match")
+
+ // Both systems produced credentials independently
+ assert(oidcVersion.get() == 1L, "OIDC version should remain at 1")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC failure does not affect HadoopDelegationTokenManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ // OIDC manager with a FAILING ingestor (simulates missing/corrupt token file)
+ val failingOidcManager = new UserCredentialManager(
+ conf,
+ createFailingIngestor(),
+ (_, _) => ())
+
+ // DT manager should work independently
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // OIDC start should fail (empty token)
+ val oidcException = intercept[Exception] {
+ failingOidcManager.start()
+ }
+ assert(oidcException != null, "OIDC manager should fail with empty token")
+
+ // DT manager should still work perfectly despite OIDC failure
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should succeed despite OIDC failure")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should be unaffected by OIDC failure")
+ } finally {
+ failingOidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: DT provider failure does not affect UserCredentialManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // Enable direct providers but disable all succeeding providers
+ // so only the failing one runs (forces DT total failure)
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+ .set("spark.security.credentials.test-direct.enabled", "false")
+ .set("spark.security.credentials.test-noexpiry.enabled", "false")
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => oidcCallbackRef.set(bytes))
+
+ // DT manager with only failing provider active
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // DT manager start returns null when all providers fail
+ val dtTokens = dtManager.start()
+ assert(dtTokens == null, "DT manager should return null when all providers fail")
+
+ // OIDC manager should still work perfectly
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L, "OIDC version should be 1")
+ assert(oidcBytes != null, "OIDC should produce credentials despite DT failure")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should be unaffected by DT failure")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+
+ test("SPARK-57896: UserCredentials serialization roundtrip preserves all schemes") {
Review Comment:
This test only tests deserialization idempotency, not a full serialize->deserialize->serialize roundtrip.
Please consider renaming to match actual verification.
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,591 @@
+/*
+ * 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 java.io.{File, PrintWriter}
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ if (tokenFile != null) {
+ tokenFile.delete()
+ }
+ super.afterEach()
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ val pw = new PrintWriter(tokenFile)
+ try {
+ pw.print(content)
+ } finally {
+ pw.close()
+ }
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ // Verify callback was invoked with initial credentials
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ // Deserialize and verify the credential content
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ // FakeCredentialProvider resolves for scheme "fake"
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L) // 2s before expiry
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L) // 500ms min interval
+
+ // Create a short-lived context (expires in 3 seconds)
+ val ctx = createUserContext(expiresInSeconds = 3)
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, _) => {
+ latestVersion.set(version)
+ updateCount.incrementAndGet()
+ })
+
+ try {
+ manager.start()
+ assert(updateCount.get() == 1, "Should have exactly 1 update after start()")
+
+ // Wait for renewal to trigger (safety margin is 2s, credential expires in 3s,
+ // so renewal should happen around t=1s)
+ eventually(timeout(10.seconds)) {
+ assert(updateCount.get() >= 2,
+ s"Expected at least 2 updates (got ${updateCount.get()}), " +
+ "indicating credential renewal occurred")
+ }
+
+ // Verify version is monotonically increasing
+ assert(latestVersion.get() >= 2L,
+ "Version should be at least 2 after renewal")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: per-user identity token produces valid credentials") {
+ val conf = createOidcConf()
+
+ // Create a per-user context (different principal, same mechanism)
+ val userCtx = createUserContext(principal = "alice@corp.example.com")
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(userCtx),
+ (_, bytes) => callbackRef.set(bytes))
+
+ try {
+ val (_, initialBytes) = manager.start()
+
+ // Verify per-user token produces the same credential structure as workload token
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent,
+ "Per-user token should produce credentials for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Per-user credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Per-user credential should not be expired")
+
+ // Verify the credential is identical in structure to workload token output
+ // (FakeCredentialProvider doesn't distinguish -- same ServiceCredential for any UserContext)
+ val workloadCtx = createUserContext(principal = "workload-identity")
+ val workloadManager = new UserCredentialManager(
+ conf,
+ createIngestor(workloadCtx),
+ (_, _) => ())
+ try {
+ val (_, workloadBytes) = workloadManager.start()
+ val workloadCreds = UserCredentialManager.deserializeUserCredentials(workloadBytes)
+ val workloadFake = workloadCreds.forScheme("fake")
+ assert(workloadFake.isPresent)
+ // Both produce the same credential properties (provider=fake)
+ assert(workloadFake.get().getProperties == fakeCred.get().getProperties,
+ "Per-user and workload tokens should produce identical credential properties")
+ } finally {
+ workloadManager.stop()
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: UserCredentialManager and HadoopDelegationTokenManager coexist") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // Configure for BOTH OIDC and direct credential providers (non-Kerberos DT path)
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+ val oidcVersion = new AtomicReference[Long](0L)
+
+ // Start UserCredentialManager (OIDC)
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ oidcVersion.set(version)
+ oidcCallbackRef.set(bytes)
+ })
+
+ // Start HadoopDelegationTokenManager (Kerberos/direct)
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // Start OIDC manager
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L)
+ assert(oidcBytes != null)
+
+ // Start DT manager (uses TestNonKerberosTokenProvider from NonKerberosCredentialsSuite)
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should produce tokens")
+
+ // Verify OIDC credentials are valid
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should contain 'fake' scheme")
+
+ // Verify DT credentials were sent via RPC
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ eventually(timeout(5.seconds)) {
+ verify(mockRef).send(captor.capture())
+ }
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should contain test.direct.credential")
+ assert(new String(dtCreds.getSecretKey(new Text("test.direct.credential"))) === "test-token",
+ "DT credential value should match")
+
+ // Both systems produced credentials independently
+ assert(oidcVersion.get() == 1L, "OIDC version should remain at 1")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC failure does not affect HadoopDelegationTokenManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ // OIDC manager with a FAILING ingestor (simulates missing/corrupt token file)
+ val failingOidcManager = new UserCredentialManager(
+ conf,
+ createFailingIngestor(),
+ (_, _) => ())
+
+ // DT manager should work independently
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // OIDC start should fail (empty token)
+ val oidcException = intercept[Exception] {
+ failingOidcManager.start()
+ }
+ assert(oidcException != null, "OIDC manager should fail with empty token")
+
+ // DT manager should still work perfectly despite OIDC failure
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should succeed despite OIDC failure")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should be unaffected by OIDC failure")
+ } finally {
+ failingOidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: DT provider failure does not affect UserCredentialManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // Enable direct providers but disable all succeeding providers
+ // so only the failing one runs (forces DT total failure)
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+ .set("spark.security.credentials.test-direct.enabled", "false")
+ .set("spark.security.credentials.test-noexpiry.enabled", "false")
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => oidcCallbackRef.set(bytes))
+
+ // DT manager with only failing provider active
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // DT manager start returns null when all providers fail
+ val dtTokens = dtManager.start()
+ assert(dtTokens == null, "DT manager should return null when all providers fail")
+
+ // OIDC manager should still work perfectly
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L, "OIDC version should be 1")
+ assert(oidcBytes != null, "OIDC should produce credentials despite DT failure")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should be unaffected by DT failure")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+
+ test("SPARK-57896: UserCredentials serialization roundtrip preserves all schemes") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ // Simulate what the executor does: deserialize the byte array
+ val bytes = serializedRef.get()
+ assert(bytes != null && bytes.length > 0, "Serialized credentials should be non-empty")
+
+ // First deserialization
+ val creds1 = UserCredentialManager.deserializeUserCredentials(bytes)
+ // Second deserialization of the same bytes (idempotency)
+ val creds2 = UserCredentialManager.deserializeUserCredentials(bytes)
+
+ // Both produce identical results
+ assert(creds1.forScheme("fake").isPresent)
+ assert(creds2.forScheme("fake").isPresent)
+ assert(creds1.forScheme("fake").get().getProperties ==
+ creds2.forScheme("fake").get().getProperties,
+ "Multiple deserializations of same bytes should produce identical credentials")
+
+ // Verify the credential is well-formed
+ val cred = creds1.forScheme("fake").get()
+ assert(cred.getProperties.containsKey("provider"))
+ assert(cred.getExpiresAt != null, "Credential should have an expiry set")
+ assert(!cred.isExpired(Instant.now()), "Freshly resolved credential should not be expired")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: multiple URI schemes resolved in single credential bundle") {
+ // FakeCredentialProvider supports both "fake" and "shared" schemes.
+ // When multiple target URIs are configured, all are resolved.
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ val creds = UserCredentialManager.deserializeUserCredentials(serializedRef.get())
+
+ // FakeCredentialProvider declares supportedSchemes = Set("fake", "shared")
+ // but "shared" is ambiguous (AnotherFakeCredentialProvider also claims it),
+ // so only "fake" auto-resolves without explicit config.
+ assert(creds.forScheme("fake").isPresent, "Should resolve 'fake' scheme")
+
+ // Verify case-insensitive lookup works
+ assert(creds.forScheme("FAKE").isPresent,
+ "Scheme lookup should be case-insensitive")
+ assert(creds.forScheme("Fake").isPresent,
+ "Scheme lookup should be case-insensitive")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: ServiceCredential.isExpired detects expiry correctly") {
+ // Unit-level verification that the executor can detect stale credentials
+ val now = Instant.now()
+
+ // Credential that expires in the future -- not expired
+ val validCred = new ServiceCredential(
+ java.util.Map.of("provider", "test"), now.plusSeconds(300))
+ assert(!validCred.isExpired(now), "Future-expiry credential should not be expired")
+
+ // Credential that already expired -- is expired
+ val expiredCred = new ServiceCredential(
+ java.util.Map.of("provider", "test"), now.minusSeconds(10))
+ assert(expiredCred.isExpired(now), "Past-expiry credential should be expired")
+
+ // Credential with null expiry -- never expires
+ val noExpiryCred = new ServiceCredential(
+ java.util.Map.of("provider", "test"), null)
+ assert(!noExpiryCred.isExpired(now),
+ "Null-expiry credential should never be considered expired")
+ }
+
+ test("SPARK-57896: stop() after start() completes cleanly without exceptions") {
+ val conf = createOidcConf()
+ val ctx = createUserContext(expiresInSeconds = 60)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, _) => ())
+
+ // start + immediate stop should not throw or leave dangling threads
+ manager.start()
+ manager.stop()
+
+ // Double stop should also be safe
+ manager.stop()
+ }
+
+ test("SPARK-57896: credential version is monotonically increasing across renewals") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val ctx = createUserContext(expiresInSeconds = 3)
+ val versions = new java.util.concurrent.CopyOnWriteArrayList[Long]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, _) => versions.add(version))
+
+ try {
+ manager.start()
+
+ // Wait for at least 3 renewals
+ eventually(timeout(15.seconds)) {
+ assert(versions.size() >= 3,
+ s"Expected at least 3 callbacks (got ${versions.size()})")
+ }
+
+ // Verify strict monotonicity
+ val versionList = new java.util.ArrayList(versions)
+ for (i <- 1 until versionList.size()) {
+ assert(versionList.get(i) > versionList.get(i - 1),
+ s"Version ${versionList.get(i)} should be > ${versionList.get(i - 1)} " +
+ s"at index $i (full list: $versionList)")
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: every renewal callback provides non-null non-empty credentials") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val ctx = createUserContext(expiresInSeconds = 3)
+ val allBytes = new java.util.concurrent.CopyOnWriteArrayList[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => allBytes.add(bytes))
+
+ try {
+ manager.start()
+
+ eventually(timeout(15.seconds)) {
+ assert(allBytes.size() >= 2,
+ s"Expected at least 2 callbacks (got ${allBytes.size()})")
+ }
+
+ // Every single callback must have valid, deserializable credentials
+ val it = allBytes.iterator()
+ while (it.hasNext) {
+ val bytes = it.next()
+ assert(bytes != null, "Callback bytes should never be null")
+ assert(bytes.length > 0, "Callback bytes should never be empty")
+ val creds = UserCredentialManager.deserializeUserCredentials(bytes)
+ assert(creds.forScheme("fake").isPresent,
+ "Every renewed credential bundle should contain 'fake' scheme")
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC disabled does not interfere with DT manager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // OIDC explicitly DISABLED -- only DT enabled
+ val conf = new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, false)
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ // UserCredentialManager.create() should return None
+ val oidcManager = UserCredentialManager.create(conf, (_, _) => ())
+ assert(oidcManager.isEmpty, "OIDC manager should not be created when disabled")
+
+ // DT manager should work independently
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+ try {
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should work when OIDC is disabled")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ eventually(timeout(5.seconds)) {
+ verify(mockRef).send(captor.capture())
Review Comment:
Ditto.
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,591 @@
+/*
+ * 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 java.io.{File, PrintWriter}
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ if (tokenFile != null) {
+ tokenFile.delete()
+ }
+ super.afterEach()
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ val pw = new PrintWriter(tokenFile)
+ try {
+ pw.print(content)
+ } finally {
+ pw.close()
+ }
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ // Verify callback was invoked with initial credentials
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ // Deserialize and verify the credential content
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ // FakeCredentialProvider resolves for scheme "fake"
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L) // 2s before expiry
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L) // 500ms min interval
+
+ // Create a short-lived context (expires in 3 seconds)
+ val ctx = createUserContext(expiresInSeconds = 3)
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, _) => {
+ latestVersion.set(version)
+ updateCount.incrementAndGet()
+ })
+
+ try {
+ manager.start()
+ assert(updateCount.get() == 1, "Should have exactly 1 update after start()")
+
+ // Wait for renewal to trigger (safety margin is 2s, credential expires in 3s,
+ // so renewal should happen around t=1s)
+ eventually(timeout(10.seconds)) {
+ assert(updateCount.get() >= 2,
+ s"Expected at least 2 updates (got ${updateCount.get()}), " +
+ "indicating credential renewal occurred")
+ }
+
+ // Verify version is monotonically increasing
+ assert(latestVersion.get() >= 2L,
+ "Version should be at least 2 after renewal")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: per-user identity token produces valid credentials") {
+ val conf = createOidcConf()
+
+ // Create a per-user context (different principal, same mechanism)
+ val userCtx = createUserContext(principal = "alice@corp.example.com")
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(userCtx),
+ (_, bytes) => callbackRef.set(bytes))
+
+ try {
+ val (_, initialBytes) = manager.start()
+
+ // Verify per-user token produces the same credential structure as workload token
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent,
+ "Per-user token should produce credentials for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Per-user credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Per-user credential should not be expired")
+
+ // Verify the credential is identical in structure to workload token output
+ // (FakeCredentialProvider doesn't distinguish -- same ServiceCredential for any UserContext)
+ val workloadCtx = createUserContext(principal = "workload-identity")
+ val workloadManager = new UserCredentialManager(
+ conf,
+ createIngestor(workloadCtx),
+ (_, _) => ())
+ try {
+ val (_, workloadBytes) = workloadManager.start()
+ val workloadCreds = UserCredentialManager.deserializeUserCredentials(workloadBytes)
+ val workloadFake = workloadCreds.forScheme("fake")
+ assert(workloadFake.isPresent)
+ // Both produce the same credential properties (provider=fake)
+ assert(workloadFake.get().getProperties == fakeCred.get().getProperties,
+ "Per-user and workload tokens should produce identical credential properties")
+ } finally {
+ workloadManager.stop()
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: UserCredentialManager and HadoopDelegationTokenManager coexist") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // Configure for BOTH OIDC and direct credential providers (non-Kerberos DT path)
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+ val oidcVersion = new AtomicReference[Long](0L)
+
+ // Start UserCredentialManager (OIDC)
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ oidcVersion.set(version)
+ oidcCallbackRef.set(bytes)
+ })
+
+ // Start HadoopDelegationTokenManager (Kerberos/direct)
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // Start OIDC manager
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L)
+ assert(oidcBytes != null)
+
+ // Start DT manager (uses TestNonKerberosTokenProvider from NonKerberosCredentialsSuite)
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should produce tokens")
+
+ // Verify OIDC credentials are valid
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should contain 'fake' scheme")
+
+ // Verify DT credentials were sent via RPC
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ eventually(timeout(5.seconds)) {
+ verify(mockRef).send(captor.capture())
+ }
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should contain test.direct.credential")
+ assert(new String(dtCreds.getSecretKey(new Text("test.direct.credential"))) === "test-token",
+ "DT credential value should match")
+
+ // Both systems produced credentials independently
+ assert(oidcVersion.get() == 1L, "OIDC version should remain at 1")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC failure does not affect HadoopDelegationTokenManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ // OIDC manager with a FAILING ingestor (simulates missing/corrupt token file)
+ val failingOidcManager = new UserCredentialManager(
+ conf,
+ createFailingIngestor(),
+ (_, _) => ())
+
+ // DT manager should work independently
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // OIDC start should fail (empty token)
+ val oidcException = intercept[Exception] {
+ failingOidcManager.start()
+ }
+ assert(oidcException != null, "OIDC manager should fail with empty token")
+
+ // DT manager should still work perfectly despite OIDC failure
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should succeed despite OIDC failure")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should be unaffected by OIDC failure")
+ } finally {
+ failingOidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: DT provider failure does not affect UserCredentialManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ // Enable direct providers but disable all succeeding providers
+ // so only the failing one runs (forces DT total failure)
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+ .set("spark.security.credentials.test-direct.enabled", "false")
+ .set("spark.security.credentials.test-noexpiry.enabled", "false")
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => oidcCallbackRef.set(bytes))
+
+ // DT manager with only failing provider active
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // DT manager start returns null when all providers fail
+ val dtTokens = dtManager.start()
+ assert(dtTokens == null, "DT manager should return null when all providers fail")
+
+ // OIDC manager should still work perfectly
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L, "OIDC version should be 1")
+ assert(oidcBytes != null, "OIDC should produce credentials despite DT failure")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should be unaffected by DT failure")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+
+ test("SPARK-57896: UserCredentials serialization roundtrip preserves all schemes") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ // Simulate what the executor does: deserialize the byte array
+ val bytes = serializedRef.get()
+ assert(bytes != null && bytes.length > 0, "Serialized credentials should be non-empty")
+
+ // First deserialization
+ val creds1 = UserCredentialManager.deserializeUserCredentials(bytes)
+ // Second deserialization of the same bytes (idempotency)
+ val creds2 = UserCredentialManager.deserializeUserCredentials(bytes)
+
+ // Both produce identical results
+ assert(creds1.forScheme("fake").isPresent)
+ assert(creds2.forScheme("fake").isPresent)
+ assert(creds1.forScheme("fake").get().getProperties ==
+ creds2.forScheme("fake").get().getProperties,
+ "Multiple deserializations of same bytes should produce identical credentials")
+
+ // Verify the credential is well-formed
+ val cred = creds1.forScheme("fake").get()
+ assert(cred.getProperties.containsKey("provider"))
+ assert(cred.getExpiresAt != null, "Credential should have an expiry set")
+ assert(!cred.isExpired(Instant.now()), "Freshly resolved credential should not be expired")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: multiple URI schemes resolved in single credential bundle") {
+ // FakeCredentialProvider supports both "fake" and "shared" schemes.
+ // When multiple target URIs are configured, all are resolved.
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ val creds = UserCredentialManager.deserializeUserCredentials(serializedRef.get())
+
+ // FakeCredentialProvider declares supportedSchemes = Set("fake", "shared")
+ // but "shared" is ambiguous (AnotherFakeCredentialProvider also claims it),
+ // so only "fake" auto-resolves without explicit config.
+ assert(creds.forScheme("fake").isPresent, "Should resolve 'fake' scheme")
+
+ // Verify case-insensitive lookup works
+ assert(creds.forScheme("FAKE").isPresent,
+ "Scheme lookup should be case-insensitive")
+ assert(creds.forScheme("Fake").isPresent,
+ "Scheme lookup should be case-insensitive")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: ServiceCredential.isExpired detects expiry correctly") {
Review Comment:
This is a pure unit test of a single class method. Please consider moving to a `ServiceCredentialSuite` for better test organization in another PR.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57896][CORE][TESTS] Add Kerberos coexistence and per-user token integration tests [spark]
Posted by "shrirangmhalgi (via GitHub)" <gi...@apache.org>.
shrirangmhalgi commented on PR #57866:
URL: https://github.com/apache/spark/pull/57866#issuecomment-5224932041
@sarutak Could you please review the PR. It adds the integration test coverage for the OIDC credential framework.
Thank you
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57896][CORE][TESTS] Add Kerberos coexistence and per-user token integration tests [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on PR #57866:
URL: https://github.com/apache/spark/pull/57866#issuecomment-5298917392
Merged. Thank you @shrirangmhalgi !
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57896][CORE][TESTS] Add Kerberos coexistence and per-user token integration tests [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #57866:
URL: https://github.com/apache/spark/pull/57866#discussion_r3762710059
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,582 @@
+/*
+ * 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 java.io.File
+import java.nio.file.Files
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+import org.apache.spark.{SparkConf, SparkFunSuite, VersionedCredentials}
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ * 6. TaskDescription credentials are applied to the executor store with version guard
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ try {
+ if (tokenFile != null) tokenFile.delete()
+ } finally {
+ CredentialProviderLoader.resetForTesting()
+ super.afterEach()
+ }
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ Files.writeString(tokenFile.toPath, content)
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFreshExpiryIngestor(expiresInSeconds: Long): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] =
+ Optional.of(createUserContext(expiresInSeconds = expiresInSeconds))
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
Review Comment:
The `SPARK-57896:` prefix on every test name is not required for a new test suite (per the [Contributing Guide](https://spark.apache.org/contributing.html), it is typically added for bug fixes or when adding a few tests to an existing class). Please consider removing it.
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,582 @@
+/*
+ * 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 java.io.File
+import java.nio.file.Files
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+import org.apache.spark.{SparkConf, SparkFunSuite, VersionedCredentials}
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ * 6. TaskDescription credentials are applied to the executor store with version guard
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ try {
+ if (tokenFile != null) tokenFile.delete()
+ } finally {
+ CredentialProviderLoader.resetForTesting()
+ super.afterEach()
+ }
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ Files.writeString(tokenFile.toPath, content)
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFreshExpiryIngestor(expiresInSeconds: Long): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] =
+ Optional.of(createUserContext(expiresInSeconds = expiresInSeconds))
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicReference[Long](0L)
+
+ // Return a fresh UserContext on each load() so each renewal gets a genuinely
+ // new expiry rather than spinning on an already-expired token.
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (version, _) => {
+ latestVersion.set(version)
+ updateCount.incrementAndGet()
+ })
+
+ try {
+ manager.start()
+ assert(updateCount.get() == 1, "Should have exactly 1 update after start()")
+
+ eventually(timeout(15.seconds)) {
+ assert(updateCount.get() >= 2,
+ s"Expected at least 2 updates (got ${updateCount.get()}), " +
+ "indicating credential renewal occurred")
+ }
+
+ assert(latestVersion.get() >= 2L,
+ "Version should be at least 2 after renewal")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: per-user identity token produces valid credentials") {
+ val conf = createOidcConf()
+
+ val userCtx = createUserContext(principal = "alice@corp.example.com")
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(userCtx),
+ (_, bytes) => callbackRef.set(bytes))
+
+ try {
+ val (_, initialBytes) = manager.start()
+
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent,
+ "Per-user token should produce credentials for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Per-user credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Per-user credential should not be expired")
+
+ // Verify the credential is identical in structure to workload token output
+ val workloadCtx = createUserContext(principal = "workload-identity")
+ val workloadManager = new UserCredentialManager(
+ conf,
+ createIngestor(workloadCtx),
+ (_, _) => ())
+ try {
+ val (_, workloadBytes) = workloadManager.start()
+ val workloadCreds = UserCredentialManager.deserializeUserCredentials(workloadBytes)
+ val workloadFake = workloadCreds.forScheme("fake")
+ assert(workloadFake.isPresent)
+ assert(workloadFake.get().getProperties == fakeCred.get().getProperties,
+ "Per-user and workload tokens should produce identical credential properties")
+ } finally {
+ workloadManager.stop()
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: UserCredentialManager and HadoopDelegationTokenManager coexist") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+ val oidcVersion = new AtomicReference[Long](0L)
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ oidcVersion.set(version)
+ oidcCallbackRef.set(bytes)
+ })
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L)
+ assert(oidcBytes != null)
+
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should produce tokens")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should contain 'fake' scheme")
+
+ // HadoopDelegationTokenManager.start() is synchronous -- verify directly
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should contain test.direct.credential")
+ assert(new String(dtCreds.getSecretKey(new Text("test.direct.credential"))) === "test-token",
+ "DT credential value should match")
+
+ assert(oidcVersion.get() == 1L, "OIDC version should remain at 1")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC failure does not affect HadoopDelegationTokenManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val failingOidcManager = new UserCredentialManager(
+ conf,
+ createFailingIngestor(),
+ (_, _) => ())
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // OIDC start should fail with IllegalStateException (missing token)
+ val oidcException = intercept[IllegalStateException] {
+ failingOidcManager.start()
+ }
+ assert(oidcException.getMessage.contains(
+ "identity token file is missing or malformed"))
+
+ // DT manager should still work perfectly despite OIDC failure
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should succeed despite OIDC failure")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should be unaffected by OIDC failure")
+ } finally {
+ failingOidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: DT provider failure does not affect UserCredentialManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+ .set("spark.security.credentials.test-direct.enabled", "false")
+ .set("spark.security.credentials.test-noexpiry.enabled", "false")
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => oidcCallbackRef.set(bytes))
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ val dtTokens = dtManager.start()
+ assert(dtTokens == null, "DT manager should return null when all providers fail")
+
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L, "OIDC version should be 1")
+ assert(oidcBytes != null, "OIDC should produce credentials despite DT failure")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should be unaffected by DT failure")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: deserialization idempotency preserves credential content") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ val bytes = serializedRef.get()
+ assert(bytes != null && bytes.length > 0, "Serialized credentials should be non-empty")
+
+ val creds1 = UserCredentialManager.deserializeUserCredentials(bytes)
+ val creds2 = UserCredentialManager.deserializeUserCredentials(bytes)
+
+ assert(creds1.forScheme("fake").isPresent)
+ assert(creds2.forScheme("fake").isPresent)
+ assert(creds1.forScheme("fake").get().getProperties ==
+ creds2.forScheme("fake").get().getProperties,
+ "Multiple deserializations of same bytes should produce identical credentials")
+
+ val cred = creds1.forScheme("fake").get()
+ assert(cred.getProperties.containsKey("provider"))
+ assert(cred.getExpiresAt != null, "Credential should have an expiry set")
+ assert(!cred.isExpired(Instant.now()), "Freshly resolved credential should not be expired")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: case-insensitive scheme lookup in credential bundle") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ val creds = UserCredentialManager.deserializeUserCredentials(serializedRef.get())
+
+ // FakeCredentialProvider declares supportedSchemes = Set("fake", "shared")
+ // but "shared" is ambiguous (AnotherFakeCredentialProvider also claims it),
+ // so only "fake" auto-resolves without explicit config.
+ assert(creds.forScheme("fake").isPresent, "Should resolve 'fake' scheme")
+ assert(creds.forScheme("FAKE").isPresent,
+ "Scheme lookup should be case-insensitive")
+ assert(creds.forScheme("Fake").isPresent,
+ "Scheme lookup should be case-insensitive")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: stop() after start() completes cleanly without exceptions") {
+ val conf = createOidcConf()
+ val ctx = createUserContext(expiresInSeconds = 60)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, _) => ())
+
+ manager.start()
+ manager.stop()
+
+ // Double stop should also be safe
+ manager.stop()
+ }
+
+ test("SPARK-57896: credential version is monotonically increasing across renewals") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val versions = new java.util.concurrent.CopyOnWriteArrayList[Long]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (version, _) => versions.add(version))
+
+ try {
+ manager.start()
+
+ eventually(timeout(15.seconds)) {
+ assert(versions.size() >= 3,
+ s"Expected at least 3 callbacks (got ${versions.size()})")
+ }
+
+ val versionList = new java.util.ArrayList(versions)
+ for (i <- 1 until versionList.size()) {
+ assert(versionList.get(i) > versionList.get(i - 1),
+ s"Version ${versionList.get(i)} should be > ${versionList.get(i - 1)} " +
+ s"at index $i (full list: $versionList)")
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: every renewal callback provides non-null non-empty credentials") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val allBytes = new java.util.concurrent.CopyOnWriteArrayList[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (_, bytes) => allBytes.add(bytes))
+
+ try {
+ manager.start()
+
+ eventually(timeout(15.seconds)) {
+ assert(allBytes.size() >= 2,
+ s"Expected at least 2 callbacks (got ${allBytes.size()})")
+ }
+
+ val it = allBytes.iterator()
+ while (it.hasNext) {
+ val bytes = it.next()
+ assert(bytes != null, "Callback bytes should never be null")
+ assert(bytes.length > 0, "Callback bytes should never be empty")
+ val creds = UserCredentialManager.deserializeUserCredentials(bytes)
+ assert(creds.forScheme("fake").isPresent,
+ "Every renewed credential bundle should contain 'fake' scheme")
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC disabled does not interfere with DT manager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, false)
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val oidcManager = UserCredentialManager.create(conf, (_, _) => ())
+ assert(oidcManager.isEmpty, "OIDC manager should not be created when disabled")
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+ try {
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should work when OIDC is disabled")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null)
+ } finally {
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: TaskDescription credentials applied to executor store with version guard") {
Review Comment:
This test calls `VersionedCredentials.updateIfNewer` directly. It does not exercise `TaskRunner`, `CoarseGrainedExecutorBackend.receive`, or any actual executor code path. It is a unit test of the `updateIfNewer` helper method.
**Suggestions:**
- The test name's intent ("TaskDescription credentials applied to executor store with version guard") is correct, but the test content does not fulfill it — it only calls `VersionedCredentials.updateIfNewer` directly without going through the actual `TaskDescription` → Executor path. Please update the test to exercise the real code path, or add a separate test that does.
- **Background:** In [PR #57675](https://github.com/apache/spark/pull/57675) (Sub-task 5), dongjoon-hyun and HyukjinKwon pointed out that `TaskSetManager.resourceOffer` attaching credentials to `TaskDescription` was not tested. I (sarutak) responded that it would be covered in Sub-task 7 (this PR). Please add a test that covers this path.
**Expected test outline:**
- This can be implemented in `TaskSetManagerSuite` (which already has the necessary infrastructure: `FakeSchedulerBackend`, `TaskSchedulerImpl`, and calls to `manager.resourceOffer(...)` returning `TaskDescription`)
- Set `SparkEnv.get.userCredentials` to `VersionedCredentials(version=1, bytes=<valid serialized UserCredentials>)`
- Call `manager.resourceOffer(...)` and assert that the returned `TaskDescription.userCredentials == Some((1L, bytes))`
- Update the store to version 2, call `resourceOffer` again, and verify the new `TaskDescription` carries version 2
- This proves that `TaskSetManager` reads the current credential store at dispatch time (line 607 of `TaskSetManager.scala`: `Option(env.userCredentials.get()).map(vc => (vc.version, vc.bytes))`) and includes it in every `TaskDescription`
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,582 @@
+/*
+ * 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 java.io.File
+import java.nio.file.Files
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+import org.apache.spark.{SparkConf, SparkFunSuite, VersionedCredentials}
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ * 6. TaskDescription credentials are applied to the executor store with version guard
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ try {
+ if (tokenFile != null) tokenFile.delete()
+ } finally {
+ CredentialProviderLoader.resetForTesting()
+ super.afterEach()
+ }
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ Files.writeString(tokenFile.toPath, content)
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFreshExpiryIngestor(expiresInSeconds: Long): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] =
+ Optional.of(createUserContext(expiresInSeconds = expiresInSeconds))
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicReference[Long](0L)
+
+ // Return a fresh UserContext on each load() so each renewal gets a genuinely
+ // new expiry rather than spinning on an already-expired token.
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (version, _) => {
+ latestVersion.set(version)
+ updateCount.incrementAndGet()
+ })
+
+ try {
+ manager.start()
+ assert(updateCount.get() == 1, "Should have exactly 1 update after start()")
+
+ eventually(timeout(15.seconds)) {
+ assert(updateCount.get() >= 2,
+ s"Expected at least 2 updates (got ${updateCount.get()}), " +
+ "indicating credential renewal occurred")
+ }
+
+ assert(latestVersion.get() >= 2L,
+ "Version should be at least 2 after renewal")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: per-user identity token produces valid credentials") {
+ val conf = createOidcConf()
+
+ val userCtx = createUserContext(principal = "alice@corp.example.com")
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(userCtx),
+ (_, bytes) => callbackRef.set(bytes))
+
+ try {
+ val (_, initialBytes) = manager.start()
+
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent,
+ "Per-user token should produce credentials for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Per-user credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Per-user credential should not be expired")
+
+ // Verify the credential is identical in structure to workload token output
+ val workloadCtx = createUserContext(principal = "workload-identity")
+ val workloadManager = new UserCredentialManager(
+ conf,
+ createIngestor(workloadCtx),
+ (_, _) => ())
+ try {
+ val (_, workloadBytes) = workloadManager.start()
+ val workloadCreds = UserCredentialManager.deserializeUserCredentials(workloadBytes)
+ val workloadFake = workloadCreds.forScheme("fake")
+ assert(workloadFake.isPresent)
+ assert(workloadFake.get().getProperties == fakeCred.get().getProperties,
+ "Per-user and workload tokens should produce identical credential properties")
+ } finally {
+ workloadManager.stop()
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: UserCredentialManager and HadoopDelegationTokenManager coexist") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+ val oidcVersion = new AtomicReference[Long](0L)
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ oidcVersion.set(version)
+ oidcCallbackRef.set(bytes)
+ })
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L)
+ assert(oidcBytes != null)
+
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should produce tokens")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should contain 'fake' scheme")
+
+ // HadoopDelegationTokenManager.start() is synchronous -- verify directly
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should contain test.direct.credential")
+ assert(new String(dtCreds.getSecretKey(new Text("test.direct.credential"))) === "test-token",
+ "DT credential value should match")
+
+ assert(oidcVersion.get() == 1L, "OIDC version should remain at 1")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC failure does not affect HadoopDelegationTokenManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val failingOidcManager = new UserCredentialManager(
+ conf,
+ createFailingIngestor(),
+ (_, _) => ())
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // OIDC start should fail with IllegalStateException (missing token)
+ val oidcException = intercept[IllegalStateException] {
+ failingOidcManager.start()
+ }
+ assert(oidcException.getMessage.contains(
+ "identity token file is missing or malformed"))
+
+ // DT manager should still work perfectly despite OIDC failure
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should succeed despite OIDC failure")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should be unaffected by OIDC failure")
+ } finally {
+ failingOidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: DT provider failure does not affect UserCredentialManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+ .set("spark.security.credentials.test-direct.enabled", "false")
+ .set("spark.security.credentials.test-noexpiry.enabled", "false")
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => oidcCallbackRef.set(bytes))
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ val dtTokens = dtManager.start()
+ assert(dtTokens == null, "DT manager should return null when all providers fail")
+
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L, "OIDC version should be 1")
+ assert(oidcBytes != null, "OIDC should produce credentials despite DT failure")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should be unaffected by DT failure")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: deserialization idempotency preserves credential content") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ val bytes = serializedRef.get()
+ assert(bytes != null && bytes.length > 0, "Serialized credentials should be non-empty")
+
+ val creds1 = UserCredentialManager.deserializeUserCredentials(bytes)
+ val creds2 = UserCredentialManager.deserializeUserCredentials(bytes)
+
+ assert(creds1.forScheme("fake").isPresent)
+ assert(creds2.forScheme("fake").isPresent)
+ assert(creds1.forScheme("fake").get().getProperties ==
+ creds2.forScheme("fake").get().getProperties,
+ "Multiple deserializations of same bytes should produce identical credentials")
+
+ val cred = creds1.forScheme("fake").get()
+ assert(cred.getProperties.containsKey("provider"))
+ assert(cred.getExpiresAt != null, "Credential should have an expiry set")
+ assert(!cred.isExpired(Instant.now()), "Freshly resolved credential should not be expired")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: case-insensitive scheme lookup in credential bundle") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ val creds = UserCredentialManager.deserializeUserCredentials(serializedRef.get())
+
+ // FakeCredentialProvider declares supportedSchemes = Set("fake", "shared")
+ // but "shared" is ambiguous (AnotherFakeCredentialProvider also claims it),
+ // so only "fake" auto-resolves without explicit config.
+ assert(creds.forScheme("fake").isPresent, "Should resolve 'fake' scheme")
+ assert(creds.forScheme("FAKE").isPresent,
+ "Scheme lookup should be case-insensitive")
+ assert(creds.forScheme("Fake").isPresent,
+ "Scheme lookup should be case-insensitive")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: stop() after start() completes cleanly without exceptions") {
+ val conf = createOidcConf()
+ val ctx = createUserContext(expiresInSeconds = 60)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, _) => ())
+
+ manager.start()
+ manager.stop()
+
+ // Double stop should also be safe
+ manager.stop()
+ }
+
+ test("SPARK-57896: credential version is monotonically increasing across renewals") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val versions = new java.util.concurrent.CopyOnWriteArrayList[Long]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (version, _) => versions.add(version))
+
+ try {
+ manager.start()
+
+ eventually(timeout(15.seconds)) {
+ assert(versions.size() >= 3,
+ s"Expected at least 3 callbacks (got ${versions.size()})")
+ }
+
+ val versionList = new java.util.ArrayList(versions)
+ for (i <- 1 until versionList.size()) {
+ assert(versionList.get(i) > versionList.get(i - 1),
+ s"Version ${versionList.get(i)} should be > ${versionList.get(i - 1)} " +
+ s"at index $i (full list: $versionList)")
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: every renewal callback provides non-null non-empty credentials") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val allBytes = new java.util.concurrent.CopyOnWriteArrayList[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (_, bytes) => allBytes.add(bytes))
+
+ try {
+ manager.start()
+
+ eventually(timeout(15.seconds)) {
+ assert(allBytes.size() >= 2,
+ s"Expected at least 2 callbacks (got ${allBytes.size()})")
+ }
+
+ val it = allBytes.iterator()
+ while (it.hasNext) {
+ val bytes = it.next()
+ assert(bytes != null, "Callback bytes should never be null")
+ assert(bytes.length > 0, "Callback bytes should never be empty")
+ val creds = UserCredentialManager.deserializeUserCredentials(bytes)
+ assert(creds.forScheme("fake").isPresent,
+ "Every renewed credential bundle should contain 'fake' scheme")
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC disabled does not interfere with DT manager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, false)
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val oidcManager = UserCredentialManager.create(conf, (_, _) => ())
+ assert(oidcManager.isEmpty, "OIDC manager should not be created when disabled")
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+ try {
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should work when OIDC is disabled")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null)
+ } finally {
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: TaskDescription credentials applied to executor store with version guard") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, _) => ())
+
+ try {
+ val (version, credentialBytes) = manager.start()
+ assert(version == 1L)
+
+ // Simulate the executor-side credential store
+ val store = new AtomicReference[VersionedCredentials]()
+
+ // Simulate TaskDescription carrying credentials at dispatch time:
+ // Executor applies credentials via VersionedCredentials.updateIfNewer
+ VersionedCredentials.updateIfNewer(store, version, credentialBytes)
+
+ // Verify credentials were applied
+ val stored = store.get()
+ assert(stored != null, "Store should have credentials after TaskDescription apply")
+ assert(stored.version == 1L, "Stored version should be 1")
+
+ // Verify the stored bytes are deserializable and correct
+ val creds = UserCredentialManager.deserializeUserCredentials(stored.bytes)
+ assert(creds.forScheme("fake").isPresent,
+ "Executor store should contain valid 'fake' scheme credentials")
+
+ // Simulate RPC broadcast delivering v2 (fresher credentials)
+ val v2Bytes = Array[Byte](99, 98, 97)
Review Comment:
This is not a valid serialized `UserCredentials` (It cannot be deserialized). The test only exercises version comparison logic, which is fine, but a comment should clarify this:
```scala
// Dummy bytes. This test only verifies version guard logic, not credential deserialization.
val v2Bytes = Array[Byte](99, 98, 97)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57896][CORE][TESTS] Add Kerberos coexistence and per-user token integration tests [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #57866:
URL: https://github.com/apache/spark/pull/57866#discussion_r3762707343
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,582 @@
+/*
+ * 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 java.io.File
+import java.nio.file.Files
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+import org.apache.spark.{SparkConf, SparkFunSuite, VersionedCredentials}
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ * 6. TaskDescription credentials are applied to the executor store with version guard
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ try {
+ if (tokenFile != null) tokenFile.delete()
+ } finally {
+ CredentialProviderLoader.resetForTesting()
+ super.afterEach()
+ }
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ Files.writeString(tokenFile.toPath, content)
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFreshExpiryIngestor(expiresInSeconds: Long): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] =
+ Optional.of(createUserContext(expiresInSeconds = expiresInSeconds))
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicReference[Long](0L)
+
+ // Return a fresh UserContext on each load() so each renewal gets a genuinely
+ // new expiry rather than spinning on an already-expired token.
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (version, _) => {
+ latestVersion.set(version)
+ updateCount.incrementAndGet()
+ })
+
+ try {
+ manager.start()
+ assert(updateCount.get() == 1, "Should have exactly 1 update after start()")
+
+ eventually(timeout(15.seconds)) {
+ assert(updateCount.get() >= 2,
+ s"Expected at least 2 updates (got ${updateCount.get()}), " +
+ "indicating credential renewal occurred")
+ }
+
+ assert(latestVersion.get() >= 2L,
+ "Version should be at least 2 after renewal")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: per-user identity token produces valid credentials") {
+ val conf = createOidcConf()
+
+ val userCtx = createUserContext(principal = "alice@corp.example.com")
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(userCtx),
+ (_, bytes) => callbackRef.set(bytes))
+
+ try {
+ val (_, initialBytes) = manager.start()
+
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent,
+ "Per-user token should produce credentials for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Per-user credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Per-user credential should not be expired")
+
+ // Verify the credential is identical in structure to workload token output
+ val workloadCtx = createUserContext(principal = "workload-identity")
+ val workloadManager = new UserCredentialManager(
+ conf,
+ createIngestor(workloadCtx),
+ (_, _) => ())
+ try {
+ val (_, workloadBytes) = workloadManager.start()
+ val workloadCreds = UserCredentialManager.deserializeUserCredentials(workloadBytes)
+ val workloadFake = workloadCreds.forScheme("fake")
+ assert(workloadFake.isPresent)
+ assert(workloadFake.get().getProperties == fakeCred.get().getProperties,
+ "Per-user and workload tokens should produce identical credential properties")
+ } finally {
+ workloadManager.stop()
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: UserCredentialManager and HadoopDelegationTokenManager coexist") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+ val oidcVersion = new AtomicReference[Long](0L)
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ oidcVersion.set(version)
+ oidcCallbackRef.set(bytes)
+ })
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L)
+ assert(oidcBytes != null)
+
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should produce tokens")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should contain 'fake' scheme")
+
+ // HadoopDelegationTokenManager.start() is synchronous -- verify directly
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should contain test.direct.credential")
+ assert(new String(dtCreds.getSecretKey(new Text("test.direct.credential"))) === "test-token",
+ "DT credential value should match")
+
+ assert(oidcVersion.get() == 1L, "OIDC version should remain at 1")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC failure does not affect HadoopDelegationTokenManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val failingOidcManager = new UserCredentialManager(
+ conf,
+ createFailingIngestor(),
+ (_, _) => ())
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // OIDC start should fail with IllegalStateException (missing token)
+ val oidcException = intercept[IllegalStateException] {
+ failingOidcManager.start()
+ }
+ assert(oidcException.getMessage.contains(
+ "identity token file is missing or malformed"))
+
+ // DT manager should still work perfectly despite OIDC failure
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should succeed despite OIDC failure")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should be unaffected by OIDC failure")
+ } finally {
+ failingOidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: DT provider failure does not affect UserCredentialManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+ .set("spark.security.credentials.test-direct.enabled", "false")
+ .set("spark.security.credentials.test-noexpiry.enabled", "false")
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => oidcCallbackRef.set(bytes))
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ val dtTokens = dtManager.start()
+ assert(dtTokens == null, "DT manager should return null when all providers fail")
+
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L, "OIDC version should be 1")
+ assert(oidcBytes != null, "OIDC should produce credentials despite DT failure")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should be unaffected by DT failure")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: deserialization idempotency preserves credential content") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ val bytes = serializedRef.get()
+ assert(bytes != null && bytes.length > 0, "Serialized credentials should be non-empty")
+
+ val creds1 = UserCredentialManager.deserializeUserCredentials(bytes)
+ val creds2 = UserCredentialManager.deserializeUserCredentials(bytes)
+
+ assert(creds1.forScheme("fake").isPresent)
+ assert(creds2.forScheme("fake").isPresent)
+ assert(creds1.forScheme("fake").get().getProperties ==
+ creds2.forScheme("fake").get().getProperties,
+ "Multiple deserializations of same bytes should produce identical credentials")
+
+ val cred = creds1.forScheme("fake").get()
+ assert(cred.getProperties.containsKey("provider"))
+ assert(cred.getExpiresAt != null, "Credential should have an expiry set")
+ assert(!cred.isExpired(Instant.now()), "Freshly resolved credential should not be expired")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: case-insensitive scheme lookup in credential bundle") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ val creds = UserCredentialManager.deserializeUserCredentials(serializedRef.get())
+
+ // FakeCredentialProvider declares supportedSchemes = Set("fake", "shared")
+ // but "shared" is ambiguous (AnotherFakeCredentialProvider also claims it),
+ // so only "fake" auto-resolves without explicit config.
+ assert(creds.forScheme("fake").isPresent, "Should resolve 'fake' scheme")
+ assert(creds.forScheme("FAKE").isPresent,
+ "Scheme lookup should be case-insensitive")
+ assert(creds.forScheme("Fake").isPresent,
+ "Scheme lookup should be case-insensitive")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: stop() after start() completes cleanly without exceptions") {
+ val conf = createOidcConf()
+ val ctx = createUserContext(expiresInSeconds = 60)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, _) => ())
+
+ manager.start()
+ manager.stop()
+
+ // Double stop should also be safe
+ manager.stop()
+ }
+
+ test("SPARK-57896: credential version is monotonically increasing across renewals") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val versions = new java.util.concurrent.CopyOnWriteArrayList[Long]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (version, _) => versions.add(version))
+
+ try {
+ manager.start()
+
+ eventually(timeout(15.seconds)) {
+ assert(versions.size() >= 3,
+ s"Expected at least 3 callbacks (got ${versions.size()})")
+ }
+
+ val versionList = new java.util.ArrayList(versions)
+ for (i <- 1 until versionList.size()) {
+ assert(versionList.get(i) > versionList.get(i - 1),
+ s"Version ${versionList.get(i)} should be > ${versionList.get(i - 1)} " +
+ s"at index $i (full list: $versionList)")
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: every renewal callback provides non-null non-empty credentials") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val allBytes = new java.util.concurrent.CopyOnWriteArrayList[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (_, bytes) => allBytes.add(bytes))
+
+ try {
+ manager.start()
+
+ eventually(timeout(15.seconds)) {
+ assert(allBytes.size() >= 2,
+ s"Expected at least 2 callbacks (got ${allBytes.size()})")
+ }
+
+ val it = allBytes.iterator()
+ while (it.hasNext) {
+ val bytes = it.next()
+ assert(bytes != null, "Callback bytes should never be null")
+ assert(bytes.length > 0, "Callback bytes should never be empty")
+ val creds = UserCredentialManager.deserializeUserCredentials(bytes)
+ assert(creds.forScheme("fake").isPresent,
+ "Every renewed credential bundle should contain 'fake' scheme")
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC disabled does not interfere with DT manager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, false)
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val oidcManager = UserCredentialManager.create(conf, (_, _) => ())
+ assert(oidcManager.isEmpty, "OIDC manager should not be created when disabled")
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+ try {
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should work when OIDC is disabled")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null)
+ } finally {
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: TaskDescription credentials applied to executor store with version guard") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, _) => ())
+
+ try {
+ val (version, credentialBytes) = manager.start()
+ assert(version == 1L)
+
+ // Simulate the executor-side credential store
+ val store = new AtomicReference[VersionedCredentials]()
+
+ // Simulate TaskDescription carrying credentials at dispatch time:
+ // Executor applies credentials via VersionedCredentials.updateIfNewer
+ VersionedCredentials.updateIfNewer(store, version, credentialBytes)
+
+ // Verify credentials were applied
+ val stored = store.get()
+ assert(stored != null, "Store should have credentials after TaskDescription apply")
+ assert(stored.version == 1L, "Stored version should be 1")
+
+ // Verify the stored bytes are deserializable and correct
+ val creds = UserCredentialManager.deserializeUserCredentials(stored.bytes)
+ assert(creds.forScheme("fake").isPresent,
+ "Executor store should contain valid 'fake' scheme credentials")
+
+ // Simulate RPC broadcast delivering v2 (fresher credentials)
+ val v2Bytes = Array[Byte](99, 98, 97)
Review Comment:
This is not a valid serialized `UserCredentials` (It cannot be deserialized). The test only exercises version comparison logic, which is fine, but a comment should clarify this:
```scala
// Dummy bytes. This test only verifies version guard logic, not credential deserialization.
val v2Bytes = Array[Byte](99, 98, 97)
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57896][CORE][TESTS] Add Kerberos coexistence and per-user token integration tests [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #57866:
URL: https://github.com/apache/spark/pull/57866#discussion_r3772567125
##########
core/src/test/scala/org/apache/spark/scheduler/TaskSetManagerSuite.scala:
##########
@@ -3245,6 +3245,50 @@ class TaskSetManagerSuite
}
}
+ test("resourceOffer attaches current userCredentials to TaskDescription") {
+ sc = new SparkContext("local", "test")
+ sched = new FakeTaskScheduler(sc, ("exec1", "host1"))
+ val taskSet = FakeTask.createTaskSet(3)
+ val clock = new ManualClock()
+ val manager = new TaskSetManager(sched, taskSet, MAX_TASK_FAILURES, clock = clock)
+
+ // Initially no credentials in SparkEnv store
+ assert(SparkEnv.get.userCredentials.get() == null)
+ val taskOpt1 = manager.resourceOffer("exec1", "host1", TaskLocality.ANY)._1
+ assert(taskOpt1.isDefined)
+ assert(taskOpt1.get.userCredentials.isEmpty,
+ "TaskDescription should have None when credential store is empty")
+
+ // Set credentials to version 1
+ val v1Bytes = Array[Byte](10, 20, 30, 40, 50)
+ VersionedCredentials.updateIfNewer(SparkEnv.get.userCredentials, 1L, v1Bytes)
+
+ // Offer another task from the same set -- should carry v1
+ val taskOpt2 = manager.resourceOffer("exec1", "host1", TaskLocality.ANY)._1
+ assert(taskOpt2.isDefined)
+ assert(taskOpt2.get.userCredentials.isDefined,
+ "TaskDescription should carry credentials after store is populated")
+ assert(taskOpt2.get.userCredentials.get._1 === 1L,
+ "TaskDescription should carry version 1")
+ assert(taskOpt2.get.userCredentials.get._2 === v1Bytes,
+ "TaskDescription should carry the v1 credential bytes")
+
+ // Update store to version 2
+ val v2Bytes = Array[Byte](50, 60, 70, 80, 90)
+ VersionedCredentials.updateIfNewer(SparkEnv.get.userCredentials, 2L, v2Bytes)
+
+ // Offer another task -- should carry v2
+ val taskOpt3 = manager.resourceOffer("exec1", "host1", TaskLocality.ANY)._1
+ assert(taskOpt3.isDefined)
+ assert(taskOpt3.get.userCredentials.get._1 === 2L,
+ "After renewal, TaskDescription should carry version 2")
+ assert(taskOpt3.get.userCredentials.get._2 === v2Bytes,
+ "TaskDescription should carry the v2 credential bytes")
+
+ // Cleanup: reset credential store
+ SparkEnv.get.userCredentials.set(null)
Review Comment:
`SparkEnv.get.userCredentials.set(null)` here is not protected by `try-finally`. If an assertion fails mid-test, the credential store remains polluted. Please wrap the test body:
```scala
try {
// test body
} finally {
SparkEnv.get.userCredentials.set(null)
}
```
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,595 @@
+/*
+ * 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 java.io.File
+import java.nio.file.Files
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicLong, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+import org.apache.spark.{SparkConf, SparkFunSuite, VersionedCredentials}
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ * 6. TaskDescription credentials are applied to the executor store with version guard
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ try {
+ if (tokenFile != null) tokenFile.delete()
+ } finally {
+ CredentialProviderLoader.resetForTesting()
+ super.afterEach()
+ }
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ Files.writeString(tokenFile.toPath, content)
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ val now = Instant.now()
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ now,
+ now.plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFreshExpiryIngestor(expiresInSeconds: Long): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] =
+ Optional.of(createUserContext(expiresInSeconds = expiresInSeconds))
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicLong(0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicLong(0L)
+
+ // Return a fresh UserContext on each load() so each renewal gets a genuinely
+ // new expiry rather than spinning on an already-expired token.
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (version, _) => {
+ latestVersion.set(version)
+ updateCount.incrementAndGet()
+ })
+
+ try {
+ manager.start()
+ assert(updateCount.get() == 1, "Should have exactly 1 update after start()")
+
+ eventually(timeout(15.seconds)) {
+ assert(updateCount.get() >= 2,
+ s"Expected at least 2 updates (got ${updateCount.get()}), " +
+ "indicating credential renewal occurred")
+ }
+
+ assert(latestVersion.get() >= 2L,
+ "Version should be at least 2 after renewal")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("per-user identity token produces valid credentials") {
+ val conf = createOidcConf()
+
+ val userCtx = createUserContext(principal = "alice@corp.example.com")
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(userCtx),
+ (_, bytes) => callbackRef.set(bytes))
+
+ try {
+ val (_, initialBytes) = manager.start()
+
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent,
+ "Per-user token should produce credentials for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Per-user credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Per-user credential should not be expired")
+
+ // Verify the credential is identical in structure to workload token output
+ val workloadCtx = createUserContext(principal = "workload-identity")
+ val workloadManager = new UserCredentialManager(
+ conf,
+ createIngestor(workloadCtx),
+ (_, _) => ())
+ try {
+ val (_, workloadBytes) = workloadManager.start()
+ val workloadCreds = UserCredentialManager.deserializeUserCredentials(workloadBytes)
+ val workloadFake = workloadCreds.forScheme("fake")
+ assert(workloadFake.isPresent)
+ assert(workloadFake.get().getProperties == fakeCred.get().getProperties,
+ "Per-user and workload tokens should produce identical credential properties")
+ } finally {
+ workloadManager.stop()
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("UserCredentialManager and HadoopDelegationTokenManager coexist") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+ val oidcVersion = new AtomicLong(0L)
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ oidcVersion.set(version)
+ oidcCallbackRef.set(bytes)
+ })
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L)
+ assert(oidcBytes != null)
+
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should produce tokens")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should contain 'fake' scheme")
+
+ // HadoopDelegationTokenManager.start() is synchronous -- verify directly
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should contain test.direct.credential")
+ assert(new String(dtCreds.getSecretKey(new Text("test.direct.credential"))) === "test-token",
+ "DT credential value should match")
+
+ assert(oidcVersion.get() == 1L, "OIDC version should remain at 1")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("OIDC failure does not affect HadoopDelegationTokenManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val failingOidcManager = new UserCredentialManager(
+ conf,
+ createFailingIngestor(),
+ (_, _) => ())
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // OIDC start should fail with IllegalStateException (missing token)
+ val oidcException = intercept[IllegalStateException] {
+ failingOidcManager.start()
+ }
+ assert(oidcException.getMessage.contains(
+ "identity token file is missing or malformed"))
+
+ // DT manager should still work perfectly despite OIDC failure
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should succeed despite OIDC failure")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should be unaffected by OIDC failure")
+ } finally {
+ failingOidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("DT provider failure does not affect UserCredentialManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+ .set("spark.security.credentials.test-direct.enabled", "false")
+ .set("spark.security.credentials.test-noexpiry.enabled", "false")
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => oidcCallbackRef.set(bytes))
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ val dtTokens = dtManager.start()
+ assert(dtTokens == null, "DT manager should return null when all providers fail")
+
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L, "OIDC version should be 1")
+ assert(oidcBytes != null, "OIDC should produce credentials despite DT failure")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should be unaffected by DT failure")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("deserialization idempotency preserves credential content") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ val bytes = serializedRef.get()
+ assert(bytes != null && bytes.length > 0, "Serialized credentials should be non-empty")
+
+ val creds1 = UserCredentialManager.deserializeUserCredentials(bytes)
+ val creds2 = UserCredentialManager.deserializeUserCredentials(bytes)
+
+ assert(creds1.forScheme("fake").isPresent)
+ assert(creds2.forScheme("fake").isPresent)
+ assert(creds1.forScheme("fake").get().getProperties ==
+ creds2.forScheme("fake").get().getProperties,
+ "Multiple deserializations of same bytes should produce identical credentials")
+
+ val cred = creds1.forScheme("fake").get()
+ assert(cred.getProperties.containsKey("provider"))
+ assert(cred.getExpiresAt != null, "Credential should have an expiry set")
+ assert(!cred.isExpired(Instant.now()), "Freshly resolved credential should not be expired")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("case-insensitive scheme lookup in credential bundle") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ val creds = UserCredentialManager.deserializeUserCredentials(serializedRef.get())
+
+ // FakeCredentialProvider declares supportedSchemes = Set("fake", "shared")
+ // but "shared" is ambiguous (AnotherFakeCredentialProvider also claims it),
+ // so only "fake" auto-resolves without explicit config.
+ assert(creds.forScheme("fake").isPresent, "Should resolve 'fake' scheme")
+ assert(creds.forScheme("FAKE").isPresent,
+ "Scheme lookup should be case-insensitive")
+ assert(creds.forScheme("Fake").isPresent,
+ "Scheme lookup should be case-insensitive")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("stop() after start() completes cleanly without exceptions") {
+ val conf = createOidcConf()
+ val ctx = createUserContext(expiresInSeconds = 60)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, _) => ())
+
+ manager.start()
+ manager.stop()
+
+ // Double stop should also be safe
+ manager.stop()
+ }
+
+ test("credential version is monotonically increasing across renewals") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val versions = new java.util.concurrent.CopyOnWriteArrayList[Long]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (version, _) => versions.add(version))
+
+ try {
+ manager.start()
+
+ eventually(timeout(15.seconds)) {
+ assert(versions.size() >= 3,
+ s"Expected at least 3 callbacks (got ${versions.size()})")
+ }
+
+ val versionList = new java.util.ArrayList(versions)
+ for (i <- 1 until versionList.size()) {
+ assert(versionList.get(i) > versionList.get(i - 1),
+ s"Version ${versionList.get(i)} should be > ${versionList.get(i - 1)} " +
+ s"at index $i (full list: $versionList)")
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("every renewal callback provides non-null non-empty credentials") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val allBytes = new java.util.concurrent.CopyOnWriteArrayList[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (_, bytes) => allBytes.add(bytes))
+
+ try {
+ manager.start()
+
+ eventually(timeout(15.seconds)) {
+ assert(allBytes.size() >= 2,
+ s"Expected at least 2 callbacks (got ${allBytes.size()})")
+ }
+
+ val it = allBytes.iterator()
+ while (it.hasNext) {
+ val bytes = it.next()
+ assert(bytes != null, "Callback bytes should never be null")
+ assert(bytes.length > 0, "Callback bytes should never be empty")
+ val creds = UserCredentialManager.deserializeUserCredentials(bytes)
+ assert(creds.forScheme("fake").isPresent,
+ "Every renewed credential bundle should contain 'fake' scheme")
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("OIDC disabled does not interfere with DT manager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, false)
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val oidcManager = UserCredentialManager.create(conf, (_, _) => ())
+ assert(oidcManager.isEmpty, "OIDC manager should not be created when disabled")
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+ try {
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should work when OIDC is disabled")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null)
+ } finally {
+ dtManager.stop()
+ }
+ }
+
+ test("TaskSetManager credential attachment reads from SparkEnv store") {
Review Comment:
The test `"TaskSetManager credential attachment reads from SparkEnv store"` actually verifies that credentials produced by `UserCredentialManager` round-trip correctly through a `VersionedCredentials` store (serialize -> store -> read -> deserialize). It does not use `TaskSetManager` or `SparkEnv`. How about renaming to `"credential serialization roundtrip through VersionedCredentials store"` or similar.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57896][CORE][TESTS] Add Kerberos coexistence and per-user token integration tests [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on PR #57866:
URL: https://github.com/apache/spark/pull/57866#issuecomment-5298914153
**Merge Summary:**
- merged into master https://github.com/apache/spark/commit/1c78b9124d5bcd2ce7c843143cf85177f30c75c3
- merged into branch-4.x https://github.com/apache/spark/commit/12b333c0425f01ecc25bcab5300481803d142839
*Posted by `merge_spark_pr.py`*
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org
Re: [PR] [SPARK-57896][CORE][TESTS] Add Kerberos coexistence and per-user token integration tests [spark]
Posted by "sarutak (via GitHub)" <gi...@apache.org>.
sarutak commented on code in PR #57866:
URL: https://github.com/apache/spark/pull/57866#discussion_r3762694024
##########
core/src/test/scala/org/apache/spark/deploy/security/OidcCredentialIntegrationSuite.scala:
##########
@@ -0,0 +1,582 @@
+/*
+ * 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 java.io.File
+import java.nio.file.Files
+import java.time.Instant
+import java.util.Optional
+import java.util.concurrent.atomic.{AtomicInteger, AtomicReference}
+
+import scala.concurrent.duration._
+
+import org.apache.hadoop.conf.Configuration
+import org.apache.hadoop.io.Text
+import org.mockito.ArgumentCaptor
+import org.mockito.Mockito.{mock, verify}
+import org.scalatest.concurrent.Eventually.{eventually, timeout}
+
+import org.apache.spark.{SparkConf, SparkFunSuite, VersionedCredentials}
+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._
+
+/**
+ * Integration tests for SPARK-57896: Kerberos coexistence and per-user token tests.
+ *
+ * Verifies that:
+ * 1. UserCredentialManager (OIDC) delivers credentials via the update callback
+ * 2. Credential refresh works end-to-end with expiring tokens
+ * 3. Per-user identity tokens produce valid credentials identically to workload tokens
+ * 4. Both UserCredentialManager and HadoopDelegationTokenManager can run simultaneously
+ * without interfering with each other
+ * 5. Failure in one credential system does not affect the other
+ * 6. TaskDescription credentials are applied to the executor store with version guard
+ */
+class OidcCredentialIntegrationSuite extends SparkFunSuite {
+
+ private var tokenFile: File = _
+
+ override def beforeEach(): Unit = {
+ super.beforeEach()
+ CredentialProviderLoader.resetForTesting()
+ tokenFile = File.createTempFile("oidc-token-", ".jwt")
+ tokenFile.deleteOnExit()
+ writeTokenFile("fake.jwt.token.workload")
+ }
+
+ override def afterEach(): Unit = {
+ try {
+ if (tokenFile != null) tokenFile.delete()
+ } finally {
+ CredentialProviderLoader.resetForTesting()
+ super.afterEach()
+ }
+ }
+
+ private def writeTokenFile(content: String): Unit = {
+ Files.writeString(tokenFile.toPath, content)
+ }
+
+ private def createOidcConf(): SparkConf = {
+ new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, true)
+ .set(SECURITY_OIDC_IDENTITY_TOKEN_FILE, tokenFile.getAbsolutePath)
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 5000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 1000L)
+ }
+
+ private def createUserContext(
+ principal: String = "test-user",
+ expiresInSeconds: Long = 300): UserContext = {
+ new UserContext(
+ principal,
+ "https://issuer.example.com",
+ "fake.jwt.token",
+ Instant.now(),
+ Instant.now().plusSeconds(expiresInSeconds))
+ }
+
+ private def createIngestor(ctx: UserContext): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.of(ctx)
+ }
+ }
+
+ private def createFreshExpiryIngestor(expiresInSeconds: Long): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] =
+ Optional.of(createUserContext(expiresInSeconds = expiresInSeconds))
+ }
+ }
+
+ private def createFailingIngestor(): TokenIngestor = {
+ new TokenIngestor {
+ override def load(): Optional[UserContext] = Optional.empty()
+ }
+ }
+
+ test("SPARK-57896: OIDC credential delivery via update callback") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val callbackRef = new AtomicReference[Array[Byte]]()
+ val callbackVersion = new AtomicReference[Long](0L)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ callbackVersion.set(version)
+ callbackRef.set(bytes)
+ })
+
+ try {
+ val (version, initialBytes) = manager.start()
+
+ assert(version == 1L, "Initial version should be 1")
+ assert(initialBytes != null, "Initial credentials should not be null")
+
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ assert(credentials != null, "Deserialized credentials should not be null")
+
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent, "Should have credential for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Credential should not be expired immediately after resolution")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: credential refresh works end-to-end on expiry") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val updateCount = new AtomicInteger(0)
+ val latestVersion = new AtomicReference[Long](0L)
+
+ // Return a fresh UserContext on each load() so each renewal gets a genuinely
+ // new expiry rather than spinning on an already-expired token.
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (version, _) => {
+ latestVersion.set(version)
+ updateCount.incrementAndGet()
+ })
+
+ try {
+ manager.start()
+ assert(updateCount.get() == 1, "Should have exactly 1 update after start()")
+
+ eventually(timeout(15.seconds)) {
+ assert(updateCount.get() >= 2,
+ s"Expected at least 2 updates (got ${updateCount.get()}), " +
+ "indicating credential renewal occurred")
+ }
+
+ assert(latestVersion.get() >= 2L,
+ "Version should be at least 2 after renewal")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: per-user identity token produces valid credentials") {
+ val conf = createOidcConf()
+
+ val userCtx = createUserContext(principal = "alice@corp.example.com")
+ val callbackRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(userCtx),
+ (_, bytes) => callbackRef.set(bytes))
+
+ try {
+ val (_, initialBytes) = manager.start()
+
+ val credentials = UserCredentialManager.deserializeUserCredentials(initialBytes)
+ val fakeCred = credentials.forScheme("fake")
+ assert(fakeCred.isPresent,
+ "Per-user token should produce credentials for scheme 'fake'")
+ assert(fakeCred.get().getProperties.get("provider") == "fake",
+ "Per-user credential should come from FakeCredentialProvider")
+ assert(!fakeCred.get().isExpired(Instant.now()),
+ "Per-user credential should not be expired")
+
+ // Verify the credential is identical in structure to workload token output
+ val workloadCtx = createUserContext(principal = "workload-identity")
+ val workloadManager = new UserCredentialManager(
+ conf,
+ createIngestor(workloadCtx),
+ (_, _) => ())
+ try {
+ val (_, workloadBytes) = workloadManager.start()
+ val workloadCreds = UserCredentialManager.deserializeUserCredentials(workloadBytes)
+ val workloadFake = workloadCreds.forScheme("fake")
+ assert(workloadFake.isPresent)
+ assert(workloadFake.get().getProperties == fakeCred.get().getProperties,
+ "Per-user and workload tokens should produce identical credential properties")
+ } finally {
+ workloadManager.stop()
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: UserCredentialManager and HadoopDelegationTokenManager coexist") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+ val oidcVersion = new AtomicReference[Long](0L)
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (version, bytes) => {
+ oidcVersion.set(version)
+ oidcCallbackRef.set(bytes)
+ })
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L)
+ assert(oidcBytes != null)
+
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should produce tokens")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should contain 'fake' scheme")
+
+ // HadoopDelegationTokenManager.start() is synchronous -- verify directly
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should contain test.direct.credential")
+ assert(new String(dtCreds.getSecretKey(new Text("test.direct.credential"))) === "test-token",
+ "DT credential value should match")
+
+ assert(oidcVersion.get() == 1L, "OIDC version should remain at 1")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC failure does not affect HadoopDelegationTokenManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val failingOidcManager = new UserCredentialManager(
+ conf,
+ createFailingIngestor(),
+ (_, _) => ())
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ // OIDC start should fail with IllegalStateException (missing token)
+ val oidcException = intercept[IllegalStateException] {
+ failingOidcManager.start()
+ }
+ assert(oidcException.getMessage.contains(
+ "identity token file is missing or malformed"))
+
+ // DT manager should still work perfectly despite OIDC failure
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should succeed despite OIDC failure")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null,
+ "DT credentials should be unaffected by OIDC failure")
+ } finally {
+ failingOidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: DT provider failure does not affect UserCredentialManager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = createOidcConf()
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+ .set("spark.security.credentials.test-direct.enabled", "false")
+ .set("spark.security.credentials.test-noexpiry.enabled", "false")
+
+ val ctx = createUserContext()
+ val oidcCallbackRef = new AtomicReference[Array[Byte]]()
+
+ val oidcManager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => oidcCallbackRef.set(bytes))
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+
+ try {
+ val dtTokens = dtManager.start()
+ assert(dtTokens == null, "DT manager should return null when all providers fail")
+
+ val (oidcVer, oidcBytes) = oidcManager.start()
+ assert(oidcVer == 1L, "OIDC version should be 1")
+ assert(oidcBytes != null, "OIDC should produce credentials despite DT failure")
+
+ val oidcCreds = UserCredentialManager.deserializeUserCredentials(oidcBytes)
+ assert(oidcCreds.forScheme("fake").isPresent,
+ "OIDC credentials should be unaffected by DT failure")
+ } finally {
+ oidcManager.stop()
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: deserialization idempotency preserves credential content") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ val bytes = serializedRef.get()
+ assert(bytes != null && bytes.length > 0, "Serialized credentials should be non-empty")
+
+ val creds1 = UserCredentialManager.deserializeUserCredentials(bytes)
+ val creds2 = UserCredentialManager.deserializeUserCredentials(bytes)
+
+ assert(creds1.forScheme("fake").isPresent)
+ assert(creds2.forScheme("fake").isPresent)
+ assert(creds1.forScheme("fake").get().getProperties ==
+ creds2.forScheme("fake").get().getProperties,
+ "Multiple deserializations of same bytes should produce identical credentials")
+
+ val cred = creds1.forScheme("fake").get()
+ assert(cred.getProperties.containsKey("provider"))
+ assert(cred.getExpiresAt != null, "Credential should have an expiry set")
+ assert(!cred.isExpired(Instant.now()), "Freshly resolved credential should not be expired")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: case-insensitive scheme lookup in credential bundle") {
+ val conf = createOidcConf()
+ val ctx = createUserContext()
+ val serializedRef = new AtomicReference[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, bytes) => serializedRef.set(bytes))
+
+ try {
+ manager.start()
+
+ val creds = UserCredentialManager.deserializeUserCredentials(serializedRef.get())
+
+ // FakeCredentialProvider declares supportedSchemes = Set("fake", "shared")
+ // but "shared" is ambiguous (AnotherFakeCredentialProvider also claims it),
+ // so only "fake" auto-resolves without explicit config.
+ assert(creds.forScheme("fake").isPresent, "Should resolve 'fake' scheme")
+ assert(creds.forScheme("FAKE").isPresent,
+ "Scheme lookup should be case-insensitive")
+ assert(creds.forScheme("Fake").isPresent,
+ "Scheme lookup should be case-insensitive")
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: stop() after start() completes cleanly without exceptions") {
+ val conf = createOidcConf()
+ val ctx = createUserContext(expiresInSeconds = 60)
+
+ val manager = new UserCredentialManager(
+ conf,
+ createIngestor(ctx),
+ (_, _) => ())
+
+ manager.start()
+ manager.stop()
+
+ // Double stop should also be safe
+ manager.stop()
+ }
+
+ test("SPARK-57896: credential version is monotonically increasing across renewals") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val versions = new java.util.concurrent.CopyOnWriteArrayList[Long]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (version, _) => versions.add(version))
+
+ try {
+ manager.start()
+
+ eventually(timeout(15.seconds)) {
+ assert(versions.size() >= 3,
+ s"Expected at least 3 callbacks (got ${versions.size()})")
+ }
+
+ val versionList = new java.util.ArrayList(versions)
+ for (i <- 1 until versionList.size()) {
+ assert(versionList.get(i) > versionList.get(i - 1),
+ s"Version ${versionList.get(i)} should be > ${versionList.get(i - 1)} " +
+ s"at index $i (full list: $versionList)")
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: every renewal callback provides non-null non-empty credentials") {
+ val conf = createOidcConf()
+ .set(SECURITY_OIDC_RENEWAL_SAFETY_MARGIN, 2000L)
+ .set(SECURITY_OIDC_RENEWAL_MIN_INTERVAL, 500L)
+
+ val allBytes = new java.util.concurrent.CopyOnWriteArrayList[Array[Byte]]()
+
+ val manager = new UserCredentialManager(
+ conf,
+ createFreshExpiryIngestor(expiresInSeconds = 3),
+ (_, bytes) => allBytes.add(bytes))
+
+ try {
+ manager.start()
+
+ eventually(timeout(15.seconds)) {
+ assert(allBytes.size() >= 2,
+ s"Expected at least 2 callbacks (got ${allBytes.size()})")
+ }
+
+ val it = allBytes.iterator()
+ while (it.hasNext) {
+ val bytes = it.next()
+ assert(bytes != null, "Callback bytes should never be null")
+ assert(bytes.length > 0, "Callback bytes should never be empty")
+ val creds = UserCredentialManager.deserializeUserCredentials(bytes)
+ assert(creds.forScheme("fake").isPresent,
+ "Every renewed credential bundle should contain 'fake' scheme")
+ }
+ } finally {
+ manager.stop()
+ }
+ }
+
+ test("SPARK-57896: OIDC disabled does not interfere with DT manager") {
+ val hadoopConf = new Configuration()
+ val mockRef = mock(classOf[RpcEndpointRef])
+
+ val conf = new SparkConf(loadDefaults = false)
+ .set(SECURITY_OIDC_ENABLED, false)
+ .set(DIRECT_CREDENTIAL_PROVIDERS_ENABLED, true)
+ .set(NETWORK_AUTH_ENABLED, true)
+ .set(NETWORK_CRYPTO_ENABLED, true)
+
+ val oidcManager = UserCredentialManager.create(conf, (_, _) => ())
+ assert(oidcManager.isEmpty, "OIDC manager should not be created when disabled")
+
+ val dtManager = new HadoopDelegationTokenManager(conf, hadoopConf, mockRef)
+ try {
+ val dtTokens = dtManager.start()
+ assert(dtTokens != null, "DT manager should work when OIDC is disabled")
+
+ val captor = ArgumentCaptor.forClass(classOf[Any])
+ verify(mockRef).send(captor.capture())
+ val msg = captor.getValue.asInstanceOf[UpdateDelegationTokens]
+ val dtCreds = SparkHadoopUtil.get.deserialize(msg.tokens)
+ assert(dtCreds.getSecretKey(new Text("test.direct.credential")) != null)
+ } finally {
+ dtManager.stop()
+ }
+ }
+
+ test("SPARK-57896: TaskDescription credentials applied to executor store with version guard") {
Review Comment:
This test calls `VersionedCredentials.updateIfNewer` directly. It does not exercise `TaskRunner`, `CoarseGrainedExecutorBackend.receive`, or any actual executor code path. It is a unit test of the `updateIfNewer` helper method.
**Suggestions:**
- The test name's intent ("TaskDescription credentials applied to executor store with version guard") is correct, but the test content does not fulfill it — it only calls `VersionedCredentials.updateIfNewer` directly without going through the actual `TaskDescription` → Executor path. Please update the test to exercise the real code path, or add a separate test that does.
- **Background:** In [PR #57675](https://github.com/apache/spark/pull/57675) (Sub-task 5), dongjoon-hyun and HyukjinKwon pointed out that `TaskSetManager.resourceOffer` attaching credentials to `TaskDescription` was not tested. I responded that it would be covered in Sub-task 7 (this PR). Please add a test that covers this path.
**Expected test outline:**
- This can be implemented in `TaskSetManagerSuite` (which already has the necessary infrastructure: `FakeSchedulerBackend`, `TaskSchedulerImpl`, and calls to `manager.resourceOffer(...)` returning `TaskDescription`)
- Set `SparkEnv.get.userCredentials` to `VersionedCredentials(version=1, bytes=<valid serialized UserCredentials>)`
- Call `manager.resourceOffer(...)` and assert that the returned `TaskDescription.userCredentials == Some((1L, bytes))`
- Update the store to version 2, call `resourceOffer` again, and verify the new `TaskDescription` carries version 2
- This proves that `TaskSetManager` reads the current credential store at dispatch time (line 607 of `TaskSetManager.scala`: `Option(env.userCredentials.get()).map(vc => (vc.version, vc.bytes))`) and includes it in every `TaskDescription`
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: reviews-unsubscribe@spark.apache.org
For additional commands, e-mail: reviews-help@spark.apache.org