He-Pin opened a new pull request, #3183: URL: https://github.com/apache/pekko/pull/3183
### 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 `throttle` operator's `costCalculation` function is user-provided and may throw, 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 `costCalculation(elem)` in `Throttle.onPush()` with a `try`/`catch` (`NonFatal`) that consults the `SupervisionStrategy` decider: - **Stop** → fail the stage (unchanged fail-fast behavior) - **Resume** → drop the offending element, keep the token bucket state - **Restart** → behave the same as Resume (throttle keeps no accumulated per-element state, so there is nothing element-specific to reset) - The cost calculation is evaluated **before** `tokenBucket.offer(cost)`, so a throwing function never consumes rate budget. - Add a guard in the timer path so a supervised skip cannot dequeue from an empty buffer after an overflow handler has already mutated it. - `decider` is a `lazy val` → zero overhead on the happy path. - Document supervision adherence on the two `costCalculation` throttle overloads in the Scala/Java DSL scaladoc and the operator reference page. The element-count throttle overloads are intentionally left undocumented because they use a constant cost and never throw. ### Result `throttle` now adheres to the `SupervisionStrategy` attribute for its costCalculation function, with no buffer-underflow on supervised skip paths. ### Tests - `sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowThrottleSpec"` — **28/28 passed**, including 6 new tests: explicit Stop, default Stop, Resume (Shaping), Resume (Enforcing), Restart (same as Resume for throttle), and a last-element Resume regression. - `sbt "stream/mimaReportBinaryIssues"` — clean (binary compatible) ### References Refs #3110, Refs #3101 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]
