Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

fix: Cancel any replays when the filter stage stops #1198

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open

fix: Cancel any replays when the filter stage stops #1198

Changes from 1 commit
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
361e8f9
fix: Cancel any replays when the filter stage stops
johanandren Sep 17, 2024
723b416
defer cancellation until all replays have stopped
johanandren Sep 18, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions akka-projection-grpc/src/main/scala/akka/projection/grpc/internal/FilterStage.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -463,6 +463,11 @@ import org.slf4j.LoggerFactory
pullInEnvOrReplay()
}
})

override def postStop(): Unit = {
replayInProgress.values.foreach(_.queue.cancel())
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

That is via an async callback. Will that callback complete before the termination, or is that not what we are trying to do?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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 :/

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Pushed a quick stab at delaying cancellation but not sure it is worth the risk to break something.

replayInProgress = Map.empty
}
}

}
Loading