feat: add Sink.watchTermination operator

docs/src/main/paradox/stream/operators/Sink/watchTermination.md

@@ -0,0 +1,44 @@
+# Sink.watchTermination
+
+Wraps a sink so that in addition to the original materialized value a @scala[`Future[Done]`] @java[`CompletionStage<Done>`] is materialized that only completes after the wrapped sink has fully terminated, including its `postStop` lifecycle hook.
+
+@ref[Sink operators](../index.md#sink-operators)
+
+## Signature
+
+@apidoc[Sink.watchTermination](Sink) { scala="#watchTermination[Mat2]()(matF:(Mat,scala.concurrent.Future[org.apache.pekko.Done])=&gt;Mat2):org.apache.pekko.stream.scaladsl.Sink[In,Mat2]" java="#watchTermination(org.apache.pekko.japi.function.Function2)" }
+
+
+## Description
+
+Wraps a sink so that in addition to the original materialized value a @scala[`Future[Done]`] @java[`CompletionStage<Done>`] is materialized
+that completes when the wrapped sink has fully terminated: it completes with success after the wrapped sink's `postStop`
+lifecycle hook has run, or fails with the upstream failure when the stream failed.
+
+This differs from @ref[watchTermination](../Source-or-Flow/watchTermination.md), which is placed *before* the sink and
+therefore only signals when the upstream of the sink has terminated. Because `Sink.watchTermination` wraps the sink
+itself, the materialized @scala[`Future`] @java[`CompletionStage`] can be used to wait for any cleanup or final
+commits the sink performs in `postStop`, for example a file sink closing the file it was writing to.
+
+Only sinks that consist of a single `GraphStage` are supported, for example `Sink.ignore`, `Sink.head`,
+`Sink.queue`, `Sink.actorRef` or sinks created from custom graph stages. Composite sinks consisting of
+multiple stages — such as `Sink.foreach`, `Sink.fold`, or sinks created with `Sink.combine` or `GraphDSL` —
+are not supported and throw an @scala[`IllegalArgumentException`] @java[`IllegalArgumentException`].
+
+## Examples
+
+Scala
+:   @@snip [WatchTermination.scala](/docs/src/test/scala/docs/stream/operators/sink/WatchTermination.scala) { #watchTermination }
+
+Java
+:   @@snip [WatchTermination.java](/docs/src/test/java/jdocs/stream/operators/sink/WatchTermination.java) { #watchTermination }
+
+## Reactive Streams semantics
+
+@@@div { .callout }
+
+**backpressures** when the wrapped sink backpressures
+
+**cancels** when the wrapped sink cancels
+
+@@@

docs/src/main/paradox/stream/operators/index.md

@@ -87,6 +87,7 @@ These built-in sinks are available from @scala[`org.apache.pekko.stream.scaladsl
 |Sink|<a name="seq"></a>@ref[seq](Sink/seq.md)|Collect values emitted from the stream into a collection.|
 |Sink|<a name="source"></a>@ref[source](Sink/source.md)|A `Sink` that materializes this `Sink` itself as a `Source`, the returning `Source` can only have one subscriber.|
 |Sink|<a name="takelast"></a>@ref[takeLast](Sink/takeLast.md)|Collect the last `n` values emitted from the stream into a collection.|
+|Sink|<a name="watchtermination"></a>@ref[watchTermination](Sink/watchTermination.md)|Wraps a sink so that in addition to the original materialized value a @scala[`Future[Done]`] @java[`CompletionStage<Done>`] is materialized that only completes after the wrapped sink has fully terminated, including its `postStop` lifecycle hook.|
 
 ## Additional Sink and Source converters
 
@@ -615,6 +616,7 @@ For more background see the @ref[Error Handling in Streams](../stream-error.md)
 * [UnzipWith](UnzipWith.md)
 * [watch](Source-or-Flow/watch.md)
 * [watchTermination](Source-or-Flow/watchTermination.md)
+* [watchTermination](Sink/watchTermination.md)
 * [wireTap](Source-or-Flow/wireTap.md)
 * [withBackoff](RestartSource/withBackoff.md)
 * [withBackoff](RestartFlow/withBackoff.md)

docs/src/test/java/jdocs/stream/operators/sink/WatchTermination.java

@@ -0,0 +1,54 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package jdocs.stream.operators.sink;
+
+import java.nio.file.Paths;
+import java.util.concurrent.CompletionStage;
+import org.apache.pekko.Done;
+import org.apache.pekko.actor.ActorSystem;
+import org.apache.pekko.japi.Pair;
+import org.apache.pekko.stream.IOResult;
+import org.apache.pekko.stream.javadsl.FileIO;
+import org.apache.pekko.stream.javadsl.Keep;
+import org.apache.pekko.stream.javadsl.Sink;
+import org.apache.pekko.stream.javadsl.Source;
+import org.apache.pekko.util.ByteString;
+
+public class WatchTermination {
+
+  private ActorSystem system = null;
+
+  void example() {
+    // #watchTermination
+    final Sink<ByteString, CompletionStage<IOResult>> fileSink =
+        FileIO.toPath(Paths.get("target/watch-termination.txt"));
+
+    // In addition to the IOResult of the file sink, materialize a CompletionStage<Done>
+    // that only completes once the file has been fully written and closed.
+    final Pair<CompletionStage<IOResult>, CompletionStage<Done>> result =
+        Source.single(ByteString.fromString("Hello, world!"))
+            .runWith(fileSink.watchTermination(Keep.both()), system);
+
+    final CompletionStage<IOResult> ioResult = result.first();
+    final CompletionStage<Done> terminated = result.second();
+
+    // Once `terminated` completes the sink has stopped, including its postStop
+    // cleanup, so the file is guaranteed to be closed at this point.
+    // #watchTermination
+  }
+}

docs/src/test/scala/docs/stream/operators/sink/WatchTermination.scala

@@ -0,0 +1,50 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package docs.stream.operators.sink
+
+import java.nio.file.Paths
+
+import scala.concurrent.Future
+
+import org.apache.pekko
+import pekko.Done
+import pekko.actor.ActorSystem
+import pekko.stream.IOResult
+import pekko.stream.scaladsl.{ FileIO, Keep, Sink, Source }
+import pekko.util.ByteString
+
+object WatchTermination {
+  implicit val system: ActorSystem = ???
+
+  def watchTerminationExample(): Unit = {
+    // #watchTermination
+    val fileSink: Sink[ByteString, Future[IOResult]] =
+      FileIO.toPath(Paths.get("target/watch-termination.txt"))
+
+    // In addition to the IOResult of the file sink, materialize a Future[Done]
+    // that only completes once the file has been fully written and closed.
+    val (ioResult, terminated): (Future[IOResult], Future[Done]) =
+      Source
+        .single(ByteString("Hello, world!"))
+        .runWith(fileSink.watchTermination(Keep.both))
+
+    // Once `terminated` completes the sink has stopped, including its postStop
+    // cleanup, so the file is guaranteed to be closed at this point.
+    // #watchTermination
+  }
+}

stream-tests/src/test/java/org/apache/pekko/stream/javadsl/SinkTest.java

@@ -278,4 +278,20 @@ public void mustBeAbleToUseSinkAsSource() throws Exception {
             .get(1, TimeUnit.SECONDS);
     assertEquals(List.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10), r);
   }
+
+  @Test
+  public void mustBeAbleToUseWatchTermination() throws Exception {
+    final Pair<CompletionStage<Integer>, CompletionStage<Done>> result =
+        Source.range(1, 4).runWith(Sink.<Integer>head().watchTermination(Keep.both()), system);
+    assertEquals(1, result.first().toCompletableFuture().get(1, TimeUnit.SECONDS).intValue());
+    assertEquals(Done.done(), result.second().toCompletableFuture().get(1, TimeUnit.SECONDS));
+  }
+
+  @Test
+  public void watchTerminationMustWorkWithCompositeSinks() throws Exception {
+    final CompletionStage<Done> done =
+        Source.range(1, 4)
+            .runWith(Sink.<Integer>foreach(x -> {}).watchTermination(Keep.right()), system);
+    assertEquals(Done.done(), done.toCompletableFuture().get(1, TimeUnit.SECONDS));
+  }
 }