This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new adee66659ee1 [SPARK-58714][SS] Fix AdminClient leak in
KafkaTokenUtil.obtainToken
adee66659ee1 is described below
commit adee66659ee1fdc8e740cecb448cf40bd058415c
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]>
(cherry picked from commit e79ecd67c03a45bcfd2630c48bc2428dde53ff80)
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]
