feat: add maxTotalBufferSize to MergeHub for aggregate queue bound - #3392
Open
He-Pin wants to merge 1 commit into
Open
feat: add maxTotalBufferSize to MergeHub for aggregate queue bound#3392He-Pin wants to merge 1 commit into
He-Pin wants to merge 1 commit into
Conversation
Motivation: MergeHub's central queue is unbounded. In dynamic producer scenarios where producers continuously materialize while downstream is stalled, the queue grows without limit leading to OOM (issue #3248). Modification: Add maxTotalBufferSize parameter to MergeHub (scaladsl + javadsl). Track buffered element count with AtomicInteger. New producers are cancelled at registration when the count meets or exceeds the threshold. Default 0 preserves existing unlimited behavior. Result: Users can bound aggregate MergeHub queue growth via admission control, preventing OOM in dynamic producer churn scenarios. 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 References: Fixes #3248
pjfanning
reviewed
Jul 28, 2026
| * New producers are cancelled at registration when the buffered element count meets or exceeds this value. | ||
| * Transient overshoot up to the per-producer buffer of concurrently admitted producers is possible. | ||
| * Use 0 for unlimited (default behavior). | ||
| */ |
pjfanning
reviewed
Jul 28, 2026
| * New producers are cancelled at registration when the buffered element count meets or exceeds this value. | ||
| * Transient overshoot up to the per-producer buffer of concurrently admitted producers is possible. | ||
| * Use 0 for unlimited (default behavior). | ||
| */ |
pjfanning
reviewed
Jul 28, 2026
| * New producers are cancelled at registration when the buffered element count meets or exceeds this value. | ||
| * Transient overshoot up to the per-producer buffer of concurrently admitted producers is possible. | ||
| * Use 0 for unlimited (default behavior). | ||
| */ |
Member
There was a problem hiding this comment.
this class also needs @since 2.0.0 on the new public methods
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation
MergeHub.sourceuses an unboundedAbstractNodeQueueas its central queue. While per-producer backpressure viaperProducerBufferSizelimits 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
maxTotalBufferSizeparameter toMergeHub(scaladsl + javadsl)AtomicInteger(incremented on enqueue, decremented on consume)0preserves 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— appliedsbt headerCreateAll— appliedReferences
Fixes #3248