He-Pin opened a new pull request, #2991: URL: https://github.com/apache/pekko/pull/2991
### Motivation `prefixAndTail` emits `(prefix, tailSource)` and then waits for the `tailSource` to be materialized before pulling upstream again. When a downstream consumer intentionally discards the tail Source — a common pattern in protocol parsers using `splitWhen` + `prefixAndTail` + `mapAsync` where a known-empty substream is skipped — the stage holds the substream open until the configured `pekko.stream.materializer.subscription-timeout` fires (default 5 s), silently stalling the rest of the pipeline. Reproduces the long-standing [akka/akka#20008](https://github.com/akka/akka/issues/20008): ```scala Source(List("2", "a", "b", "0", "3", "a", "b", "c", "0", "1", "a")) .splitWhen(_.matches("\\d+")) .prefixAndTail(1) .mapAsync(1) { case (Seq("0"), _) => Future.successful("") // intentionally drop tail case (Seq(_), elements) => elements.runWith(Sink.seq).map(_.mkString) } .concatSubstreams .runWith(Sink.seq) ``` Without the fix every discarded empty tail adds a 5 s wait before the next group is emitted. ### Modification `PrefixAndTail` now issues a single speculative `pull(in)` from `openSubstream()` as soon as the tail `Source` has been emitted: - If upstream finishes before any further element arrives, the stage completes immediately instead of waiting for the substream subscription timer. - If upstream pushes an element before the tail `Source` has been materialized, the element is buffered in a one-element slot and delivered as the first element on the first downstream pull of the tail; upstream finish that arrives while the buffer is occupied is deferred and replayed together with the buffered element when the tail finally subscribes. - Subscription-timeout behaviour for tails that are not subscribed but still have unread upstream elements is unchanged. Operator docs updated to describe the new pre-pull / one-element-buffer semantics. ### Result Discarding a tail `Source` from `prefixAndTail` no longer blocks the parent flow for the subscription-timeout window when the tail is empty. The directional test for the issue scenario now completes in ~60 ms instead of ~15 s. ### Tests - ` sbt \"stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowPrefixAndTailSpec\"` – 18/18 passed locally (incl. new \"complete promptly on empty tail without waiting for subscription timeout (akka #20008)\" and \"deliver buffered tail element when tail is materialized after upstream emitted one\"). - Verified the new directional test fails against the pre-fix implementation by stashing the impl change and re-running the same `testOnly`. - `scalafmt --list` – clean for the changed files. - Full `stream-tests/Test/test` and `+mimaReportBinaryIssues` – Not run locally, deferred to CI (`Check / Binary Compatibility` should cover MiMa). ### References Refs [akka/akka#20008](https://github.com/akka/akka/issues/20008) -- 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]
