Kafkajs: Proper way of reading messages in a topic and then stopping/disconnecting consumer

[pekkalaaksonen]

What is the correct way to read all messages from (that have not been consumed yet) and then stop the consumer from reading new messages? Do I need to check after each message if the last offset of last partition has been reached and if yes, then call consumer.disconnect() or is there a simpler/better way of doing this?

[Nevon]

That's a little tricky actually. Within eachBatch you get the highwaterMark for each partition. So you could see if the last message in the batch has an offset equal to the highwater mark, and once you're at the highwater mark in every partition, that would mean that you've consumed everything in the topic.

It's not a super common use-case, as most of the time you'd want to continue consuming continuously, but for one-time operations it would make sense to have an easy way to "drain" a topic. Unfortunately I can't think of an easier way off the top of my head. Whenever I've needed to do this, it's been in an interactive context, so I've just written a script that outputs something as it's chugging along, and when it stops outputting stuff it means I've reached the end of the topic and can kill the script.

[pekkalaaksonen]

Ok, thanks. I'll look into what you suggested.

My use case here is that as a learning exercise I'm building a simple API which returns new messages from kafka topic when the API is called. This requires that once an API call is made, messages that have not been consumed are consumed and returned to requester. Therefore the consumer cannot remain consuming new messages as they are being sent to the topic as would be the usual way how consumers work. In real world it might be more sensible to store messages to database and API would return data queried from database instead of using kafka topic directly.

[Nevon]

A simpler solution might be to constantly consume messages and push them into an array, and then whenever your endpoint is called, you just return whatever is in the array and then empty it.

[pekkalaaksonen]

Very true.. why didn't I think of that...

[Nevon]

I'm gonna close this issue. Feel free to open another if you run into issues, or join us on our Slack.