feat: add supervision strategy support for throttle costCalculation

[He-Pin]

Motivation

Per the stream error handling docs, 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.