He-Pin opened a new pull request, #3179: URL: https://github.com/apache/pekko/pull/3179
### Motivation Per the [stream error handling docs](https://pekko.apache.org/docs/pekko/current/stream/stream-error.html#supervision-strategies), operators that apply user-provided functions should consult the configured `SupervisionStrategy`. The `doOnFirst` operator runs a user side-effect function `f` on the **first** element, but it was called without a try-catch, so any exception failed the stream **unconditionally** — even under `Supervision.Resume`/`Restart`. This is part of the meta-issue #3110 (add supervisor strategy support to stream operators that accept user functions). One operator per PR. ### Modification - Wrap `f(elem)` in `DoOnFirst` with a `try`/`catch` (`NonFatal`) that consults the `SupervisionStrategy` decider: - **Stop** → fail the stage (unchanged fail-fast behavior) - **Resume** → drop the failing first element and switch to pass-through; `f` is **not** retried (the one-shot is consumed) - **Restart** → re-arm first-element detection so the **next** element is treated as the first (`f` is retried on it) - The first-element `InHandler` was extracted into a named `val firstHandler` so it can be re-armed on Restart. On success, the switch to pass-through now happens after `f` (behaviorally identical on the happy path — `push` does not synchronously re-enter `onPush`). - `decider` is a `lazy val` → zero overhead on the happy path. - Fix a fully-qualified type reference in the `createLogic` signature to use the project's two-line import style. - Document supervision adherence (including the Resume/Restart semantics) in the Scala/Java DSL scaladoc and the operator reference page. ### Result `doOnFirst` now adheres to the `SupervisionStrategy` attribute with clear, distinguishable Resume (drop first, no retry) and Restart (next becomes first) semantics. ### Tests - `sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowDoOnFirstSpec"` — **6/6 passed**, including 4 new directional tests: - fails on `f` throw with `Supervision.Stop` - fails with no supervision attribute set (default Stop) — back-compat guard - `Resume` drops the first element, does not retry `f`, passes the rest through - `Restart` re-arms so `f` is retried on each subsequent "first" element - `sbt "stream/mimaReportBinaryIssues"` — clean (binary compatible) ### References Refs #3110 This is a clean-room implementation written directly for Apache Pekko. -- 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]
