You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@spark.apache.org by ch...@apache.org on 2026/08/13 03:54:50 UTC
(spark) branch branch-4.0 updated: [SPARK-58714][SS] Fix AdminClient leak in KafkaTokenUtil.obtainToken
This is an automated email from the ASF dual-hosted git repository.
pan3793 pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new 736c4df50289 [SPARK-58714][SS] Fix AdminClient leak in KafkaTokenUtil.obtainToken
736c4df50289 is described below
commit 736c4df502893d43d7755d94f4f487e286fab6fe
Author: Cheng Pan <pa...@gmail.com>
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 <pa...@gmail.com>
Signed-off-by: Cheng Pan <ch...@apache.org>
(cherry picked from commit e79ecd67c03a45bcfd2630c48bc2428dde53ff80)
Signed-off-by: Cheng Pan <ch...@apache.org>
---
.../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: commits-unsubscribe@spark.apache.org
For additional commands, e-mail: commits-help@spark.apache.org