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 <noreply@anthropic.com>

Author: reiabreu

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 boolean tryFlush() {
     /* 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 void flush() throws InterruptedException {
             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 boolean tryFlush() {
             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) {