feat: add Sink.watchTermination operator - #3409
Draft
He-Pin wants to merge 1 commit into
Draft
Conversation
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
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Motivation
Sometimes you want to wait for a
Sinkto fully complete, including any cleanup work or final commit it performs inpostStop, but the sink does not materialize aFuture[Done]. The existingwatchTerminationoperator is placed before the sink and therefore only signals when the upstream of the sink has terminated.Modification
Add
Sink.watchTerminationto the Scala and Java DSLs. It wraps sinks that consist of a singleGraphStagewith a delegating stage whose materializedFuture[Done]/CompletionStage<Done>:postStophas run,AbruptStreamTerminationExceptionwhen the stream was abruptly terminated.The original materialized value, including
mapMaterializedValuetransforms, is preserved. Composite sinks consisting of multiple stages are rejected with anIllegalArgumentException(documented in the operator page and scaladoc).Implementation details:
WatchedSinkrewrites the sink traversal, replacing the single terminal stage with aWatchedSinkStageand replaying the trailing materialized-value composition steps, preserving attributes and islands (.async).WatchedSinkLogicdelegates all port handlers and lifecycle hooks to the wrapped logic, mirroring the interpreter, port wiring,stageIdand 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).GraphStageLogicgains an internal termination hook fired fromafterPostStopso the promise also completes when the interpreter finalizes the wrapped logic directly (async-callback self-termination, e.g.Sink.queuecancellation).Result
Users can await full sink termination, including
postStopcleanup, via a materializedFuture[Done]/CompletionStage<Done>.Tests
sbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.SinkWatchTerminationSpec"- 17/17 passedsbt "stream-tests/Test/testOnly org.apache.pekko.stream.scaladsl.*Sink*"- 159 passedsbt "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 passedsbt "stream-tests/Test/testOnly org.apache.pekko.stream.javadsl.SinkTest"- passedsbt stream/mimaReportBinaryIssues- no issuessbt "++3.3.8" stream/compile- passedsbt docs/paradox- passedsbt headerCreateAll scalafmtAll scalafmtSbt javafmtCheckAll- passedscalafmt --mode diff-ref=origin/main- no changesgit diff --check- cleansbt sortImports- failed with scalafix pluginNoSuchMethodError(environment issue), imports kept consistent manuallyReferences
Fixes #2377