How to make kafka consumer to read from last consu

Question

I am new to kafka and trying to understand if there is a way to read messages from last consumed offset, but not from beginning.

I am writing an example case, so that my intention will not get deviate.

Eg:
1) I produced 5 messages at 7:00 PM and console consumer consumed those.
2) I stopped consumer at 7:10 PM
3) I produced 10 message at 7:20 PM. No consumer had read those messages.
4) Now, i have started console consumer at 7:30 PM, without from-beginning.
5) Now, it Will read the messages produced after it has started. Not the earlier ones, which were produced at 7.20 PM

Is there a way to get the messages produced from last consumed offset.?

Answer 1

I am new to kafka and trying to understand if there is a way to read messages from last consumed offset, but not from beginning.

Yes, it is possible to use console consumer to read from the last consumed offset. You have to add consumer.config flag while invoking kafka-console-consumer.

Example:-

[root@sandbox bin]# ./kafka-console-consumer.sh --topic test1 --zookeeper localhost:2181 --consumer.config /home/mrnakumar/consumer.properties

Here /home/mrnakumar/consumer.properties is a file containing group.id. Here is how the /home/mrnakumar/consumer.properties looks:-

group.id=consoleGroup

Withoug using consumer.config, it is possible to read either from beginning [ by using --from-beginning] or end of the Log only. End of the Log means all the messages published after consumer start.

Answer 2

You should set the auto.offset.reset parameter in your consumer config on largest, so it will read all messages after last committed offset.

Answer 3

Setting the auto.offset.reset=earliest, AND a fixed group.id=something in the consumer config will start the consumer at the last committed offset. In your case it should start consuming at the first message at 7:20. If you want it to start reading messages posted AFTER it starts, then the auto.offset.reset=latest will ignore the 10 messages sent at 7:20 and read any that come in after it starts.

If you want it to start at the beginning, you must either call seekToBeginning after the first consumer.poll(), or change the consumer group ID to something unique.