Skip to content

GroupBy should support infinite substreams with maxSubstreams=-1 #3102

Description

@He-Pin

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

  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.

Metadata

Metadata

Assignees

No one assigned

    Labels

    akkadotnetIssue/PR inspired by Akka.netenhancementNew feature or requestt:streamPekko Streams

    Type

    No type

    Fields

    No fields configured for issues without a type.

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions