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/07/14 10:49:36 UTC
(pekko) branch fix/partition-hub-poststop created (now f8e7669caa)
This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a change to branch fix/partition-hub-poststop
in repository https://gitbox.apache.org/repos/asf/pekko.git
at f8e7669caa fix(stream): notify PartitionHub consumers on materializer shutdown
This branch includes the following new commits:
new f8e7669caa fix(stream): notify PartitionHub consumers on materializer shutdown
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: fix(stream): notify PartitionHub consumers on materializer shutdown
Posted by he...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
He-Pin pushed a commit to branch fix/partition-hub-poststop
in repository https://gitbox.apache.org/repos/asf/pekko.git
commit f8e7669caa34135cafd7e4fef264780a89562eb6
Author: 虎鸣 <he...@alibaba-inc.com>
AuthorDate: Tue Jul 14 18:49:18 2026 +0800
fix(stream): notify PartitionHub consumers on materializer shutdown
Motivation:
When the materializer running a PartitionHub.sink is shut down (e.g. the
hosting actor stops), consumers materialized on a different materializer
may not be notified of the hub's termination. This mirrors the same bug
in BroadcastHub (see #3345).
Modification:
Move registered-consumer notification from onUpstreamFailure to postStop
so postStop is the single notification point. postStop now handles both
Open (normal termination) and Closed (after upstream failure) states,
ensuring registered consumers always receive the appropriate signal.
Result:
PartitionHub consumers are reliably notified when the hub's materializer
shuts down, preventing consumers from hanging indefinitely.
Tests:
- sbt "stream-tests / Test / testOnly org.apache.pekko.stream.scaladsl.HubSpec"
51 passed, 0 failed
References:
Refs #3345
---
.../org/apache/pekko/stream/scaladsl/HubSpec.scala | 22 ++++++++++++++++++-
.../org/apache/pekko/stream/scaladsl/Hub.scala | 25 ++++++++++++++++------
2 files changed, 39 insertions(+), 8 deletions(-)
diff --git a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/HubSpec.scala b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/HubSpec.scala
index b32c8746b1..0479ce9563 100644
--- a/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/HubSpec.scala
+++ b/stream-tests/src/test/scala/org/apache/pekko/stream/scaladsl/HubSpec.scala
@@ -19,7 +19,7 @@ import scala.concurrent.duration._
import org.apache.pekko
import pekko.Done
-import pekko.stream.KillSwitches
+import pekko.stream.{ KillSwitches, Materializer }
import pekko.stream.ThrottleMode
import pekko.stream.impl.ActorPublisher
import pekko.stream.testkit.StreamSpec
@@ -946,6 +946,26 @@ class HubSpec extends StreamSpec {
result2.futureValue should ===((1 to 9 by 2).filterNot(_ == 3))
}
+ "notify consumers when hub materializer is shut down" in {
+ val hubMat = Materializer(system)
+ val upstream = TestPublisher.probe[Int]()
+ val source = Source.fromPublisher(upstream).runWith(
+ PartitionHub.sink((size, elem) => elem % size, startAfterNrOfConsumers = 1, bufferSize = 8))(hubMat)
+
+ val downstream = TestSubscriber.probe[Int]()
+ source.runWith(Sink.fromSubscriber(downstream))
+
+ downstream.request(1)
+ Thread.sleep(100)
+
+ upstream.sendNext(0)
+ downstream.expectNext(0)
+
+ hubMat.shutdown()
+
+ downstream.expectError()
+ }
+
}
}
diff --git a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Hub.scala b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Hub.scala
index 32e647841f..df02da4237 100644
--- a/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Hub.scala
+++ b/stream/src/main/scala/org/apache/pekko/stream/scaladsl/Hub.scala
@@ -1357,32 +1357,43 @@ object PartitionHub {
override def onUpstreamFailure(ex: Throwable): Unit = {
val failMessage = HubCompleted(Some(ex))
- // Notify pending consumers and set tombstone
+ // Notify pending consumers and set tombstone.
+ // Registered consumers are notified by postStop.
state.getAndSet(Closed(Some(ex))).asInstanceOf[Open].registrations.foreach { consumer =>
consumer.callback.invoke(failMessage)
}
- // Notify registered consumers
- consumerInfo.consumers.foreach { consumer =>
- consumer.callback.invoke(failMessage)
- }
failStage(ex)
}
override def postStop(): Unit = {
- // Notify pending consumers and set tombstone
+ // Notify all consumers (pending and registered) when the stage stops.
+ // Registered consumers are notified here (not in onUpstreamFailure)
+ // so that materializer shutdown produces the correct signal.
@tailrec def tryClose(): Unit = state.get() match {
- case Closed(_) => // Already closed, ignore
+ case Closed(_) => // Already closed by onUpstreamFailure — fall through to notify registered below
+ notifyRegisteredConsumers()
case open: Open =>
if (state.compareAndSet(open, Closed(None))) {
val completedMessage = HubCompleted(None)
open.registrations.foreach { consumer =>
consumer.callback.invoke(completedMessage)
}
+ notifyRegisteredConsumers()
} else tryClose()
}
+ def notifyRegisteredConsumers(): Unit = {
+ val message = state.get() match {
+ case Closed(Some(ex)) => HubCompleted(Some(ex))
+ case _ => HubCompleted(None)
+ }
+ consumerInfo.consumers.foreach { consumer =>
+ consumer.callback.invoke(message)
+ }
+ }
+
tryClose()
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pekko.apache.org
For additional commands, e-mail: commits-help@pekko.apache.org