He-Pin opened a new pull request, #3461: URL: https://github.com/apache/pekko/pull/3461
### Motivation PR #3288 batched elements at the async output boundary and flushed them when the interpreter parked. This deferred the demand round-trip to once per flush cycle (~16 elements) instead of per element, silently breaking external-source stages that drop elements when `isAvailable(out)` is false. The `pekko-connectors` UDP connector loses ~15/16 datagrams on M4. Fixes #3459 ### Modification Revert the output-side batching: send each element directly to the `BoundarySubscriber` via `next()` during `onPush()`, restoring the pre-M4 timing where demand is replenished after every element. Remove the now-unused `enqueueBatchElement`/`flushBatch`/`clearBatch`/`flushOutputs` machinery. The `BoundarySubscriber` trait, `OnNextBatch` event class, and input-side `receiveNextBatch` are retained as internal API surface (no behavioral change on the input side). ### Result Demand across an internal async boundary is again replenished per element, fixing silent datagram loss in the UDP connector and any other stage that treats "no demand right now" as a signal to drop. ### Tests - `sbt "stream-tests / Test / testOnly org.apache.pekko.stream.FusingSpec"` -- 19/19 passed - `sbt "stream-tests / Test / testOnly org.apache.pekko.stream.scaladsl.PublisherSinkSpec org.apache.pekko.stream.io.FileSinkSpec org.apache.pekko.stream.scaladsl.FlowMapWithResourceSpec org.apache.pekko.stream.scaladsl.FlowSpec"` -- 78/78 passed - `sbt "stream / mimaReportBinaryIssues"` -- passed ### References Fixes #3459 -- 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]
