He-Pin opened a new issue, #3248: URL: https://github.com/apache/pekko/issues/3248
### Motivation `MergeHub.source` uses an unbounded `AbstractNodeQueue[Event]` as its central queue (`Hub.scala:182`). While per-producer backpressure via `perProducerBufferSize` (default 16) limits each individual producer's contribution, there is **no aggregate bound** across all producers. In dynamic scenarios where producers continuously materialize while downstream is stalled, the queue grows without limit, leading to OOM. ### Current Behavior - Each producer is limited to `perProducerBufferSize` elements before self-backpressuring - However, new producer materializations unconditionally enqueue `Register` events and begin pushing (`Hub.scala:325-339`) - No check on queue size, number of active producers, or downstream health exists - `AbstractNodeQueue.add()` (`AbstractNodeQueue.java:91-94`) always succeeds — no capacity check ### Queue Growth Formula Given `C` producer connect/disconnect cycles with stalled downstream and `perProducerBufferSize = B`: - Max queue size ≈ `C × (B + 2)` events (B elements + Register + Deregister per cycle) - With B=16 and 1,000,000 cycles: ~18,000,000 events in queue (~1GB+ heap) ### Expected Behavior The central queue should have a configurable capacity limit, or new producer connections should be rejected/backpressured when aggregate queue size exceeds a threshold. ### Reproduction 1. Create a `MergeHub.source` with default `perProducerBufferSize` 2. Dynamically materialize many producers while downstream is stalled (no demand) 3. Observe unbounded queue growth until OOM ### Environment - Pekko Stream (any version) - `Hub.scala:182`, `Hub.scala:322-339`, `Hub.scala:163-168` ### References None -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
