Confluent-kafka-go: docs: At least once processing guarantee

[wedneyyuri]

Description

Hi guys, thanks for the work and the great tools you have produced for the Kafka community.

Today I am trying to understand what I need to do to guarantee at least once processing using Kafka. As far as I know the autocommit functionality is enabled by default and so if there is a problem in the message processing (for example a latency at the other end) the offset still will be updated (respecting auto.commit.interval.ms) and the message will be lost forever in case of process failure (for example a panic).

Generally the solutions I see to work around this problem involve manually managing the position of the offset, however I am having trouble finding examples of how to solve this using this golang package.

Proposal

What I have in mind:

• Receive 10,000 messages
• Process the 10000 messages (if it is necessary to parallelize more I can use goroutines, channels, waitGroups, ...)
• Update the offsets
• Receive the next 10,000 messages and start over

Further details about this behavior that I'm trying to explain are available on the newrelic blog: https://blog.newrelic.com/engineering/kafka-consumer-config-auto-commit-data-loss/

[wedneyyuri]

I'm not sure, but looks like making examples for manual commits will avoid common mistakes: https://github.com/confluentinc/confluent-kafka-go/issues/350, https://github.com/confluentinc/confluent-kafka-go/issues/286 and https://github.com/confluentinc/confluent-kafka-go/issues/195

[mhowlett]

with librdkafka based clients, auto.offset.commit / enable.auto.offset.store are true / true by default and as you note, this does not give at least once semantics because the commit happens in a background thread independently of the application polling the consumer for new messages. By comparison, with the java client, auto.offset.commit does give at least once semantics, because the commit happens as a side effect of the next consumer.poll.

The best way to achieve at least once semantics is to disable enable.auto.offset.store, which marks a message as eligible for commit as soon as it's delivered to the application and use StoreOffsets to manually indicate this instead. leave auto.offset.commit as true.

https://github.com/confluentinc/confluent-kafka-go/blob/master/kafka/consumer.go#L206

[wedneyyuri]

Thanks for your guidance and patience @mhowlett, I didn't find a auto.offset.commit configuration here. It's enable.auto.commit or I'm looking at the wrong place?

[mhowlett]

yes, i mean enable.auto.commit.

[marizaj]

@mhowlett Thanks for the explanation on achieving at-least-once delivery. I have a question regarding the strategy you have provided. So let's say that an error happens and we do not mark the offset as eligible for committing, but the interval for auto-commit hasn't kicked in yet and we continue with the next message which is processed successfully and we store the offset of a later message. What happens to that message that we didn't commit? Should I use the synchronous Commit() method after an error instead of waiting for auto-commit?
Thanks!

[mhowlett]

Kafka doesn't maintain per-message commit information - only one offset per partition. You are not committing a message, just updating the offset position. If the messages were on the same partition, then the message in error is effectively "committed".

[marizaj]

So, if we put this in other words, when auto.commit is enabled it is almost impossible to reach at_least_once delivery (no message is ever discarded) ?

[mhowlett]

it was the choice of your application to commit the offset in the knowledge a previous message was in error - Kafka gives you at-least once delivery.

Kafka does not give you work queue semantics out of the box. You can build this on top though (e.g. confluent's parallel consumer for java).