This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/pekko.git
The following commit(s) were added to refs/heads/main by this push:
new 5d1b67dedf docs: document preMaterialize error-propagation caveat
(#3272)
5d1b67dedf is described below
commit 5d1b67dedfe0d947d1d69bec5ba3c78efca1d5e6
Author: He-Pin(kerr) <[email protected]>
AuthorDate: Tue Jun 30 19:56:38 2026 +0800
docs: document preMaterialize error-propagation caveat (#3272)
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]
