feat: Add asJavaStream to Sink

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

@@ -0,0 +1,38 @@
+# Sink.asJavaStream
+
+Create a sink which materializes into Java 8 `Stream` that can be run to trigger demand through the sink.
+
+@ref[Sink operators](../index.md#sink-operators)
+
+## Signature
+
+@apidoc[Sink.asJavaStream](Sink$) { scala="#asJavaStream[T]():org.apache.pekko.stream.scaladsl.Sink[T,java.util.stream.Stream[T]]" java="#asJavaStream()" }
+
+## Description
+
+Create a sink which materializes into Java 8 `Stream` that can be run to trigger demand through the sink.
+Elements emitted through the stream will be available for reading through the Java 8 `Stream`.
+
+The Java 8 `Stream` will be ended when the stream flowing into this `Sink` completes, and closing the Java
+`Stream` will cancel the inflow of this `Sink`. If the Java `Stream` throws an exception, the Pekko stream is cancelled.
+
+Be aware that Java `Stream` blocks current thread while waiting on next element from downstream.
+
+## Example
+
+Here is an example of a @apidoc[Sink] that materializes into a @javadoc[java.util.stream.Stream](java.util.stream.Stream). 
+
+Scala
+:   @@snip [StreamConvertersToJava.scala](/docs/src/test/scala/docs/stream/operators/converters/StreamConvertersToJava.scala) { #import #asJavaStreamOnSink }
+
+Java
+:   @@snip [StreamConvertersToJava.java](/docs/src/test/java/jdocs/stream/operators/converters/StreamConvertersToJava.java) { #import #asJavaStreamOnSink }
+
+
+## Reactive Streams semantics
+
+@@@div { .callout }
+**cancels** when the Java Stream is closed
+
+**backpressures** when no read is pending on the Java Stream
+@@@

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

@@ -52,6 +52,7 @@ These built-in sinks are available from @scala[`org.apache.pekko.stream.scaladsl
 
 | |Operator|Description|
 |--|--|--|
+|Sink|<a name="asjavastream"></a>@ref[asJavaStream](Sink/asJavaStream.md)|Create a sink which materializes into Java 8 `Stream` that can be run to trigger demand through the sink.|
 |Sink|<a name="aspublisher"></a>@ref[asPublisher](Sink/asPublisher.md)|Integration with Reactive Streams, materializes into a `org.reactivestreams.Publisher`.|
 |Sink|<a name="cancelled"></a>@ref[cancelled](Sink/cancelled.md)|Immediately cancel the stream|
 |Sink|<a name="collect"></a>@ref[collect](Sink/collect.md)|Collect all input elements using a Java @javadoc[Collector](java.util.stream.Collector).|
@@ -407,6 +408,7 @@ For more background see the @ref[Error Handling in Streams](../stream-error.md)
 * [alsoToAll](Source-or-Flow/alsoToAll.md)
 * [asFlowWithContext](Flow/asFlowWithContext.md)
 * [asInputStream](StreamConverters/asInputStream.md)
+* [asJavaStream](Sink/asJavaStream.md)
 * [asJavaStream](StreamConverters/asJavaStream.md)
 * [ask](Source-or-Flow/ask.md)
 * [ask](ActorFlow/ask.md)

docs/src/test/java/jdocs/stream/operators/converters/StreamConvertersToJava.java

@@ -66,6 +66,17 @@ public void demonstrateConverterToJava8Stream() {
     assertEquals(5, jStream.count());
   }
 
+  @Test
+  public void demonstrateConverterToJava8StreamOnSink() {
+    // #asJavaStreamOnSink
+
+    Source<Integer, NotUsed> source = Source.range(0, 9).filter(i -> i % 2 == 0);
+    Stream<Integer> jStream = source.runWith(Sink.asJavaStream(), system);
+
+    // #asJavaStreamOnSink
+    assertEquals(5, jStream.count());
+  }
+
   @Test
   public void demonstrateCreatingASourceFromJava8Stream()
       throws InterruptedException, ExecutionException, TimeoutException {

docs/src/test/scala/docs/stream/operators/converters/StreamConvertersToJava.scala

@@ -43,6 +43,15 @@ class StreamConvertersToJava extends PekkoSpec with Futures {
     jStream.count should be(5)
   }
 
+  "demonstrate materialization to Java8 streams with methods on Sink" in {
+    // #asJavaStreamOnSink
+    val source: Source[Int, NotUsed] = Source(0 to 9).filter(_ % 2 == 0)
+
+    val jStream: java.util.stream.Stream[Int] = source.runWith(Sink.asJavaStream[Int]())
+    // #asJavaStreamOnSink
+    jStream.count should be(5)
+  }
+
   "demonstrate conversion from Java8 streams" in {
     // #fromJavaStream
     def factory(): IntStream = IntStream.rangeClosed(0, 9)

stream-tests/src/test/java/org/apache/pekko/stream/io/SinkAsJavaSourceTest.java

@@ -43,4 +43,12 @@ public void mustBeAbleToUseAsJavaStream() throws Exception {
     java.util.stream.Stream<Integer> javaStream = Source.from(list).runWith(streamSink, system);
     assertEquals(list, javaStream.toList());
   }
+
+  @Test
+  public void mustBeAbleToUseAsJavaStreamOnSink() throws Exception {
+    final List<Integer> list = Arrays.asList(1, 2, 3);
+    java.util.stream.Stream<Integer> javaStream =
+        Source.from(list).runWith(Sink.asJavaStream(), system);
+    assertEquals(list, javaStream.collect(Collectors.toList()));
+  }
 }

stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/SinkAsJavaStreamSpec.scala

@@ -32,6 +32,9 @@ class SinkAsJavaStreamSpec extends StreamSpec(UnboundedMailboxConfig) {
     "work in happy case" in {
       val javaSource = Source(1 to 100).runWith(StreamConverters.asJavaStream())
       javaSource.count() should ===(100L)
+      //
+      Source(1 to 100).runWith(Sink.asJavaStream())
+        .count() should ===(100L)
     }
 
     "fail if parent stream is failed" in {

stream/src/main/scala/org/apache/pekko/stream/impl/fusing/GraphInterpreter.scala

@@ -342,12 +342,25 @@ import pekko.stream.stage._
     var i = 0
     while (i < logics.length) {
       val logic = logics(i)
-      if (!isStageCompleted(logic) && !isStageFinalized(logic)) {
+      if ((logic ne null) && !isStageCompleted(logic) && !isStageFinalized(logic)) {
         markStageFinalized(logic)
         finalizeStage(logic)
       }
+      // Release reference to the stage logic so it can be garbage collected
+      // even if the GraphInterpreter is still alive due to other references
+      logics(i) = null
       i += 1
     }
+    // Null connection handlers for all finalized stages to break handler-to-logic reference chains
+    var j = 0
+    while (j < connections.length) {
+      val conn = connections(j)
+      if (conn ne null) {
+        conn.inHandler = null
+        conn.outHandler = null
+      }
+      j += 1
+    }
   }
 
   // Debug name for a connections input part
@@ -357,10 +370,16 @@ import pekko.stream.stage._
   private def outOwnerName(connection: Connection): String = connection.outOwner.toString
 
   // Debug name for a connections input part
-  private def inLogicName(connection: Connection): String = logics(connection.inOwner.stageId).toString
+  private def inLogicName(connection: Connection): String = {
+    val logic = logics(connection.inOwner.stageId)
+    if (logic ne null) logic.toString else "<completed>"
+  }
 
   // Debug name for a connections output part
-  private def outLogicName(connection: Connection): String = logics(connection.outOwner.stageId).toString
+  private def outLogicName(connection: Connection): String = {
+    val logic = logics(connection.outOwner.stageId)
+    if (logic ne null) logic.toString else "<completed>"
+  }
 
   private def shutdownCounters: String =
     shutdownCounter.map(x => if (x >= KeepGoingFlag) s"${x & KeepGoingMask}(KeepGoing)" else x.toString).mkString(",")

stream/src/main/scala/org/apache/pekko/stream/javadsl/Sink.scala

@@ -228,6 +228,23 @@ object Sink {
   def source[T](): Sink[T, Source[T, NotUsed]] = new Sink(scaladsl.Sink.source[T])
     .mapMaterializedValue(src => src.asJava)
 
+  /**
+   * Creates a sink which materializes into Java 8 ``Stream`` that can be run to trigger demand through the sink.
+   * Elements emitted through the stream will be available for reading through the Java 8 ``Stream``.
+   *
+   * The Java 8 ``Stream`` will be ended when the stream flowing into this ``Sink`` completes, and closing the Java
+   * ``Stream`` will cancel the inflow of this ``Sink``.
+   *
+   * Java 8 ``Stream`` throws exception in case reactive stream failed.
+   *
+   * Be aware that Java ``Stream`` blocks current thread while waiting on next element from downstream.
+   * As it is interacting with blocking API the implementation runs on a separate dispatcher
+   * configured through the ``pekko.stream.blocking-io-dispatcher``.
+   *
+   * @since 2.0.0
+   */
+  def asJavaStream[T](): Sink[T, java.util.stream.Stream[T]] = new Sink(scaladsl.StreamConverters.asJavaStream())
+
   /**
    * A `Sink` that will invoke the given procedure for each received element. The sink is materialized
    * into a [[java.util.concurrent.CompletionStage]] which will be completed with `Success` when reaching the

stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala

@@ -323,6 +323,23 @@ object Sink {
 
   private val _sourceSink = fromGraph(SourceSink)
 
+  /**
+   * Creates a sink which materializes into Java 8 ``Stream`` that can be run to trigger demand through the sink.
+   * Elements emitted through the stream will be available for reading through the Java 8 ``Stream``.
+   *
+   * The Java 8 ``Stream`` will be ended when the stream flowing into this ``Sink`` completes, and closing the Java
+   * ``Stream`` will cancel the inflow of this ``Sink``.
+   *
+   * If the Java 8 ``Stream`` throws exception the Pekko stream is cancelled.
+   *
+   * Be aware that Java ``Stream`` blocks current thread while waiting on next element from downstream.
+   * As it is interacting with blocking API the implementation runs on a separate dispatcher
+   * configured through the ``pekko.stream.blocking-io-dispatcher``.
+   *
+   * @since 2.0.0
+   */
+  def asJavaStream[T](): Sink[T, java.util.stream.Stream[T]] = StreamConverters.asJavaStream()
+
   /**
    * A `Sink` that will consume the stream and discard the elements.
    */