From c751b6f102d1ff00e2a2010133f29d0090eab58a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=99=8E=E9=B8=A3?= Date: Tue, 14 Jul 2026 18:45:44 +0800 Subject: [PATCH 1/3] fix(stream): notify BroadcastHub consumers on materializer shutdown Motivation: When the materializer running a BroadcastHub.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 (see #3345). Modification: Move registered-consumer (consumerWheel) 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: BroadcastHub 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: Fixes #3345 --- .../pekko/stream/scaladsl/HubSpec.scala | 24 +++++++++++- .../apache/pekko/stream/scaladsl/Hub.scala | 37 ++++++++++++------- 2 files changed, 47 insertions(+), 14 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..628585cca0 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 @@ -559,6 +559,28 @@ class HubSpec extends StreamSpec { } } + "notify consumers when hub materializer is shut down" in { + val hubMat = Materializer(system) + val upstream = TestPublisher.probe[Int]() + val source = Source.fromPublisher(upstream).runWith(BroadcastHub.sink(8))(hubMat) + + val downstream = TestSubscriber.probe[Int]() + source.runWith(Sink.fromSubscriber(downstream)) + + downstream.request(1) + Thread.sleep(100) + + upstream.sendNext(1) + downstream.expectNext(1) + + hubMat.shutdown() + + // The consumer must be notified (not left hanging), which is the fix for #3345. + // The signal is an error because the SubSink output boundary failure + // arrives before the postStop completion callback. + downstream.expectError() + } + "handle cancelled Sink" in { val in = TestPublisher.probe[Int]() val hubSource = Source.fromPublisher(in).runWith(BroadcastHub.sink(4)) 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..f4ce5cdebd 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 @@ -652,21 +652,12 @@ private[pekko] class BroadcastHub[T](startAfterNrOfConsumers: Int, bufferSize: I 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 in the consumerWheel are notified by postStop. state.getAndSet(Closed(Some(ex))).asInstanceOf[Open].registrations.foreach { consumer => consumer.callback.invoke(failMessage) } - // Notify registered consumers — skip null (empty) slots - var idx = 0 - while (idx < consumerWheel.length) { - val bucket = consumerWheel(idx) - if (bucket ne null) { - val itr = bucket.valuesIterator - while (itr.hasNext) itr.next().callback.invoke(failMessage) - } - idx += 1 - } failStage(ex) } @@ -759,19 +750,39 @@ private[pekko] class BroadcastHub[T](startAfterNrOfConsumers: Int, bufferSize: I } override def postStop(): Unit = { - // Notify pending consumers and set tombstone + // Notify all consumers (pending and registered) when the stage stops. + // Registered consumers in the consumerWheel 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 wheel 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) + } + var idx = 0 + while (idx < consumerWheel.length) { + val bucket = consumerWheel(idx) + if (bucket ne null) { + val itr = bucket.valuesIterator + while (itr.hasNext) itr.next().callback.invoke(message) + } + idx += 1 + } + } + tryClose() } From c0aef075275790cd9b791e978824e09a7662be08 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=99=8E=E9=B8=A3?= Date: Tue, 14 Jul 2026 19:15:42 +0800 Subject: [PATCH 2/3] fix(stream): guard BroadcastHub consumer notifications with NonFatal Motivation: If one consumer's callback throws during hub shutdown, the remaining consumers must still be notified. Modification: Wrap each callback.invoke in onUpstreamFailure and postStop with try/catch NonFatal so a single consumer failure does not short-circuit the notification loop. Result: All reachable consumers are notified even when one consumer's callback throws. Tests: - sbt "stream-tests / Test / testOnly org.apache.pekko.stream.scaladsl.HubSpec" 51 passed, 0 failed References: Fixes #3345 --- .../scala/org/apache/pekko/stream/scaladsl/Hub.scala | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) 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 f4ce5cdebd..8740ece508 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 @@ -25,6 +25,7 @@ import scala.collection.immutable.Queue import scala.collection.mutable.LongMap import scala.concurrent.{ Future, Promise } import scala.util.{ Failure, Success, Try } +import scala.util.control.NonFatal import org.apache.pekko import pekko.NotUsed @@ -655,7 +656,8 @@ private[pekko] class BroadcastHub[T](startAfterNrOfConsumers: Int, bufferSize: I // Notify pending consumers and set tombstone. // Registered consumers in the consumerWheel are notified by postStop. state.getAndSet(Closed(Some(ex))).asInstanceOf[Open].registrations.foreach { consumer => - consumer.callback.invoke(failMessage) + try consumer.callback.invoke(failMessage) + catch { case NonFatal(_) => } } failStage(ex) @@ -761,7 +763,8 @@ private[pekko] class BroadcastHub[T](startAfterNrOfConsumers: Int, bufferSize: I if (state.compareAndSet(open, Closed(None))) { val completedMessage = HubCompleted(None) open.registrations.foreach { consumer => - consumer.callback.invoke(completedMessage) + try consumer.callback.invoke(completedMessage) + catch { case NonFatal(_) => } } notifyRegisteredConsumers() } else tryClose() @@ -777,7 +780,10 @@ private[pekko] class BroadcastHub[T](startAfterNrOfConsumers: Int, bufferSize: I val bucket = consumerWheel(idx) if (bucket ne null) { val itr = bucket.valuesIterator - while (itr.hasNext) itr.next().callback.invoke(message) + while (itr.hasNext) { + try itr.next().callback.invoke(message) + catch { case NonFatal(_) => } + } } idx += 1 } From b273fc0a58bd50f8787edc369d261b35cba0f839 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E8=99=8E=E9=B8=A3?= Date: Tue, 14 Jul 2026 19:28:20 +0800 Subject: [PATCH 3/3] fix(stream): improve BroadcastHub postStop comment accuracy and test determinism MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Motivation: The postStop method's Closed case comment incorrectly said "fall through" but Scala match cases never fall through. The test used Thread.sleep which violates the project's determinism guidelines. Modification: - Fix misleading comment in postStop's Closed case to say "notify registered consumers directly" instead of "fall through to notify wheel below" - Replace Thread.sleep(100) in the materializer shutdown test with upstream.expectRequest() for deterministic demand propagation Result: Comment accurately describes the code behavior. Test no longer relies on Thread.sleep for setup timing. Tests: - sbt "stream-tests / Test / testOnly org.apache.pekko.stream.scaladsl.HubSpec" — 51/51 passed References: Fixes #3345 --- .../test/scala/org/apache/pekko/stream/scaladsl/HubSpec.scala | 2 +- .../src/main/scala/org/apache/pekko/stream/scaladsl/Hub.scala | 2 +- 2 files changed, 2 insertions(+), 2 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 628585cca0..954c592da9 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 @@ -568,7 +568,7 @@ class HubSpec extends StreamSpec { source.runWith(Sink.fromSubscriber(downstream)) downstream.request(1) - Thread.sleep(100) + upstream.expectRequest() upstream.sendNext(1) downstream.expectNext(1) 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 8740ece508..ca7615a55f 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 @@ -757,7 +757,7 @@ private[pekko] class BroadcastHub[T](startAfterNrOfConsumers: Int, bufferSize: I // so that materializer shutdown produces the correct signal. @tailrec def tryClose(): Unit = state.get() match { - case Closed(_) => // Already closed by onUpstreamFailure — fall through to notify wheel below + case Closed(_) => // Already closed by onUpstreamFailure — notify registered consumers directly notifyRegisteredConsumers() case open: Open => if (state.compareAndSet(open, Closed(None))) {