reiabreu commented on PR #8796: URL: https://github.com/apache/storm/pull/8796#issuecomment-4791588080
@GGraziadei thanks for the submission. Overall it looks good.
I always try to run the change set through an LLM to make sure we are not
missing anything.
Here are some suggestions. Please note that point A was already present
before your PR.
### A. Non-Static ThreadLocal Memory Leak Cycle
The PR defines the thread-local fields as instance variables of JCQueue :
private final ThreadLocal<BatchInserter> thdLocalBatcher = new
ThreadLocal<BatchInserter>();
private final ThreadLocal<DynamicBatchInserter> thdLocalDynamicBatcher =
new ThreadLocal<DynamicBatchInserter>();
In Java, this creates a cyclic reference cycle:
• Thread holds ThreadLocalMap .
• ThreadLocalMap contains value DynamicBatchInserter .
• DynamicBatchInserter holds a strong reference back to the outer
JCQueue instance via this.queue .
• JCQueue holds the ThreadLocal key instance.
Even if the JCQueue reference is discarded elsewhere, it cannot be
garbage collected because it is reachable from the thread's map, which prevents
the key from being weakly cleared.
│ [!NOTE]
│ This pattern already existed for thdLocalBatcher in JCQueue prior to
this PR. However, adding thdLocalDynamicBatcher replicates the leak. While
usually bounded by the lifetime of the worker thread in Storm, in environments
│ where topologies are frequently started/stopped on static threads, this
will leak JCQueue memory.
│
│ Mitigation: The ThreadLocal could be made static , using a composite
key (like ThreadLocal<Map<JCQueue, Inserter>> where keys are weak references)
or cleaning them up during close() .
### B. Transient Backpressure Smoothing
In tryFlush() , when the queue is full under backpressure,
tryPublishInternal returns 0. The method returns false but still executes
the afterFlush(wasFull) hook:
public boolean tryFlush() {
...
boolean wasFull = currentBatch.size() >= batchSize();
int publishCount = queue.tryPublishInternal(currentBatch);
if (publishCount == 0) {
...
afterFlush(wasFull);
return false;
...
If wasFull was true , effectiveBatchSz increases.
When the caller retries tryPublish() , the new effectiveBatchSz is
larger. As a result, the check currentBatch.size() >= batchSize() now returns
false .
This allows the retried item to be appended to the local batch, and
tryPublish() returns true (success) instead of immediately propagating
backpressure to the caller.
│ [!TIP]
│ This behaves as a transient "buffer absorber" under sudden backpressure.
The backpressure signal is temporarily delayed until the local batch size
reaches the hard cap ( maxBatchSz ), at which point it consistently blocks.
This is
a
│ safe and beneficial smoothing behavior, but a subtle consequence of
calling afterFlush() on failed flushes.
--
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]
