You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pekko.apache.org by he...@apache.org on 2026/06/30 06:13:40 UTC
(pekko) branch port/b7e625fb12-prematerialize-caveat created (now 0b66e4ea16)
This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a change to branch port/b7e625fb12-prematerialize-caveat
in repository https://gitbox.apache.org/repos/asf/pekko.git
at 0b66e4ea16 docs: document preMaterialize error-propagation caveat
This branch includes the following new commits:
new 0b66e4ea16 docs: document preMaterialize error-propagation caveat
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org
(pekko) 01/01: docs: document preMaterialize error-propagation caveat
Posted by he...@apache.org.
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) <he...@gmail.com>
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: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org