MergeHub central queue unbounded in dynamic producer scenarios

[He-Pin]

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