This is an automated email from the ASF dual-hosted git repository.
HeartSaVioR pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 52eb078867cf [SPARK-28173][SS] Support obtaining Kafka delegation
token for proxy user
52eb078867cf is described below
commit 52eb078867cf809a92fd6ebdc96c9d72bce63417
Author: Cheng Pan <[email protected]>
AuthorDate: Thu Aug 20 16:26:38 2026 +0900
[SPARK-28173][SS] Support obtaining Kafka delegation token for proxy user
### What changes were proposed in this pull request?
Support obtaining Kafka delegation tokens when the current user is a proxy
user, based on
KAFKA-6945 (Kafka 3.3.0):
- `KafkaTokenUtil.obtainToken` requests the token with
`CreateDelegationTokenOptions.owner(<proxy user>)` when the current UGI
is a proxy user, so the
token is requested with the real user's credentials but owned by the
proxy user. The
`checkProxyUser()` guard that rejected proxy users is removed.
- The two failure modes of the owner-set path are translated into
actionable `SparkException`s:
`UnsupportedVersionException` (broker older than 3.3.0) and
`DelegationTokenAuthorizationException` (real user lacks the
`CreateTokens` operation on the
`User:<proxy user>` resource).
- `printToken` logs the token's REQUESTER alongside OWNER, which differ for
proxy-user tokens.
- The SS Kafka integration guide's caveat is updated: broker version
requirement, the
`kafka-acls.sh` grant example, and the constraint that the real user's
credentials come from the
ticket cache (`--proxy-user` cannot be combined with
`--principal`/`--keytab`) with the token
obtained once and not renewed.
### Why are the changes needed?
Applications submitted with `--proxy-user` cannot use Kafka delegation
tokens: token acquisition
is skipped with a warning, and connectors fall back to other
authentication. KAFKA-6945 added the
ability to create tokens owned by another principal, which makes
impersonation work end to end.
### Does this PR introduce _any_ user-facing change?
Yes. Proxy-user applications now obtain Kafka delegation tokens owned by
the proxy user
(previously acquisition was skipped with a warning). Requires Kafka 3.3.0+
brokers and the real
user granted `CreateTokens` on the `User:<proxy user>` resource;
misconfigurations of either now
fail with a descriptive error.
### How was this patch tested?
UT/IT:
- `KafkaTokenUtilSuite`: `createDelegationTokenOptions` sets the owner for
a proxy user and
leaves it unset otherwise.
- `KafkaDelegationTokenSuite` (embedded secure Kafka + MiniKdc): asserts
the broker-recorded
owner/requester for both the plain and the proxy-user roundtrip; the
proxy roundtrip runs with
DENY read/write ACLs for the real user on the test topic so it passes
only if connectors
authenticate as the proxy user; a negative test asserts the descriptive
error when
`CreateTokens` is not granted.
Real cluster tests:
- Plain token via `KafkaTokenUtil.obtainToken` — → success (token owner
`User:spark`, requester `User:spark`).
- Proxy-user token - `alice` with CreateTokens ACL → success (token owner
`User:alice`).
- Proxy-user token - `bob` without ACL → failure, clean `SparkException:
The real user must be granted the CreateTokens operation on the User:bob
resource.`; broker recorded 0 tokens for `bob`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-fable-5)
Closes #58064 from pan3793/SPARK-28173.
Authored-by: Cheng Pan <[email protected]>
Signed-off-by: Jungtaek Lim <[email protected]>
---
.../sql/kafka010/KafkaDelegationTokenSuite.scala | 130 +++++++++++++++++++--
.../apache/spark/sql/kafka010/KafkaTestUtils.scala | 94 ++++++++++++---
.../org/apache/spark/kafka010/KafkaTokenUtil.scala | 55 ++++++---
.../spark/kafka010/KafkaTokenUtilSuite.scala | 21 +++-
.../structured-streaming-kafka-integration.md | 13 ++-
5 files changed, 272 insertions(+), 41 deletions(-)
diff --git
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDelegationTokenSuite.scala
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDelegationTokenSuite.scala
index a55b6e006851..5d6356cb0dad 100644
---
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDelegationTokenSuite.scala
+++
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaDelegationTokenSuite.scala
@@ -17,15 +17,23 @@
package org.apache.spark.sql.kafka010
+import java.security.PrivilegedExceptionAction
import java.util.UUID
+import java.util.concurrent.ExecutionException
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.security.{Credentials, UserGroupInformation}
+import org.apache.kafka.common.acl.{AccessControlEntry, AclBinding,
AclOperation, AclPermissionType}
+import org.apache.kafka.common.errors.DelegationTokenAuthorizationException
+import org.apache.kafka.common.resource.{PatternType, ResourcePattern,
ResourceType}
import org.apache.kafka.common.security.auth.SecurityProtocol.SASL_PLAINTEXT
+import org.apache.kafka.common.security.token.delegation.TokenInformation
+import org.apache.spark.SparkException
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.deploy.security.HadoopDelegationTokenManager
import org.apache.spark.internal.config.{KEYTAB, PRINCIPAL}
+import org.apache.spark.kafka010.{KafkaTokenSparkConf, KafkaTokenUtil}
import org.apache.spark.sql.execution.streaming.runtime.MemoryStream
import org.apache.spark.sql.streaming.{OutputMode, StreamTest}
import org.apache.spark.sql.test.SharedSparkSession
@@ -34,6 +42,13 @@ class KafkaDelegationTokenSuite extends StreamTest with
SharedSparkSession with
import testImplicits._
+ private val clusterIdentifier = "cluster1"
+
+ // The client principal is allowed to obtain tokens for `proxyUser` but not
for
+ // `deniedProxyUser`.
+ private val proxyUser = "proxyUser"
+ private val deniedProxyUser = "deniedProxyUser"
+
protected var testUtils: KafkaTestUtils = _
protected override def sparkConf = super.sparkConf
@@ -41,12 +56,26 @@ class KafkaDelegationTokenSuite extends StreamTest with
SharedSparkSession with
.set("spark.security.credentials.hbase.enabled", "false")
.set(KEYTAB, testUtils.clientKeytab)
.set(PRINCIPAL, testUtils.clientPrincipal)
- .set("spark.kafka.clusters.cluster1.auth.bootstrap.servers",
testUtils.brokerAddress)
- .set("spark.kafka.clusters.cluster1.security.protocol",
SASL_PLAINTEXT.name)
+ .set(s"spark.kafka.clusters.$clusterIdentifier.auth.bootstrap.servers",
testUtils.brokerAddress)
+ .set(s"spark.kafka.clusters.$clusterIdentifier.security.protocol",
SASL_PLAINTEXT.name)
override def beforeAll(): Unit = {
testUtils = new KafkaTestUtils(Map.empty, true)
- testUtils.setup()
+ try {
+ testUtils.setup(
+ testUtils.allowAllAcls(testUtils.kafkaPrincipal(proxyUser).toString) :+
+ createTokensAcl(proxyUser))
+ } catch {
+ case e: Throwable =>
+ // ScalaTest skips afterAll when beforeAll throws, so tear down here
to avoid leaking
+ // the KDC, broker, and the global JAAS system property into later
suites.
+ try {
+ testUtils.teardown()
+ } finally {
+ testUtils = null
+ }
+ throw e
+ }
super.beforeAll()
}
@@ -62,17 +91,53 @@ class KafkaDelegationTokenSuite extends StreamTest with
SharedSparkSession with
}
}
- testRetry("Roundtrip", 3) {
- val hadoopConf = new Configuration()
- val manager = new HadoopDelegationTokenManager(spark.sparkContext.conf,
hadoopConf, null)
+ /**
+ * Allow the client principal to obtain tokens owned by `owner`. The broker
looks the ACL up
+ * by the `KafkaPrincipal.toString` of the owner, hence the `User:` prefixed
resource name.
+ */
+ private def createTokensAcl(owner: String): AclBinding = new AclBinding(
+ new ResourcePattern(
+ ResourceType.USER, testUtils.kafkaPrincipal(owner).toString,
PatternType.LITERAL),
+ new AccessControlEntry(
+ testUtils.clientKafkaPrincipal, "*", AclOperation.CREATE_TOKENS,
AclPermissionType.ALLOW))
+
+ private def proxyUgi(user: String): UserGroupInformation =
+ UserGroupInformation.createProxyUser(user,
UserGroupInformation.getCurrentUser())
+
+ private def obtainDelegationTokens(ugi: UserGroupInformation): Credentials =
{
val credentials = new Credentials()
- manager.obtainDelegationTokens(credentials)
+ ugi.doAs(new PrivilegedExceptionAction[Unit]() {
+ override def run(): Unit = {
+ val manager = new HadoopDelegationTokenManager(
+ spark.sparkContext.conf, new Configuration(), null)
+ manager.obtainDelegationTokens(credentials)
+ }
+ })
+ credentials
+ }
+
+ /** Ask the broker what it recorded for the token which was just obtained. */
+ private def describeToken(credentials: Credentials): TokenInformation = {
+ val token =
credentials.getToken(KafkaTokenUtil.getTokenService(clusterIdentifier))
+ assert(token != null, s"No delegation token was obtained for cluster
$clusterIdentifier")
+ val tokenId = new String(token.getIdentifier)
+ testUtils.describeDelegationTokens().find(_.tokenId() == tokenId)
+ .getOrElse(fail(s"Token $tokenId is unknown to the broker"))
+ }
+
+ private def distributeTokens(credentials: Credentials): Unit = {
val serializedCredentials = SparkHadoopUtil.get.serialize(credentials)
SparkHadoopUtil.get.addDelegationTokens(serializedCredentials,
spark.sparkContext.conf)
+ }
+ private def createTopic(): String = {
val topic = "topic-" + UUID.randomUUID().toString
testUtils.createTopic(topic, partitions = 5)
+ topic
+ }
+ /** Write to and read back from `topic`, authenticating with the distributed
token. */
+ private def roundtrip(topic: String): Unit = {
withTempDir { checkpointDir =>
val input = MemoryStream[String]
@@ -115,4 +180,55 @@ class KafkaDelegationTokenSuite extends StreamTest with
SharedSparkSession with
StopStream
)
}
+
+ testRetry("Roundtrip", 3) {
+ val credentials =
obtainDelegationTokens(UserGroupInformation.getCurrentUser())
+
+ // Without impersonation no owner is sent, so the broker assigns ownership
to the requester.
+ val tokenInfo = describeToken(credentials)
+ assert(tokenInfo.owner().toString === testUtils.clientKafkaPrincipal)
+ assert(tokenInfo.tokenRequester().toString ===
testUtils.clientKafkaPrincipal)
+
+ distributeTokens(credentials)
+ roundtrip(createTopic())
+ }
+
+ testRetry("SPARK-28173: Roundtrip with proxy user", 3) {
+ // The manager preserves the proxy UGI here despite KEYTAB/PRINCIPAL being
set only because
+ // Hadoop security is SIMPLE in this JVM (a keytab re-login is then a
no-op). The production
+ // proxy flow (ticket cache or direct providers, no principal) preserves
it by design.
+ val credentials = obtainDelegationTokens(proxyUgi(proxyUser))
+
+ // The token is requested with the client's credentials but owned by the
proxy user, so
+ // connectors authenticate to Kafka as the proxy user.
+ val tokenInfo = describeToken(credentials)
+ assert(tokenInfo.owner() === testUtils.kafkaPrincipal(proxyUser))
+ assert(tokenInfo.tokenRequester().toString ===
testUtils.clientKafkaPrincipal)
+
+ distributeTokens(credentials)
+ val topic = createTopic()
+ // Deny the client principal on the topic so the roundtrip only succeeds
when the connectors
+ // authenticate as the proxy user (deny overrides the wildcard allow).
+ testUtils.createAcls(Seq(AclOperation.READ, AclOperation.WRITE).map { op =>
+ new AclBinding(
+ new ResourcePattern(ResourceType.TOPIC, topic, PatternType.LITERAL),
+ new AccessControlEntry(testUtils.clientKafkaPrincipal, "*", op,
AclPermissionType.DENY))
+ })
+ roundtrip(topic)
+ }
+
+ test("SPARK-28173: Obtaining token for proxy user without CreateTokens
permission fails") {
+ val conf = spark.sparkContext.conf
+ val clusterConf = KafkaTokenSparkConf.getClusterConfig(conf,
clusterIdentifier)
+ proxyUgi(deniedProxyUser).doAs(new PrivilegedExceptionAction[Unit]() {
+ override def run(): Unit = {
+ val e = intercept[SparkException] {
+ KafkaTokenUtil.obtainToken(conf, clusterConf)
+ }
+ assert(e.getMessage.contains("CreateTokens"))
+ assert(e.getCause.asInstanceOf[ExecutionException]
+ .getCause.isInstanceOf[DelegationTokenAuthorizationException])
+ }
+ })
+ }
}
diff --git
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala
index b31f6af1c794..14ba16667ac8 100644
---
a/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala
+++
b/connector/kafka-0-10-sql/src/test/scala/org/apache/spark/sql/kafka010/KafkaTestUtils.scala
@@ -28,6 +28,7 @@ import scala.io.Source
import scala.jdk.CollectionConverters._
import kafka.log.LogManager
+import kafka.security.authorizer.AclAuthorizer
import kafka.server.{HostedPartition, KafkaConfig, KafkaServer}
import kafka.server.checkpoints.OffsetCheckpointFile
import kafka.zk.KafkaZkClient
@@ -37,10 +38,14 @@ import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.clients.admin._
import org.apache.kafka.clients.producer._
import org.apache.kafka.common.TopicPartition
+import org.apache.kafka.common.acl.{AccessControlEntry, AclBinding,
AclOperation, AclPermissionType}
import org.apache.kafka.common.config.SaslConfigs
import org.apache.kafka.common.network.ListenerName
import org.apache.kafka.common.requests.FetchRequest
+import org.apache.kafka.common.resource.{PatternType, Resource,
ResourcePattern, ResourceType}
+import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.security.auth.SecurityProtocol.{PLAINTEXT,
SASL_PLAINTEXT}
+import org.apache.kafka.common.security.token.delegation.TokenInformation
import org.apache.kafka.common.serialization.StringSerializer
import org.apache.kafka.common.utils.Time
import org.apache.zookeeper.client.ZKClientConfig
@@ -95,7 +100,10 @@ class KafkaTestUtils(
private var brokerConf: KafkaConfig = _
private val brokerServiceName = "kafka"
- private val clientUser = s"client/$localCanonicalHostName"
+ private val brokerUser = s"$brokerServiceName/$localCanonicalHostName"
+ private var brokerKeytabFile: File = _
+ private val clientShortName = "client"
+ private val clientUser = s"$clientShortName/$localCanonicalHostName"
private var clientKeytabFile: File = _
// Kafka broker server
@@ -137,6 +145,16 @@ class KafkaTestUtils(
clientKeytabFile.getAbsolutePath()
}
+ /** The Kafka principal for `user`, i.e. `User:<user>`. */
+ def kafkaPrincipal(user: String): KafkaPrincipal =
+ new KafkaPrincipal(KafkaPrincipal.USER_TYPE, user)
+
+ /**
+ * The Kafka principal the client keytab authenticates as. The broker
derives it from
+ * `client/<host>@<realm>` with the default
`sasl.kerberos.principal.to.local.rules`.
+ */
+ def clientKafkaPrincipal: String = kafkaPrincipal(clientShortName).toString
+
private def setUpMiniKdc(): Unit = {
val kdcDir = Utils.createTempDir()
val kdcConf = MiniKdc.createConf()
@@ -197,10 +215,9 @@ class KafkaTestUtils(
kdc.createPrincipal(zkClientKeytabFile, zkClientUser)
logDebug(s"Created keytab file: ${zkClientKeytabFile.getAbsolutePath()}")
- val kafkaServerUser = s"kafka/$localCanonicalHostName"
- val kafkaServerKeytabFile = new File(baseDir, "kafka.keytab")
- kdc.createPrincipal(kafkaServerKeytabFile, kafkaServerUser)
- logDebug(s"Created keytab file:
${kafkaServerKeytabFile.getAbsolutePath()}")
+ brokerKeytabFile = new File(baseDir, "kafka.keytab")
+ kdc.createPrincipal(brokerKeytabFile, brokerUser)
+ logDebug(s"Created keytab file: ${brokerKeytabFile.getAbsolutePath()}")
clientKeytabFile = new File(baseDir, "client.keytab")
kdc.createPrincipal(clientKeytabFile, clientUser)
@@ -235,8 +252,8 @@ class KafkaTestUtils(
| serviceName="$brokerServiceName"
| useKeyTab=true
| storeKey=true
- | keyTab="${kafkaServerKeytabFile.getAbsolutePath()}"
- | principal="$kafkaServerUser@$realm";
+ | keyTab="${brokerKeytabFile.getAbsolutePath()}"
+ | principal="$brokerUser@$realm";
|};
""".stripMargin.trim
Files.writeString(file.toPath, content)
@@ -276,8 +293,12 @@ class KafkaTestUtils(
brokerReady = true
}
- /** setup the whole embedded servers, including Zookeeper and Kafka brokers
*/
- def setup(): Unit = {
+ /**
+ * Setup the whole embedded servers, including Zookeeper and Kafka brokers.
In secure mode,
+ * `extraAcls` are created together with the client principal's allow-all
ACLs.
+ */
+ def setup(extraAcls: Seq[AclBinding] = Nil): Unit = {
+ assert(secure || extraAcls.isEmpty, "ACLs require the cluster to be set up
in secure mode")
// Set up a KafkaTestUtils leak detector so that we can see where the leak
KafkaTestUtils is
// created.
val exception = new SparkException("It was created at: ")
@@ -299,8 +320,45 @@ class KafkaTestUtils(
eventually(timeout(1.minute)) {
assert(zkClient.getAllBrokersInCluster.nonEmpty, "Broker was not up in
60 seconds")
}
+ if (secure) {
+ createAcls(allowAllAcls(clientKafkaPrincipal) ++ extraAcls)
+ }
}
+ /** All the ACLs granting `principal` unrestricted access to the cluster. */
+ def allowAllAcls(principal: String): Seq[AclBinding] = {
+ val entry = new AccessControlEntry(principal, "*", AclOperation.ALL,
AclPermissionType.ALLOW)
+ def wildcard(resourceType: ResourceType): ResourcePattern =
+ new ResourcePattern(resourceType, ResourcePattern.WILDCARD_RESOURCE,
PatternType.LITERAL)
+ Seq(
+ wildcard(ResourceType.TOPIC),
+ wildcard(ResourceType.GROUP),
+ wildcard(ResourceType.TRANSACTIONAL_ID),
+ new ResourcePattern(ResourceType.CLUSTER, Resource.CLUSTER_NAME,
PatternType.LITERAL)
+ ).map(new AclBinding(_, entry))
+ }
+
+ /**
+ * Add ACLs as a super user. The single zk-mode broker updates its
authorizer cache before
+ * completing the request, so the ACLs are enforced once this returns.
+ */
+ def createAcls(bindings: Seq[AclBinding]): Unit = {
+ assert(secure, "ACLs are only enforced when the cluster is set up in
secure mode")
+ val props = new Properties()
+ props.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerAddress)
+ addSaslClientProps(props, brokerKeytabFile,
s"$brokerUser@${kdc.getRealm()}")
+ val superuserAdminClient = AdminClient.create(props)
+ try {
+ superuserAdminClient.createAcls(bindings.asJava).all().get()
+ } finally {
+ superuserAdminClient.close()
+ }
+ }
+
+ /** All delegation tokens the broker knows, as seen by the client principal.
*/
+ def describeDelegationTokens(): Seq[TokenInformation] =
+
adminClient.describeDelegationToken().delegationTokens().get().asScala.toSeq.map(_.tokenInfo())
+
/** Teardown the whole servers, including Kafka broker and Zookeeper */
def teardown(): Unit = {
if (leakDetector != null) {
@@ -501,6 +559,13 @@ class KafkaTestUtils(
props.put("inter.broker.listener.name", "SASL_PLAINTEXT")
props.put("delegation.token.master.key", UUID.randomUUID().toString)
props.put("sasl.enabled.mechanisms", "GSSAPI,SCRAM-SHA-512")
+ // Enforce ACLs so that delegation token authorization can be tested.
The broker
+ // authenticates to itself as `kafka/<host>@<realm>`, which the default
+ // principal-to-local rules map to `User:kafka`, and it must stay
unrestricted.
+ // `setup` grants the client principal full access, so tests which do
not care about
+ // ACLs are unaffected.
+ props.put("authorizer.class.name", classOf[AclAuthorizer].getName)
+ props.put("super.users", kafkaPrincipal(brokerServiceName).toString)
}
props.putAll(withBrokerProps.asJava)
@@ -542,13 +607,16 @@ class KafkaTestUtils(
private def setAuthenticationConfigIfNeeded(props: Properties): Unit = {
if (secure) {
- val jaasParams = KafkaTokenUtil.getKeytabJaasParams(
- clientKeytabFile.getAbsolutePath, clientPrincipal, brokerServiceName)
- props.put(SaslConfigs.SASL_JAAS_CONFIG, jaasParams)
- props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG,
SASL_PLAINTEXT.name)
+ addSaslClientProps(props, clientKeytabFile, clientPrincipal)
}
}
+ private def addSaslClientProps(props: Properties, keytabFile: File,
principal: String): Unit = {
+ props.put(SaslConfigs.SASL_JAAS_CONFIG, KafkaTokenUtil.getKeytabJaasParams(
+ keytabFile.getAbsolutePath, principal, brokerServiceName))
+ props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG,
SASL_PLAINTEXT.name)
+ }
+
/** Verify topic is deleted in all places, e.g, brokers, zookeeper. */
private def verifyTopicDeletion(
topic: String,
diff --git
a/connector/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaTokenUtil.scala
b/connector/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaTokenUtil.scala
index cbcc27ff7d28..9a2325cce3b2 100644
---
a/connector/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaTokenUtil.scala
+++
b/connector/kafka-0-10-token-provider/src/main/scala/org/apache/spark/kafka010/KafkaTokenUtil.scala
@@ -20,6 +20,7 @@ package org.apache.spark.kafka010
import java.{util => ju}
import java.time.{Instant, ZoneId}
import java.time.format.DateTimeFormatter
+import java.util.concurrent.ExecutionException
import java.util.regex.Pattern
import scala.jdk.CollectionConverters._
@@ -32,14 +33,17 @@ import
org.apache.hadoop.security.token.delegation.AbstractDelegationTokenIdenti
import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.clients.admin.{AdminClient,
CreateDelegationTokenOptions}
import org.apache.kafka.common.config.{SaslConfigs, SslConfigs}
+import org.apache.kafka.common.errors.{DelegationTokenAuthorizationException,
UnsupportedVersionException}
import org.apache.kafka.common.security.JaasContext
+import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.security.auth.SecurityProtocol.{SASL_PLAINTEXT,
SASL_SSL, SSL}
import org.apache.kafka.common.security.scram.ScramLoginModule
import org.apache.kafka.common.security.token.delegation.DelegationToken
-import org.apache.spark.SparkConf
+import org.apache.spark.{SparkConf, SparkException}
import org.apache.spark.deploy.SparkHadoopUtil
import org.apache.spark.internal.Logging
+import org.apache.spark.internal.LogKeys.USER_NAME
import org.apache.spark.internal.config._
import org.apache.spark.util.{SecurityUtils, Utils}
import org.apache.spark.util.Utils.REDACTION_REPLACEMENT_TEXT
@@ -65,13 +69,21 @@ object KafkaTokenUtil extends Logging {
def obtainToken(
sparkConf: SparkConf,
clusterConf: KafkaTokenClusterConf):
(Token[KafkaDelegationTokenIdentifier], Long) = {
- checkProxyUser()
-
+ val options = createDelegationTokenOptions()
val adminClient =
AdminClient.create(createAdminClientProperties(sparkConf, clusterConf))
val token = try {
- val createDelegationTokenOptions = new CreateDelegationTokenOptions()
- val createResult =
adminClient.createDelegationToken(createDelegationTokenOptions)
- createResult.delegationToken().get()
+ adminClient.createDelegationToken(options).delegationToken().get()
+ } catch {
+ case e: ExecutionException if options.owner().isPresent =>
+ e.getCause match {
+ case _: UnsupportedVersionException =>
+ throw new SparkException("Obtaining delegation token for proxy
user requires " +
+ "Kafka 3.3.0 or later brokers (KAFKA-6945).", e)
+ case _: DelegationTokenAuthorizationException =>
+ throw new SparkException("The real user must be granted the
CreateTokens operation " +
+ s"on the ${options.owner().get()} resource.", e)
+ case _ => throw e
+ }
} finally {
Utils.closeQuietly(adminClient)
}
@@ -85,12 +97,26 @@ object KafkaTokenUtil extends Logging {
), token.tokenInfo.expiryTimestamp)
}
- def checkProxyUser(): Unit = {
+ /**
+ * When impersonating, the token is requested with the real user's
credentials but owned by the
+ * proxy user, so connectors authenticate to Kafka as the proxy user. This
needs Kafka 3.3.0 or
+ * later brokers (KAFKA-6945), and the real user must be granted the
`CreateTokens` operation on
+ * the `User:<proxy user>` resource.
+ *
+ * Must be invoked under the UGI that should own the token.
HadoopDelegationTokenManager
+ * preserves a proxy UGI only when no principal/keytab is configured
(spark-submit forbids
+ * combining --proxy-user with --principal) or when direct credential
providers are used;
+ * a keytab re-login drops the proxy identity and the token is then owned by
the real user.
+ */
+ private[kafka010] def createDelegationTokenOptions():
CreateDelegationTokenOptions = {
+ val options = new CreateDelegationTokenOptions()
val currentUser = UserGroupInformation.getCurrentUser()
- // Obtaining delegation token for proxy user is planned but not yet
implemented
- // See https://issues.apache.org/jira/browse/KAFKA-6945
- require(!SparkHadoopUtil.get.isProxyUser(currentUser), "Obtaining
delegation token for proxy " +
- "user is not yet supported.")
+ if (SparkHadoopUtil.get.isProxyUser(currentUser)) {
+ val owner = currentUser.getUserName
+ logInfo(log"Obtaining kafka delegation token for proxy user
${MDC(USER_NAME, owner)}.")
+ options.owner(new KafkaPrincipal(KafkaPrincipal.USER_TYPE, owner))
+ }
+ options
}
def createAdminClientProperties(
@@ -229,13 +255,14 @@ object KafkaTokenUtil extends Logging {
private def printToken(token: DelegationToken): Unit = {
if (log.isDebugEnabled) {
- logDebug("%-15s %-30s %-15s %-25s %-15s %-15s %-15s".format(
- "TOKENID", "HMAC", "OWNER", "RENEWERS", "ISSUEDATE", "EXPIRYDATE",
"MAXDATE"))
+ logDebug("%-15s %-30s %-15s %-15s %-25s %-15s %-15s %-15s".format(
+ "TOKENID", "HMAC", "OWNER", "REQUESTER", "RENEWERS", "ISSUEDATE",
"EXPIRYDATE", "MAXDATE"))
val tokenInfo = token.tokenInfo
- logDebug("%-15s %-15s %-15s %-25s %-15s %-15s %-15s".format(
+ logDebug("%-15s %-30s %-15s %-15s %-25s %-15s %-15s %-15s".format(
tokenInfo.tokenId,
REDACTION_REPLACEMENT_TEXT,
tokenInfo.owner,
+ tokenInfo.tokenRequester,
tokenInfo.renewersAsString,
DATE_TIME_FORMATTER.format(Instant.ofEpochMilli(tokenInfo.issueTimestamp)),
DATE_TIME_FORMATTER.format(Instant.ofEpochMilli(tokenInfo.expiryTimestamp)),
diff --git
a/connector/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaTokenUtilSuite.scala
b/connector/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaTokenUtilSuite.scala
index 8a606a1adc76..c7d6fa647db9 100644
---
a/connector/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaTokenUtilSuite.scala
+++
b/connector/kafka-0-10-token-provider/src/test/scala/org/apache/spark/kafka010/KafkaTokenUtilSuite.scala
@@ -26,6 +26,7 @@ import org.apache.hadoop.io.Text
import org.apache.hadoop.security.UserGroupInformation
import org.apache.kafka.clients.CommonClientConfigs
import org.apache.kafka.common.config.{SaslConfigs, SslConfigs}
+import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.security.auth.SecurityProtocol.{SASL_PLAINTEXT,
SASL_SSL, SSL}
import org.apache.spark.{SparkConf, SparkFunSuite}
@@ -39,16 +40,24 @@ class KafkaTokenUtilSuite extends SparkFunSuite with
KafkaDelegationTokenTest {
sparkConf = new SparkConf()
}
- test("checkProxyUser with proxy current user should throw exception") {
+ test("SPARK-28173: createDelegationTokenOptions without proxy user should
not set owner") {
+ UserGroupInformation.createUserForTesting("realUser", Array()).doAs(
+ new PrivilegedExceptionAction[Unit]() {
+ override def run(): Unit = {
+
assert(!KafkaTokenUtil.createDelegationTokenOptions().owner().isPresent)
+ }
+ }
+ )
+ }
+
+ test("SPARK-28173: createDelegationTokenOptions with proxy user should set
owner") {
val realUser = UserGroupInformation.createUserForTesting("realUser",
Array())
UserGroupInformation.createProxyUserForTesting("proxyUser", realUser,
Array()).doAs(
new PrivilegedExceptionAction[Unit]() {
override def run(): Unit = {
- val thrown = intercept[IllegalArgumentException] {
- KafkaTokenUtil.checkProxyUser()
- }
- assert(thrown.getMessage contains
- "Obtaining delegation token for proxy user is not yet supported.")
+ val owner = KafkaTokenUtil.createDelegationTokenOptions().owner()
+ assert(owner.isPresent)
+ assert(owner.get() === new KafkaPrincipal(KafkaPrincipal.USER_TYPE,
"proxyUser"))
}
}
)
diff --git a/docs/streaming/structured-streaming-kafka-integration.md
b/docs/streaming/structured-streaming-kafka-integration.md
index 17741fc55644..105dac7fb4fc 100644
--- a/docs/streaming/structured-streaming-kafka-integration.md
+++ b/docs/streaming/structured-streaming-kafka-integration.md
@@ -1226,7 +1226,18 @@ For possible Kafka parameters, see [Kafka adminclient
config docs](http://kafka.
#### Caveats
-- Obtaining delegation token for proxy user is not yet supported
([KAFKA-6945](https://issues.apache.org/jira/browse/KAFKA-6945)).
+- Obtaining delegation token for [proxy user](../security.html#proxy-user)
requires Kafka broker 3.3.0 or higher
+ ([KAFKA-6945](https://issues.apache.org/jira/browse/KAFKA-6945)). The token
is requested with the real user's
+ credentials but owned by the proxy user, therefore the real user must be
granted the `CreateTokens` operation
+ on the `User:<proxy user>` resource, such as,
+
+ ./bin/kafka-acls.sh --bootstrap-server <KAFKA_SERVERS> --add \
+ --allow-principal User:<real user> --operation CreateTokens
--user-principal "User:<proxy user>"
+
+ Since `--proxy-user` cannot be combined with `--principal`/`--keytab`, the
real user's Kerberos credentials
+ come from the ticket cache, and the token is obtained once at startup and
not renewed: the application must
+ finish before the token's max lifetime, unless direct credential providers
+ (`spark.security.directCredentialProviders.enabled`) with non-Kerberos Kafka
authentication are used.
### JAAS login configuration
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
