reiabreu opened a new pull request, #8812:
URL: https://github.com/apache/storm/pull/8812
Fixes #8810
## Problem
`BatchInserter` held a **strong** `JCQueue queue` reference, and inserter
instances live in **instance-field** `ThreadLocal`s on the same `JCQueue`. This
creates a cycle that prevents the normal ThreadLocal weak-key expunge path from
ever firing:
```
Thread
└─ ThreadLocalMap
key (weak): JCQueue.thdLocalBatcher ← the ThreadLocal object
value (strong): BatchInserter
└─ queue: JCQueue (strong)
└─ thdLocalBatcher: ThreadLocal ← key is
now strongly reachable!
```
The weak key in `ThreadLocalMap` becomes eligible for expunge only when it
is *not* strongly reachable. Here it always is — via `value → queue → field` —
so the entry is never cleaned up, and the `JCQueue` (recv queue, overflow
queue, metrics, batch buffer) is retained for as long as the producer thread
lives.
In production this is harmless: Storm runs workers as dedicated JVM
processes where threads and queues share a lifetime. In `LocalCluster`,
embedded deployments, or test harnesses that repeatedly start/stop topologies
on long-lived threads, each stopped topology can strand its `JCQueue` instances.
## Fix
Store the `JCQueue` as a `WeakReference` inside `BatchInserter`. This cuts
the `value → JCQueue` strong path:
```
Thread
└─ ThreadLocalMap
key (weak): JCQueue.thdLocalBatcher
value (strong): BatchInserter
└─ queueRef: WeakReference → JCQueue (no longer a
strong path to the key)
```
When the last external strong ref to the `JCQueue` is dropped, the
`thdLocalBatcher` field (and therefore the `ThreadLocalMap` key) becomes
weakly-reachable, and the entry is expunged on the next `ThreadLocal` access by
the producer thread. `BatchInserter` is then released too.
`flush()` and `tryFlush()` dereference the `WeakReference` once at entry and
bail out silently if the queue was already collected (the topology is dead;
in-flight tuples are already lost). `publish()` and `tryPublish()` need no
changes — they only touch `currentBatch`.
## Note for PR #8796
`DynamicBatchInserter` (introduced in #8796) extends `BatchInserter` and its
own overrides (`batchSize()`, `afterFlush()`) never access `queue` directly, so
this fix covers it automatically — no separate change is needed there.
## Test plan
- [ ] `mvn test -pl storm-client -Dtest=JCQueueTest` passes
- [ ] Optionally: reproduce with a `LocalCluster` that repeatedly
submits/kills a topology and verify the `JCQueue` instances are no longer
retained after kill
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
