[I] Meta: Add supervisor strategy support to stream operators that accept user functions [pekko]

["He-Pin (via GitHub)" <gi...@apache.org>]

He-Pin opened a new issue, #3110:
URL: https://github.com/apache/pekko/issues/3110

   ### Motivation
   
   Per the [stream error handling docs](https://pekko.apache.org/docs/pekko/current/stream/stream-error.html#supervision-strategies):
   
   > The operators that support supervision strategies are explicitly documented to do so, if there is nothing in the documentation of an operator saying that it adheres to the supervision strategy it means it fails rather than applies supervision.
   
   Several stream operators accept user-provided functions that can throw exceptions, yet do not consult the `SupervisionStrategy` decider. This means that if the user function throws, the stream fails unconditionally, regardless of the configured supervision strategy. For consistency and usability, all operators that apply user functions to stream elements should support supervisor strategy.
   
   Related: #3101 (Throttle-specific issue)
   
   ### Operators missing supervisor strategy support
   
   The following operators accept user-provided functions but do **not** implement supervision. Each entry includes the function parameter, the source location, and the specific code path where the function is called without try-catch.
   
   #### 1. `throttle` (with `costCalculation`)
   
   - **Function**: `costCalculation: Out => Int`
   - **Implementation**: `Throttle.scala:73` — `costCalculation(elem)` in `onPush()` without try-catch
   - **Impact**: If `costCalculation` throws (e.g. for malformed input), the stream fails even with `Supervision.Resume`
   - **Tracked in**: #3101
   
   #### 2. `expand`
   
   - **Function**: `expander: Out => Iterator[U]`
   - **Implementation**: `Ops.scala:1210` — `extrapolate(grab(in))` in `onPush()` without try-catch
   - **Doc**: Explicitly states "Expand does not support `Supervision.Restart` and `Supervision.Resume`. Exceptions from the `seed` function will complete the stream with failure."
   - **Impact**: Should be enhanced to support supervision, or the limitation should remain clearly documented if semantics prevent it
   
   #### 3. `groupedWeighted`
   
   - **Function**: `costFn: Out => Long`
   - **Implementation**: `Ops.scala:816` — `costFn(elem)` in `GroupedWeighted.onPush()` without try-catch
   - **Impact**: Negative cost is handled, but a throwing `costFn` is not supervised
   
   #### 4. `groupedWeightedWithin`
   
   - **Function**: `costFn: Out => Long`
   - **Implementation**: `Ops.scala:1797` — `costFn(elem)` in `GroupedWeightedWithin.nextElement()` without try-catch
   - **Impact**: Same as `groupedWeighted`
   
   #### 5. `groupedAdjacentBy` / `groupedAdjacentByWeighted`
   
   - **Functions**: `f: Out => T` and `costFn: Out => Long`
   - **Implementation**: `GroupedAdjacentByWeighted.scala:60,67` — both `costFn(elem)` and `f(elem)` called without try-catch
   - **Impact**: Neither the key function nor the cost function is supervised
   
   #### 6. `aggregateWithBoundary`
   
   - **Functions**: `allocate: () => Agg`, `aggregate: (Agg, In) => (Agg, Boolean)`, `harvest: Agg => Out`, `emitOnTimer: Option[(Agg => Boolean, FiniteDuration)]`
   - **Implementation**: `AggregateWithBoundary.scala` — all user functions called without try-catch
   - **Impact**: Any exception in the aggregation, harvest, or timer functions fails the stream unconditionally
   
   #### 7. `doOnFirst`
   
   - **Function**: `f: Out => Unit`
   - **Implementation**: `DoOnFirst.scala:48` — `f(elem)` called without try-catch
   - **Impact**: Side-effect function failure fails the stream
   
   #### 8. `delay` (with custom `DelayStrategy`)
   
   - **Function**: `DelayStrategy.nextDelay(element)` (via `delayStrategySupplier`)
   - **Implementation**: `Ops.scala:1994` — `delayStrategy.nextDelay(element)` in `grabAndPull()` without try-catch
   - **Impact**: If the delay strategy throws for a particular element, the stream fails
   
   #### 9. `zipWith` / `zipWithN`
   
   - **Function**: `combine` / `zipper` function
   - **Implementation**: `ZipWithApply.scala.template:51` — `zipper(...)` in `pushAll()` without try-catch; `Graph.scala:1241` — same pattern in `ZipWithN`
   - **Impact**: Combine function exceptions fail the stream unconditionally
   
   #### 10. `batch` (partial gap — `costFn` only)
   
   - **Function**: `costFn: In => Long`
   - **Implementation**: `Ops.scala:1108` — `costFn(elem)` called **outside** the try-catch block in `onPush()`, while `seed` and `aggregate` calls are properly supervised
   - **Impact**: `seed` and `aggregate` are supervised, but `costFn` is not — inconsistent within the same operator
   
   ### Operators that already support supervision (for reference)
   
   The following operators correctly implement supervision: `map`, `filter`, `filterNot`, `mapOption`, `takeWhile`, `dropWhile`, `collect`, `collectWhile`, `collectFirst`, `scan`, `scanAsync`, `fold`, `foldAsync`, `reduce`, `limitWeighted`, `batch` (seed/aggregate only), `mapConcat`, `statefulMapConcat`, `statefulMap`, `mapAsync`, `mapAsyncUnordered`, `mapAsyncPartitioned`, `log`, `logWithMarker`, `conflate`/`conflateWithSeed`, `batchWeighted`, `groupBy`, `splitWhen`/`splitAfter`, `partition`, `dropRepeated`, `mapWithResource`, `unfoldResource`, `unfoldResourceAsync`, `iteratorSource`, `iterableSource`, `iterableConcat`
   
   ### Proposed approach
   
   For each operator listed above:
   
   1. Add `private lazy val decider = inheritedAttributes.mandatoryAttribute[SupervisionStrategy].decider` in `createLogic`
   2. Wrap user function calls in try-catch with `NonFatal` matching
   3. Consult the decider on exception:
      - `Stop` → `failStage(ex)` (current behavior)
      - `Resume` → skip the element, continue processing (e.g. `pull(in)`)
      - `Restart` → reset any accumulated state, then continue
   4. Mix in `StageLogging` where needed for error logging on resume/restart
   5. Add tests verifying all three supervision directives work correctly
   6. Update scaladoc to include "Adheres to the `[[ActorAttributes.SupervisionStrategy]]` attribute."
   7. For `expand`: if supervision semantics are genuinely incompatible (e.g. the extrapolation iterator state cannot be meaningfully reset), keep the explicit documentation but consider supporting at least `Resume`
   
   ### Notes on semantics
   
   Some operators may need special consideration:
   
   - **`expand`**: The extrapolation iterator is stateful; `Restart` would need to reset it. `Resume` semantics are unclear when the expander itself fails mid-iteration.
   - **`zipWith`/`zipWithN`**: When the combine function fails, the elements from all upstreams have already been grabbed. On `Resume`, those elements would be lost, which could be surprising but is consistent with how `map` handles it.
   - **`delay`**: The delay strategy is created once at materialization. `Restart` semantics for a stateful `DelayStrategy` would need to re-create it, which may not be feasible without a factory.
   - **`aggregateWithBoundary`**: With 4 user functions, each may need different supervision semantics. Consider whether all should be supervised uniformly.
   - **Source operators** (`unfold`, `unfoldAsync`): Supervision for sources is less clear since there's no upstream element to "skip". These are lower priority.
   
   ### References
   
   - #3101 — Throttle-specific issue
   - [Stream Error Handling Docs](https://pekko.apache.org/docs/pekko/current/stream/stream-error.html#supervision-strategies)


-- 
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: notifications-unsubscribe@pekko.apache.org.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org

Re: [I] Meta: Add supervisor strategy support to stream operators that accept user functions [pekko]

["He-Pin (via GitHub)" <gi...@apache.org>]

He-Pin closed issue #3110: Meta: Add supervisor strategy support to stream operators that accept user functions
URL: https://github.com/apache/pekko/issues/3110


-- 
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: notifications-unsubscribe@pekko.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@pekko.apache.org
For additional commands, e-mail: notifications-help@pekko.apache.org