This is an automated email from the ASF dual-hosted git repository. He-Pin pushed a commit to branch port/b7e625fb12-prematerialize-caveat in repository https://gitbox.apache.org/repos/asf/pekko.git
commit 0b66e4ea1624784f83a08f3807f743b96c052362 Author: He-Pin(kerr) <[email protected]> AuthorDate: Tue Jun 30 14:07:11 2026 +0800 docs: document preMaterialize error-propagation caveat Motivation: preMaterialize crosses a reactive-streams Publisher/Subscriber boundary, which introduces a buffer and turns upstream failures into cancellations without error details. This surprising behaviour was undocumented. Modification: Add the caveat to the scaladoc of every preMaterialize overload on Source/Sink/Flow in both the Scala and Java DSLs, and to the Sink and Source-or-Flow paradox operator pages. For consistency the caveat is also added to the Java DSL `materializer` overloads of Sink/Source, which have no direct upstream counterpart. Result: Users are clearly warned that preMaterialize buffers elements and does not propagate upstream errors. Tests: Not run - docs only References: Upstream: akka/akka-core@b7e625fb1274323afb9e8c5f0c2f4c9560cc4da1 (akka#31982), which is now Apache licensed. --- docs/src/main/paradox/stream/operators/Sink/preMaterialize.md | 3 +++ .../paradox/stream/operators/Source-or-Flow/preMaterialize.md | 4 ++++ stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala | 8 ++++++++ stream/src/main/scala/org/apache/pekko/stream/javadsl/Sink.scala | 6 ++++++ .../src/main/scala/org/apache/pekko/stream/javadsl/Source.scala | 6 ++++++ stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala | 4 ++++ stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala | 3 +++ .../src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala | 3 +++ 8 files changed, 37 insertions(+) diff --git a/docs/src/main/paradox/stream/operators/Sink/preMaterialize.md b/docs/src/main/paradox/stream/operators/Sink/preMaterialize.md index e7b9790915..c421788d2c 100644 --- a/docs/src/main/paradox/stream/operators/Sink/preMaterialize.md +++ b/docs/src/main/paradox/stream/operators/Sink/preMaterialize.md @@ -13,3 +13,6 @@ Materializes this Sink, immediately returning (1) its materialized value, and (2 Materializes this Sink, immediately returning (1) its materialized value, and (2) a new Sink that can be consume elements 'into' the pre-materialized one. Useful for when you need a materialized value of a Sink when handing it out to someone to materialize it for you. +Note that `preMaterialize` is implemented through a reactive streams `Subscriber` which means that a buffer is introduced +and that errors are not propagated upstream but are turned into cancellations without error details. + diff --git a/docs/src/main/paradox/stream/operators/Source-or-Flow/preMaterialize.md b/docs/src/main/paradox/stream/operators/Source-or-Flow/preMaterialize.md index a06eed6c9d..9ba70713f9 100644 --- a/docs/src/main/paradox/stream/operators/Source-or-Flow/preMaterialize.md +++ b/docs/src/main/paradox/stream/operators/Source-or-Flow/preMaterialize.md @@ -14,3 +14,7 @@ Materializes this Graph, immediately returning (1) its materialized value, and ( Materializes this Graph, immediately returning (1) its materialized value, and (2) a new pre-materialized Graph. +Note that `preMaterialize` is implemented through a reactive streams `Publisher` for a `Source` or a `Publisher` +and `Subscriber` pair for a `Flow` which means that a buffer is introduced and that errors are not propagated upstream +but are turned into cancellations without error details. + diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala index 525a0e25a7..0b80d72ebb 100755 --- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Flow.scala @@ -389,6 +389,10 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr /** * Materializes this [[Flow]], immediately returning (1) its materialized value, and (2) a newly materialized [[Flow]]. + * + * Note that `preMaterialize` is implemented through a reactive streams `Publisher` and `Subscriber` pair which means + * that a buffer is introduced and that errors are not propagated upstream but are turned into cancellations without + * error details. */ def preMaterialize( systemProvider: ClassicActorSystemProvider): pekko.japi.Pair[Mat @uncheckedVariance, Flow[In, Out, NotUsed]] = { @@ -398,6 +402,10 @@ final class Flow[In, Out, Mat](delegate: scaladsl.Flow[In, Out, Mat]) extends Gr /** * Materializes this [[Flow]], immediately returning (1) its materialized value, and (2) a newly materialized [[Flow]]. * The returned flow is partial materialized and do not support multiple times materialization. + * + * Note that `preMaterialize` is implemented through a reactive streams `Publisher` and `Subscriber` pair which means + * that a buffer is introduced and that errors are not propagated upstream but are turned into cancellations without + * error details. */ def preMaterialize(materializer: Materializer): pekko.japi.Pair[Mat @uncheckedVariance, Flow[In, Out, NotUsed]] = { val (mat, flow) = delegate.preMaterialize()(materializer) diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Sink.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Sink.scala index f9baca42ec..fbf1becfbd 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Sink.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Sink.scala @@ -631,6 +631,9 @@ final class Sink[In, Mat](delegate: scaladsl.Sink[In, Mat]) extends Graph[SinkSh * Useful for when you need a materialized value of a Sink when handing it out to someone to materialize it for you. * * Note that the `ActorSystem` can be used as the `systemProvider` parameter. + * + * Note that `preMaterialize` is implemented through a reactive streams `Subscriber` which means that a buffer is + * introduced and that errors are not propagated upstream but are turned into cancellations without error details. */ def preMaterialize(systemProvider: ClassicActorSystemProvider) : japi.Pair[Mat @uncheckedVariance, Sink[In @uncheckedVariance, NotUsed]] = { @@ -645,6 +648,9 @@ final class Sink[In, Mat](delegate: scaladsl.Sink[In, Mat]) extends Graph[SinkSh * Useful for when you need a materialized value of a Sink when handing it out to someone to materialize it for you. * * Prefer the method taking an ActorSystem unless you have special requirements. + * + * Note that `preMaterialize` is implemented through a reactive streams `Subscriber` which means that a buffer is + * introduced and that errors are not propagated upstream but are turned into cancellations without error details. */ def preMaterialize( materializer: Materializer): japi.Pair[Mat @uncheckedVariance, Sink[In @uncheckedVariance, NotUsed]] = { diff --git a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala index 9893ffdcd9..ae3a668983 100755 --- a/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/javadsl/Source.scala @@ -890,6 +890,9 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[ * that can be used to consume elements from the newly materialized Source. * * Note that the `ActorSystem` can be used as the `systemProvider` parameter. + * + * Note that `preMaterialize` is implemented through a reactive streams `Publisher` which means that a buffer is + * introduced and that errors are not propagated upstream but are turned into cancellations without error details. */ def preMaterialize(systemProvider: ClassicActorSystemProvider) : Pair[Mat @uncheckedVariance, Source[Out @uncheckedVariance, NotUsed]] = { @@ -902,6 +905,9 @@ final class Source[Out, Mat](delegate: scaladsl.Source[Out, Mat]) extends Graph[ * that can be used to consume elements from the newly materialized Source. * * Prefer the method taking an `ActorSystem` unless you have special requirements. + * + * Note that `preMaterialize` is implemented through a reactive streams `Publisher` which means that a buffer is + * introduced and that errors are not propagated upstream but are turned into cancellations without error details. */ def preMaterialize( materializer: Materializer): Pair[Mat @uncheckedVariance, Source[Out @uncheckedVariance, NotUsed]] = { diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala index f9addecd4a..70a8f263ec 100755 --- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Flow.scala @@ -173,6 +173,10 @@ final class Flow[-In, +Out, +Mat]( /** * Materializes this [[Flow]], immediately returning (1) its materialized value, and (2) a newly materialized [[Flow]]. * The returned flow is partial materialized and do not support multiple times materialization. + * + * Note that `preMaterialize` is implemented through a reactive streams `Publisher` and `Subscriber` pair which means + * that a buffer is introduced and that errors are not propagated upstream but are turned into cancellations without + * error details. */ def preMaterialize()(implicit materializer: Materializer): (Mat, ReprMat[Out, NotUsed]) = { val ((sub, mat), pub) = diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala index c61c7959ea..a020284b63 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Sink.scala @@ -73,6 +73,9 @@ final class Sink[-In, +Mat](override val traversalBuilder: LinearTraversalBuilde * that can be consume elements 'into' the pre-materialized one. * * Useful for when you need a materialized value of a Sink when handing it out to someone to materialize it for you. + * + * Note that `preMaterialize` is implemented through a reactive streams `Subscriber` which means that a buffer is + * introduced and that errors are not propagated upstream but are turned into cancellations without error details. */ def preMaterialize()(implicit materializer: Materializer): (Mat, Sink[In, NotUsed]) = { val (sub, mat) = Source.asSubscriber.toMat(this)(Keep.both).run() diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala index 8b4e60d539..6dab7058ab 100644 --- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala +++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Source.scala @@ -111,6 +111,9 @@ final class Source[+Out, +Mat]( /** * Materializes this Source, immediately returning (1) its materialized value, and (2) a new Source * that can be used to consume elements from the newly materialized Source. + * + * Note that `preMaterialize` is implemented through a reactive streams `Publisher` which means that a buffer is + * introduced and that errors are not propagated upstream but are turned into cancellations without error details. */ def preMaterialize()(implicit materializer: Materializer): (Mat, ReprMat[Out, NotUsed]) = { val (mat, pub) = toMat(Sink.asPublisher(fanout = true))(Keep.both).run() --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
