GroupBy should support infinite substreams with maxSubstreams=-1
- Dominant language
- Scala
- Stars
- 1.6k
- Forks
- 211
- Avg merge
- 1d 6h
- Merged PRs (30d)
- 89
Description
### 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`:
```scala
} 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
1. Add explicit guard: `if (maxSubstreams >= 0 && activeSubstreamsMap.size + closedSubstreams.size == maxSubstreams)`
2. Update Scaladoc in `Flow.scala:groupBy` to document `-1` as "unlimited substreams"
3. Add `require(maxSubstreams >= -1, ...)` validation
4. Add tests for `maxSubstreams = -1` with many distinct keys
5. Document the memory implication: with `allowClosedSubstreamRecreation = false`, `closedSubstreams` grows unbounded. Recommend using `allowClosedSubstreamRecreation = true` with infinite substreams.
Contributor guide
Research direction
Start with the GroupBy implementation at StreamOfStreams.scala:485 and the groupBy Scaladoc in Flow.scala. Add coverage for maxSubstreams = -1 with many distinct keys and validation for values below -1; done means unlimited substreams are documented and supported, with the closed-substream memory implication explained.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- scala
- Domain
- stream-processing
- Issue type
- Feature
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 72/100