Summary
When materializing a BroadcastHub on an actor's materializer, stopping/crashing the actor does not complete the BroadcastHub Source.
Analysis
Two independent actors are running in this situation:
The first ("Worker") is materializing a Source via a BroadcastHub.sink, running with a Materializer bound to the actor. The source created by the BroadcastHub is then passed to a different actor ("Listener") that runs it to itself using an ActorSink.
When the Worker stops (or is crashed), the source created by the BroadcastHub is not terminated (i.e. the Listener does not receive a Completed/Error message).
The source that runs into the BroadcastHub is cancelled, as expected.
If the source that runs the BroadcastHub is completed/failed manually, the BroadcastHub source completes as expected.
The reproducer has a third actor that sends a Crash/Stop message to the Worker after a couple of seconds.
Reproducer
import org.apache.pekko.NotUsed
import org.apache.pekko.actor.typed.scaladsl.Behaviors
import org.apache.pekko.actor.typed.{ActorRef, ActorSystem, Behavior}
import org.apache.pekko.dispatch.ExecutionContexts
import org.apache.pekko.stream.scaladsl.{BroadcastHub, Keep, Source, SourceQueueWithComplete}
import org.apache.pekko.stream.typed.scaladsl.ActorSink
import org.apache.pekko.stream.{Materializer, OverflowStrategy}
import scala.concurrent.duration.DurationInt
import scala.util.{Failure, Success}
// Actor 1: creates a Source and materializes it to a BroadcastHub. Uses Materializer bound to the actor.
sealed trait WorkerMsg
case object CrashOrStop extends WorkerMsg
object Worker {
private def running(queue: SourceQueueWithComplete[Int]): Behavior[WorkerMsg] = Behaviors.receiveMessage {
case CrashOrStop =>
// queue.complete() // => expected Behavior -- Source is closed
// queue.fail(RuntimeException("Worker crashed")) // => expected Behavior -- same
// throw RuntimeException("Worker has crashed") // => unexpected Behavior -- Source not closed
Behaviors.stopped // => unexpected Behavior -- same
}
def apply(listenerRef: ActorRef[ListenerMsg]): Behavior[WorkerMsg] = Behaviors.setup { ctx =>
given Materializer = Materializer(ctx)
// Primary Flow materializing to the BroadcastHub
val ((queue, terminationFuture), source) = Source
.queue[Int](10, OverflowStrategy.backpressure)
.watchTermination() { (mat, future) => (mat, future) }
.toMat(BroadcastHub.sink(bufferSize = 4))(Keep.both)
.run()
// (no influence on the issue -- just for seeing that this Flow really completes)
terminationFuture.onComplete {
case Success(_) => println("Worker: BroadcastHub Flow completed successful")
case Failure(ex) => println(s"Worker: BroadcastHub Flow completed with error: $ex")
}(using ExecutionContexts.global())
listenerRef ! SourceOpened(source)
// Minified example -- no one writing to the queue. With a separate writing actor the issue still happens.
running(queue)
}
}
// Actor 2: Receives the source coming from Actor 1 (Worker)'s BroadcastHub and materializes it to itself using an ActorSink.
// Uses materializer bound to actor
sealed trait ListenerMsg
case class SourceOpened(source: Source[Int, NotUsed]) extends ListenerMsg
case object SourceCompleted extends ListenerMsg
case class SourceFailed(ex: Throwable) extends ListenerMsg
case class PrintMsg(s: String) extends ListenerMsg
object Listener {
def apply(): Behavior[ListenerMsg] = Behaviors.receive {
case (ctx, PrintMsg(s)) =>
Behaviors.same // pseudo-processing
case (ctx, SourceFailed(ex)) =>
println(s"Listener: Source Closed -- $ex") // either this or SourceCompleted expected
Behaviors.stopped
case (ctx, SourceCompleted) =>
println("Listener: Source Closed successfully")
Behaviors.stopped
case (ctx, SourceOpened(incomingSource)) =>
given Materializer = Materializer(ctx)
// Second actor materializing the BroadcastHub's output source
incomingSource
.map(i => PrintMsg(i.toString))
.to(ActorSink.actorRef(ctx.self, SourceCompleted, ex => SourceFailed(ex)))
.run()
Behaviors.same
}
}
// Actor 3: sends the Stop/Crash message to the Worker after a while
sealed trait WorkerTerminatorMsg
case object TerminateWorker extends WorkerTerminatorMsg
object WorkerTerminator {
def apply(worker: ActorRef[WorkerMsg]): Behavior[WorkerTerminatorMsg] = Behaviors.withTimers[WorkerTerminatorMsg] { timers =>
timers.startSingleTimer(TerminateWorker, 3.second)
Behaviors.receiveMessage {
case TerminateWorker =>
println("Time to terminate the worker")
worker ! CrashOrStop
Behaviors.stopped
}
}
}
object Guardian {
def apply(): Behavior[NotUsed] = Behaviors.setup { context =>
val listener = context.spawn(Listener(), "Listener")
val workerRef = context.spawn(Worker(listener), "Worker")
context.spawn(WorkerTerminator(workerRef), "Worker-Terminator")
Behaviors.empty
}
}
@main
def main(): Unit = {
ActorSystem(Guardian(), "MySystem")
}
Summary
When materializing a BroadcastHub on an actor's materializer, stopping/crashing the actor does not complete the BroadcastHub Source.
Analysis
Two independent actors are running in this situation:
The first ("Worker") is materializing a Source via a
BroadcastHub.sink, running with a Materializer bound to the actor. The source created by the BroadcastHub is then passed to a different actor ("Listener") that runs it to itself using an ActorSink.When the Worker stops (or is crashed), the source created by the BroadcastHub is not terminated (i.e. the Listener does not receive a Completed/Error message).
The source that runs into the BroadcastHub is cancelled, as expected.
If the source that runs the BroadcastHub is completed/failed manually, the BroadcastHub source completes as expected.
The reproducer has a third actor that sends a Crash/Stop message to the Worker after a couple of seconds.
Reproducer