Replies: 1 comment 3 replies
|
Your understanding of the offset side is mostly right, but I would separate two things: Kafka listener shutdown and the lifecycle of your downstream Mongo resources. For async The error you are seeing looks like Mongo is being closed while the listener coroutine is still doing work: That is a lifecycle/timing problem more than an offset problem. Things I would check/tune:
For the coroutine cancellation expectation: I would not rely on Spring Kafka forcibly cancelling the running suspend body as the primary shutdown mechanism. The more robust setup is to let the listener stop receiving new batches and give the current async result enough time to complete. If you want active cancellation, add your own application-level cancellation signal and make the Mongo work cooperative with that signal. So the first practical experiment I would run is: set a shutdown timeout of e.g. 90s, set the pod termination grace above that, and verify from logs that Kafka listener containers stop before the reactive Mongo client is closed. If the errors disappear, the root cause was lifecycle ordering/timeout, not offset loss. |
|
Thank you for the response, I agree that the ordering of the components during shutdown is important, and making sure that the timeouts cascade properly is crucial as well. I don't think that increasing the timeouts is enough, though because it does not remove the race condition, but only reduces the likelihood of it causing issues. From reading the code, my understanding is this:
So What we're currently testing is an app-level I'll report back how well that works for us. I do still think that there's room for improvement in spring-kafka here. I'm thinking of something like retaining the |
|
Yes, that reading is more precise than my first timeout suggestion. If the async/suspend work is not part of the container's stop wait, then a longer For Spring Kafka itself, an opt-in "await/cancel in-flight async listener completions on stop" knob would make sense to me. It would need to be conservative because cancelling a coroutine/ |
|
After thinking about it more, I decided to create an issue for this: #4519 because I think that this should be part of spring-kafka |
Uh oh!
There was an error while loading. Please reload this page.
Context
We are using asynchronous batch listeners to consume messages from Kafka in our application. In particular, we are using Kotlin Coroutines to implement that.
Our listener function looks like this:
We have around ten of these listeners consuming messages from different topics, all of them are conceptually identical: parse the message body, convert it to a reasonable data format and then persist the batch of updates to a mongodb collection.
We're on spring boot 4.1.0 with spring-boot-starter-kafka 4.1.0, our application runs in a Kubernetes cluster, and we're using horizontal pod autoscaling to make sure we're provisioning enough resources for different load scenarios.
Problem
We're seeing error logs in scenarios where there's load on the system, and pods are being shut down (this is nothing unusual, it could be a scale down, a deployment of a new version, or a rescheduling of the pod because two k8s nodes are being consolidated). A few of our batch listeners can take some time for each batch to process (20-40 seconds is typical then). If such a batch is being processed while the application is about to shut down, we see error logs like the following:
Stacktrace 1
Stacktrace 2
If these errors occur, there are often a couple of each of them.
There are several questions I am looking to clarify here:
suspend funthat's annotated with@KafkaListener) should be cancelled (maybe after waiting for a given time to allow it to finish successfully). That cancellation should propagate through to the underlying coroutines and allow for a smoother shutdown, in particular making sure that the error logs don't occur.I am happy to hear your thoughts on that! Right now I have the feeling that spring kafka is the right place to ask these questions, let me know if you disagree with that and where I should go with that issue.
All reactions