He-Pin opened a new issue, #3101:
URL: https://github.com/apache/pekko/issues/3101
### Motivation
The `Throttle` stage's `costCalculation` function is called without any
exception handling. If the user-provided cost function throws, the exception
propagates directly and fails the stage unconditionally. Unlike other stages
(e.g., `MapAsync`, `Log`, `Filter`), Throttle does not consult the
`SupervisionStrategy` decider, preventing users from configuring Resume or
Restart behavior for bad elements.
### Current behavior
In `Throttle.scala:71-84`, `costCalculation(elem)` is called without
try-catch:
```scala
override def onPush(): Unit = {
val elem = grab(in)
val cost = costCalculation(elem) // No error handling
val delayNanos = tokenBucket.offer(cost)
...
}
```
### Proposed fix
1. Add `lazy val decider =
inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider` in
`createLogic`
2. Mix in `StageLogging` trait for error logging
3. Wrap `costCalculation(elem)` in try-catch, consult decider on exception:
- `Stop` → `failStage(ex)` (current behavior)
- `Resume`/`Restart` → log error and `pull(in)` to skip the element
4. Add tests verifying supervision directives work with Throttle
### References
- Inspired by Akka.NET
[#6886](https://github.com/akkadotnet/akka.net/pull/6886)
- Similar pattern exists in Pekko's `MapAsync` (`Ops.scala:1301`) and `Log`
(`Ops.scala:1542`)
--
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]
