CODEWITHSUNDEEP
apache-kafka
Lucas Andrade
Lucas Andrade

Reputation: 195

Kafka consumer is processing all messages at startup

I am new to Kafka, and am developing a personal project with a few services and the communication between them is made through Kafka and I am using Confluent for housing Kafka remotely.

All works fine, but when I startup a server it will try to process all the old messages in the topics that were generated as I was testing the system.

I would like to avoid this because it is time consuming and those messages were already processed, when the server was up the last time. Is there any way to prevent this in the development environment?

Am I even using Kafka correctly? Are there good practises that I missed?

Upvotes: 0

Views: 1524

Answers (1)

OneCricketeer
OneCricketeer

Reputation: 192013

By "server", I assume you mean consumer. The broker server doesn't process data, only stores it.

If you have auto.offset.reset=earliest + enable.auto.commit=false + are not committing the records in your code (or are overall using a new group.id each time), this is the expected behavior since your group.id is not tracking already consumed data.

Since you're now in a situation where you have processed data, but no stored offsets, first set a static group id, then your options include

  • re-process all the data again, accepting the duplicates, perhaps adding some conditional filter in your consumer code to skip records
  • skip all processed and un-processed data and only start consuming brand-new records after the consumer starts, by either setting a new group.id + auto.offset.reset=latest, or use consumer.seekToEnd() / the kafka-consumer-groups CLI tool ; downside of setting auto.offset.reset=latest is that you might run into a situation where the consumer group has been idle too long, and the group expires, causing you to go back to the end of the topic, even though there may still be un-processed data
  • manually find the offsets for all the partitions for the last processed data and consumer.seek() to those offsets

Upvotes: 2

Related Questions