ifta-zoy opened a new issue, #3345:
URL: https://github.com/apache/pekko/issues/3345
# 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
```Scala
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")
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
