Skip to content

feat: add Sink.watchTermination operator - #3409

Draft
He-Pin wants to merge 1 commit into
mainfrom
feat/sink-watch-termination
Draft

feat: add Sink.watchTermination operator#3409
He-Pin wants to merge 1 commit into
mainfrom
feat/sink-watch-termination

Conversation

@He-Pin

@He-Pin He-Pin commented Aug 4, 2026

Copy link
Copy Markdown
Member

Motivation

Sometimes you want to wait for a Sink to fully complete, including any cleanup work or final commit it performs in postStop, but the sink does not materialize a Future[Done]. The existing watchTermination operator is placed before the sink and therefore only signals when the upstream of the sink has terminated.

Modification

Add Sink.watchTermination to the Scala and Java DSLs. It wraps sinks that consist of a single GraphStage with a delegating stage whose materialized Future[Done]/CompletionStage<Done>:

  • completes with success only after the wrapped sink's postStop has run,
  • fails with the upstream failure when the stream failed,
  • fails with an AbruptStreamTerminationException when the stream was abruptly terminated.

The original materialized value, including mapMaterializedValue transforms, is preserved. Composite sinks consisting of multiple stages are rejected with an IllegalArgumentException (documented in the operator page and scaladoc).

Implementation details:

  • WatchedSink rewrites the sink traversal, replacing the single terminal stage with a WatchedSinkStage and replaying the trailing materialized-value composition steps, preserving attributes and islands (.async).
  • WatchedSinkLogic delegates all port handlers and lifecycle hooks to the wrapped logic, mirroring the interpreter, port wiring, stageId and attributes, and records termination causes from the delegated handlers, handler exceptions, and the connection failure slot (covering wrapped stages that swap their inlet handler after materialization, e.g. lazy sinks).
  • GraphStageLogic gains an internal termination hook fired from afterPostStop so the promise also completes when the interpreter finalizes the wrapped logic directly (async-callback self-termination, e.g. Sink.queue cancellation).
  • Operator docs page, operator index entries and Scala/Java doc snippets are included.

Result

Users can await full sink termination, including postStop cleanup, via a materialized Future[Done] / CompletionStage<Done>.

Tests

  • sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.SinkWatchTerminationSpec" - 17/17 passed
  • sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.*Sink*" - 159 passed
  • sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowWatchTerminationSpec org.apache.pekko.stream.scaladsl.QueueSinkSpec org.apache.pekko.stream.scaladsl.GraphStageTimersSpec org.apache.pekko.stream.impl.GraphStageLogicSpec org.apache.pekko.stream.impl.SubInletOutletSpec org.apache.pekko.stream.impl.LinearTraversalBuilderSpec org.apache.pekko.stream.DslConsistencySpec" - 124 passed
  • sbt "stream-tests/Test/testOnly org.apache.pekko.stream.javadsl.SinkTest" - passed
  • sbt stream/mimaReportBinaryIssues - no issues
  • sbt "++3.3.8" stream/compile - passed
  • sbt docs/paradox - passed
  • sbt headerCreateAll scalafmtAll scalafmtSbt javafmtCheckAll - passed
  • scalafmt --mode diff-ref=origin/main - no changes
  • git diff --check - clean
  • sbt sortImports - failed with scalafix plugin NoSuchMethodError (environment issue), imports kept consistent manually

References

Fixes #2377

Motivation:
Sometimes you want to wait for a Sink to fully complete, including any
cleanup work or final commit it performs in postStop, but the sink does
not materialize a Future[Done]. The existing watchTermination operator is
placed before the sink and therefore only signals when the upstream of
the sink has terminated (see #2377, akka/akka-core#22546).

Modification:
Add Sink.watchTermination to the Scala and Java DSLs. It wraps sinks that
consist of a single GraphStage with a delegating stage whose materialized
Future[Done] completes only after the wrapped sink's postStop has run,
fails with the upstream failure when the stream failed, and fails with an
AbruptStreamTerminationException when the stream was abruptly terminated.
The original materialized value, including mapMaterializedValue
transforms, is preserved. Composite sinks consisting of multiple stages
are rejected with an IllegalArgumentException. Implementation details:
- WatchedSink rewrites the sink traversal, replacing the single terminal
  stage with a WatchedSinkStage and replaying the trailing materialized
  value composition steps.
- WatchedSinkLogic delegates all port handlers and lifecycle hooks to the
  wrapped logic, mirroring interpreter, port wiring, stageId and
  attributes, and records termination causes from the delegated handlers,
  handler exceptions, and the connection failure slot (covering wrapped
  stages that swap their inlet handler after materialization).
- GraphStageLogic gains an internal termination hook fired from
  afterPostStop so the promise also completes when the interpreter
  finalizes the wrapped logic directly (async-callback self-termination).

Result:
Users can await full sink termination, including postStop cleanup, via a
materialized Future[Done] / CompletionStage<Done>.

Tests:
- sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.SinkWatchTerminationSpec" - 17/17 passed
- sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.*Sink*" - 159 passed
- sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.FlowWatchTerminationSpec org.apache.pekko.stream.scaladsl.QueueSinkSpec org.apache.pekko.stream.scaladsl.GraphStageTimersSpec org.apache.pekko.stream.impl.GraphStageLogicSpec org.apache.pekko.stream.impl.SubInletOutletSpec org.apache.pekko.stream.impl.LinearTraversalBuilderSpec org.apache.pekko.stream.DslConsistencySpec" - 124 passed
- sbt "stream-tests/Test/testOnly org.apache.pekko.stream.javadsl.SinkTest" - passed
- sbt stream/mimaReportBinaryIssues - no issues
- sbt "++3.3.8" stream/compile - passed
- sbt docs/paradox - passed
- sbt headerCreateAll scalafmtAll scalafmtSbt javafmtCheckAll - passed
- scalafmt --mode diff-ref=origin/main - no changes
- git diff --check - clean
- sbt sortImports - failed with scalafix plugin NoSuchMethodError (environment issue), imports kept consistent manually

References:
Fixes #2377
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

Add Sink.watchTermination operator

1 participant