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: 虎鸣 <[email protected]> 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: [email protected] For additional commands, e-mail: [email protected]
