This is an automated email from the ASF dual-hosted git repository.
pan3793 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 e79ecd67c03a [SPARK-58714][SS] Fix AdminClient leak in
KafkaTokenUtil.obtainToken
e79ecd67c03a is described below
commit e79ecd67c03a45bcfd2630c48bc2428dde53ff80
Author: Cheng Pan <[email protected]>
AuthorDate: Thu Aug 13 11:53:27 2026 +0800
[SPARK-58714][SS] Fix AdminClient leak in KafkaTokenUtil.obtainToken
### What changes were proposed in this pull request?
Wrap the token acquisition in `KafkaTokenUtil.obtainToken` in
`try`/`finally` so the created `AdminClient` is always closed, via
`Utils.closeQuietly`.
### Why are the changes needed?
`obtainToken` creates an `AdminClient` but never closes it, on either the
success or the exception path. Each token obtain (per cluster, per renewal
cycle) and each failed attempt leaks the client's network thread and sockets,
which accumulate in long-running drivers.
Present since the feature was introduced in SPARK-25501 (Spark 3.0.0).
### Does this PR introduce _any_ user-facing change?
No.
### How was this patch tested?
Existing tests (`KafkaTokenUtilSuite`).
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (claude-fable-5)
Closes #57953 from pan3793/kafka-admin-client-leak.
Authored-by: Cheng Pan <[email protected]>
Signed-off-by: Cheng Pan <[email protected]>
---
.../main/scala/org/apache/spark/kafka010/KafkaTokenUtil.scala | 10 +++++++---
1 file changed, 7 insertions(+), 3 deletions(-)
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 54014859bc5e..cbcc27ff7d28 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
@@ -68,9 +68,13 @@ object KafkaTokenUtil extends Logging {
checkProxyUser()
val adminClient =
AdminClient.create(createAdminClientProperties(sparkConf, clusterConf))
- val createDelegationTokenOptions = new CreateDelegationTokenOptions()
- val createResult =
adminClient.createDelegationToken(createDelegationTokenOptions)
- val token = createResult.delegationToken().get()
+ val token = try {
+ val createDelegationTokenOptions = new CreateDelegationTokenOptions()
+ val createResult =
adminClient.createDelegationToken(createDelegationTokenOptions)
+ createResult.delegationToken().get()
+ } finally {
+ Utils.closeQuietly(adminClient)
+ }
printToken(token)
(new Token[KafkaDelegationTokenIdentifier](
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
