He-Pin opened a new pull request, #3392: URL: https://github.com/apache/pekko/pull/3392
### Motivation `MergeHub.source` uses an unbounded `AbstractNodeQueue` as its central queue. While per-producer backpressure via `perProducerBufferSize` 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. Fixes #3248 ### Modification - Add `maxTotalBufferSize` parameter to `MergeHub` (scaladsl + javadsl) - Track buffered element count with `AtomicInteger` (incremented on enqueue, decremented on consume) - New producers are cancelled at registration when the count meets or exceeds the threshold - Default `0` preserves existing unlimited behavior (fully backward compatible) New API methods: - `MergeHub.source[T](perProducerBufferSize, maxTotalBufferSize)` (scaladsl) - `MergeHub.sourceWithDraining[T](perProducerBufferSize, maxTotalBufferSize)` (scaladsl) - `MergeHub.of[T](clazz, perProducerBufferSize, maxTotalBufferSize)` (javadsl) - `MergeHub.withDraining[T](clazz, perProducerBufferSize, maxTotalBufferSize)` (javadsl) ### Result Users can bound aggregate MergeHub queue growth via admission control, preventing OOM in dynamic producer churn scenarios. The bound is a soft admission threshold — transient overshoot up to the per-producer buffer of concurrently admitted producers is possible. ### Tests - `sbt "stream-tests / Test / testOnly org.apache.pekko.stream.scaladsl.HubSpec -- -z MergeHub"` — 17 tests passed (including 2 new directional tests) - `sbt "stream / mimaReportBinaryIssues"` — passed (purely additive API) - `scalafmt --mode diff-ref=origin/main` — applied - `sbt headerCreateAll` — applied ### References Fixes #3248 -- 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]
