apache kafka - How to make restart-able producer?

Question

Latest version of kafka support exactly-once-semantics (EoS). To support this notion, extra details are added to each message. This means that at your consumer; if you print offsets of messages they won't be necessarily sequential. This makes harder to poll a topic to read the last committed message.

In my case, consumer printed something like this

Offset-0 0
Offset-2 1
Offset-4 2

Problem: In order to write restart-able proudcer; I poll the topic and read the content of last message. In this case; last message would be offset#5 which is not a valid consumer record. Hence, I see errors in my code.

I can use the solution provided at : Getting the last message sent to a kafka topic. The only problem is that instead of using consumer.seek(partition, last_offset=1); I would use consumer.seek(partition, last_offset-2). This can immediately resolve my issue, but it's not an ideal solution.

What would be the most reliable and best solution to get last committed message for a consumer written in Java? OR

Is it possible to use local state-store for a partition? OR

What is the most recommended way to store last message to withstand producer-failure? OR

Are kafka connectors restartable? Is there any specific API that I can use to make producers restartable?

FYI- I am not looking for quick fix

Answer 1

In my case, multiple producers push data to one big topic. Therefore, reading entire topic would be nightmare.

The solution that I found is to maintain another topic i.e. "P1_Track" where producer can store metadata. Within a transaction a producer will send data to one big topic and P1_Track.

When I restart a producer, it will read P1_Track and figure out where to start from.

Thinking about storing last committed message in a database and using it when producer process restarts.