This is an automated email from the ASF dual-hosted git repository. reiabreu pushed a commit to branch jcqueue-weak-ref-batcher-leak in repository https://gitbox.apache.org/repos/asf/storm.git
commit c9ca62fecba4ccb7808e3e7aa029b9ffa30f2c91 Author: Rui Abreu <[email protected]> AuthorDate: Sun Jun 28 15:47:48 2026 +0100 utils: use WeakReference in BatchInserter to fix ThreadLocal retention of JCQueue BatchInserter held a strong reference to its owning JCQueue, and the inserters live in instance-field ThreadLocals on the same JCQueue. This formed a cycle through ThreadLocalMap: value (BatchInserter) -> queue (JCQueue) -> thdLocalBatcher (ThreadLocal) = key Because the key was strongly reachable via the value, the weak-key expunge path in ThreadLocalMap never triggered, and the JCQueue (along with its metrics, recv/overflow queues and batch buffers) could not be GC'd for as long as any producer thread that ever published to it stayed alive. The fix stores the JCQueue as a WeakReference inside BatchInserter, cutting the value->key path. When the last external strong ref to the JCQueue is dropped, the ThreadLocal field it owns becomes weakly reachable, the ThreadLocalMap key can be expunged, and both the BatchInserter and the JCQueue are released. flush() and tryFlush() dereference the WeakReference once at entry and bail out cleanly if the queue has already been collected (dead topology in LocalCluster/embedded scenarios). publish() and tryPublish() are unchanged — they only manipulate currentBatch. Fixes #8810 Co-Authored-By: Claude Sonnet 4.6 <[email protected]> --- .../src/jvm/org/apache/storm/utils/JCQueue.java | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java b/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java index 6aa668f56..fe942b985 100644 --- a/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java +++ b/storm-client/src/jvm/org/apache/storm/utils/JCQueue.java @@ -19,6 +19,7 @@ package org.apache.storm.utils; import java.io.Closeable; +import java.lang.ref.WeakReference; import java.util.ArrayList; import java.util.List; import org.apache.storm.metrics2.StormMetricRegistry; @@ -325,11 +326,16 @@ public class JCQueue implements Closeable { /* Not thread safe. Have one instance per producer thread or synchronize externally */ private static class BatchInserter implements Inserter { private final int batchSz; - private JCQueue queue; + // WeakReference breaks the ThreadLocal retention cycle: thdLocalBatcher is an instance field + // of JCQueue, so the ThreadLocalMap key (the ThreadLocal object) is kept strongly reachable + // via value(BatchInserter) -> queue(JCQueue) -> field. A WeakReference here cuts that path, + // allowing the key to become weakly-reachable and the entry to be expunged once the JCQueue + // is no longer externally referenced. + private final WeakReference<JCQueue> queueRef; private ArrayList<Object> currentBatch; BatchInserter(JCQueue queue, int batchSz) { - this.queue = queue; + this.queueRef = new WeakReference<>(queue); this.batchSz = batchSz; this.currentBatch = new ArrayList<>(batchSz + 1); } @@ -368,6 +374,11 @@ public class JCQueue implements Closeable { if (currentBatch.isEmpty()) { return; } + JCQueue queue = queueRef.get(); + if (queue == null) { + currentBatch.clear(); + return; + } int publishCount = queue.tryPublishInternal(currentBatch); int retryCount = 0; while (publishCount == 0) { // retry till at least 1 element is drained @@ -396,6 +407,11 @@ public class JCQueue implements Closeable { if (currentBatch.isEmpty()) { return true; } + JCQueue queue = queueRef.get(); + if (queue == null) { + currentBatch.clear(); + return true; + } int publishCount = queue.tryPublishInternal(currentBatch); if (publishCount == 0) { for (JCQueueMetrics jcQueueMetric : queue.jcqMetrics) {
