He-Pin opened a new pull request, #3182: URL: https://github.com/apache/pekko/pull/3182
### 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 `delayWith` operator computes each element's delay via a user-provided `DelayStrategy.nextDelay`, 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 `delayStrategy.nextDelay(element)` in `Delay.grabAndPull()` with a `try`/`catch` (`NonFatal`) that consults the `SupervisionStrategy` decider: - **Stop** → fail the stage (unchanged fail-fast behavior) - **Resume** → drop the offending element, keep already-buffered elements and the current strategy - **Restart** → recreate the (potentially stateful) delay strategy from the supplier; already-buffered elements keep their computed delays - `delayStrategy` became a `var` so Restart can recreate it. - **Buffer-underflow guard**: because a dropped element no longer enqueues, `onPush` and `onTimer` now guard their buffer access with a non-empty check. In particular, `onTimer` could previously hit a `NullPointerException` (the delay buffer is an unchecked fixed-size buffer) when an overflow handler emptied the buffer while a timer was still armed and a `nextDelay` exception was then skipped. A dedicated regression test reproduces this (it fails without the guard). - `decider` is a `lazy val` → zero overhead on the happy path. - Document supervision adherence on `delayWith` in the Scala/Java DSL scaladoc and the operator reference page (plain `delay` is unaffected — it uses a fixed internal strategy that never throws). ### Result `delayWith` now adheres to the `SupervisionStrategy` attribute with well-defined Resume/Restart semantics and no buffer underflow on the supervised paths. ### Tests - `sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowDelaySpec"` — **26/26 passed**, including 6 new tests: Stop, default Stop, Resume (drops element), Restart (recreates the stateful strategy — verified via an instance counter), Resume (keeps the stateful strategy — inverted mirror), and an overflow-handler + throw regression that guards the `onTimer` underflow fix. - `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]
