fix: Cancel any replays when the filter stage stops - #1198
Conversation
| }) | ||
|
|
||
| override def postStop(): Unit = { | ||
| replayInProgress.values.foreach(_.queue.cancel()) |
There was a problem hiding this comment.
That is via an async callback. Will that callback complete before the termination, or is that not what we are trying to do?
There was a problem hiding this comment.
It's not waiting on that cancel in any way, so probably still a race, but at least somewhat less risk that any of those queue streams are still running when/if this stage stops vs not doing anything.
There was a problem hiding this comment.
Ah, crap you are probably on to something, the queue:s are always submaterialized, so no chance that they will stop unless we add some logic that defers this stage from stopping.
There was a problem hiding this comment.
So to get rid of that abrupt termination error we'd need to keep the stage running here, watchTermination on any queue, so we can complete only once any replays complete :/
There was a problem hiding this comment.
hm, and there is no way to silence the abrupt termination logging for these specific stages?
we would anyway want to stop the replay as quickly as possible?
There was a problem hiding this comment.
I think it is the QueueSink logic doing it when there are queued elements here, I guess we could introduce some option to silence that: https://github.com/akka/akka/blob/5e9e11b5f6df03ed89c5ca37bdbb295ff5c19ec4/akka-stream/src/main/scala/akka/stream/impl/Sinks.scala#L365
There was a problem hiding this comment.
Pushed a quick stab at delaying cancellation but not sure it is worth the risk to break something.
2 participants
No description provided.