Motivation
GroupBy requires a fixed maxSubstreams limit. When the number of distinct keys exceeds this limit, the stream fails with TooManySubstreamsOpenException. Some use cases (e.g., event routing by tenant/user ID) need to handle an unbounded number of keys without a predefined limit.
Current behavior
StreamOfStreams.scala:485:
} else if (activeSubstreamsMap.size + closedSubstreams.size == maxSubstreams) {
throw tooManySubstreamsOpenException
}
Passing maxSubstreams = -1 technically bypasses the check (sizes can never equal -1), but this is undocumented and not an official API contract.
Proposed fix
- Add explicit guard:
if (maxSubstreams >= 0 && activeSubstreamsMap.size + closedSubstreams.size == maxSubstreams)
- Update Scaladoc in
Flow.scala:groupBy to document -1 as "unlimited substreams"
- Add
require(maxSubstreams >= -1, ...) validation
- Add tests for
maxSubstreams = -1 with many distinct keys
- Document the memory implication: with
allowClosedSubstreamRecreation = false, closedSubstreams grows unbounded. Recommend using allowClosedSubstreamRecreation = true with infinite substreams.
Motivation
GroupByrequires a fixedmaxSubstreamslimit. When the number of distinct keys exceeds this limit, the stream fails withTooManySubstreamsOpenException. Some use cases (e.g., event routing by tenant/user ID) need to handle an unbounded number of keys without a predefined limit.Current behavior
StreamOfStreams.scala:485:Passing
maxSubstreams = -1technically bypasses the check (sizes can never equal -1), but this is undocumented and not an official API contract.Proposed fix
if (maxSubstreams >= 0 && activeSubstreamsMap.size + closedSubstreams.size == maxSubstreams)Flow.scala:groupByto document-1as "unlimited substreams"require(maxSubstreams >= -1, ...)validationmaxSubstreams = -1with many distinct keysallowClosedSubstreamRecreation = false,closedSubstreamsgrows unbounded. Recommend usingallowClosedSubstreamRecreation = truewith infinite substreams.