Kafka — What are the delivery guarantees supported by a Kafka Consumer?
In this second & concluding part of delivery guarantees, we’ll see the message delivery guarantees for consumers and transactions.
In the previous write-up, we covered delivery guarantees from the producer to the broker & 2 Generals’ problems (in case you haven't seen that here is the link). In this, we deliver semantics from the broker to the consumer.
The delivery guarantees of consumers depend on offset, let's understand this with the analogy of a bookmark.
Analogy
In the context of a physical book, a bookmark can be placed between the pages to indicate where you stopped reading. This way, you can easily pick up where you left off without having to flip through the entire book to find your place.
It could also happen that a book is read by a few members of the family and that each of them can place their own bookmark in the book.
In both cases, the state is saved within the book, in the form of a thin bookmark. The other alternative is that the person could remember where he/she left reading & in this case, the book is not storing the state, instead, the reader is keeping track of it (either in memory or as a note).
Kafka works on a similar principle & I’ll come back to this analogy a few times, but to start, think of the book to be a Kafka topic, the human reader to be a consumer application, multiple family members as different application groups & bookmark as an offset.
Now let’s develop an understanding of this offset in the context of Kafka.
What is the Significance of offset?
When a message from a producer arrives at the broker, it is given a special number called an offset, which is the logical position of that record in the topic.
To be precise, the broker’s storage is sharded into partitions and the offset is the logical position in the partition where the message is placed. As can be seen in the image above.
When a consumer issues a fetch request to read records, it uses the offset like a bookmark to read messages from a position where it left last time.
Where are the offsets stored?
The offset represents the progress of a consumer & it is stored in the consumer’s memory.
With default consumer settings, the consumer also stores the state in the Kafka broker by issuing OffsetCommitRequest. The broker then stores the offset for the consumer group in an internal Kafka topic called __consumer_offset. Going back to the analogy, this is like a bookmark and the book maintaining the state of the last read page.
On the other hand, the consumer can keep this in an external store, and this is analogous to the reader keeping track of it.
So just like a human reader, it's the consumer client application that is responsible for keeping track of the progress & storing the offset to the broker or an external persistent store.
How does committing offset form the basis for delivery guarantees?
Kafka uses pull-based consumers. With every fetch request, the consumers ask the broker to return messages from an offset. And controls when to save the last processed offset.
When a consumer starts up, it usually has no state about the last consumed offset. In this case, it asks the broker to either return the messages from where it left last known as the Earliest Offset, or the newest messages from this point onwards known as the latest offset.
Let's see how this choice affects the delivery guarantees —
Delivery guarantees
1. At most once — Messages may be lost but are never re-delivered.
If the consumer chooses the latest messages, it means it will only see the new messages arriving since it registered. The consumer will miss any messages produced while it is unavailable.
This could be desirable when the consumer is interested in the current state & losing old messages is acceptable.
This behaviour falls under at-most-once delivery. The same message is not read again even if the consumer crashes while processing.
2. At least once — Messages are never lost but may be redelivered.
If losing messages is not acceptable, it can choose to read the earliest messages and in this case, the broker returns the messages since its last committed offset.
So if the consumer crashes without commit offset, it’ll read the message again.
In this case, it's actually the processing of the message that usually matters not just delivery.
So the delivery guarantee for consumers should be looked at as delivery & processing guarantee, depending on —
- The start-up behaviour of a consumer.
2. How the consumer ensures that the message is being processed.
This brings us to the point of controlling when to commit offset to the broker.
The consumer application can choose to either automatically commit offset —in which a background thread periodically commits the offset to the broker. Or have a finer control, to commit the offset at a point that makes sense to the application rather than a timer.
While auto-commit is quite convenient, its not always desired. Let's see how.
Consider a case, where the Automatic commit is turned on. The sequence of events that happen in the application are the following
- Consumer application pulls message from broker
- Consumer application processes the message
In parallel to this, the background thread keeps committing the offset to the Kafka broker as per the set interval.
While the consumer is churning messages it's no problem. But what if, the consumer crashes? Now, it matters when the offset is committed.
Case 1 — Offset is committed before the messages are processed & the consumer application crashes before processing the directive.
- Consumer application pulls message from broker
- Backgroud thread commit offset.
- The Consumer application crashes before it can process the message.
In this case, the message is lost.
Case 2 — The message is processed but the application crashes before Offset is committed.
- Consumer application pulls message from broker
- Consumer application processes the message
- The application crashes before the background thread commits offset.
In this case, we end up with duplicate messages, as the consumer restarts from the last committed offset.
This uncertainty isn't always acceptable. So in order to be sure that the messages are processed at least once the commit must happen after the message has been processed.
This is achieved by stopping automatic commit & explicitly committing after the message (or better a batch of messages) is processed.
Note: Over-committing has performance implications because it's a call to the broker. Moreover, there are more records produced in __consumer_offset topic.
This way the consumer has achieved an At-least-Once guarantee. But then it may receive duplicate messages.
3. Exactly once — Also known as effectively once. Each message is delivered once and only once.
This is challenging because of the uncertainties of distributed computing as we saw in 2 Generals’ problem. Also, from the last writeup we saw that the producer was able to simulate exactly once using a combination of at least once and a deduplication.
Now let's see what possibilities are there for a consumer to provide this guarantee —
- The offsets are stored in an external data store rather than Kafka. In case of a crash, the consumer reads the last processed offset, and then reads from the Kafka using this saved offset (seek to the offset location).
- Implement an Idempotent logic in the application — this could be to store the offset of the message along with the data before processing it. So if the message is reread after a crash, it is ignored due to the offset being already available in the database.
Kafka Transactions
If we are reading from one Kafka topic, processing & writing to another Kafka topic then we can use Kafka transactions. At the moment, this is supported only when the data is being read and written to Kafka.
The transaction is initiated by the Producer client of the application using a transaction ID & timeout.
Committing offset to the consumer client & message to the producer is handled by transaction and is stored in the state uncommitted until the commit transaction is called.
This commits both, the message to the producer topic & the offset to the consumer topic.
If the operation fails, the transaction can be aborted. This will roll back pending offsets (at consumer) & messages (at producer).
Internally, the downstream consumers still see the message but they are in an uncommitted state. So it's important that the consumers are using the correct isolation level - readcommited.
Summary
End-to-end delivery is a 2-step process, producer -> Kafka guarantee & Kafka -> Consumer guarantee.
Kafka -> Consumer guarantee is based on offsets and there are 3 possible guarantees. At most once, At least once, and Exactly once. The application has to be implemented to support exactly once.
Kafka transaction is supported when an application has Kafka as source & destination. Because the transaction controls offsets to the consumer and message to the producer. In an uncommitted transaction, the messages can still be read in an uncommitted state.
I hope you liked my writeup & in case of any questions please write me a comment. Thanks
Reference & Further Study
My other Articles on Kafka
What is 2 Generals’ Problem? And can it affect my Kafka Producer? https://medium.com/codex/kafka-what-is-2-generals-problem-and-can-it-affect-my-kafka-producer-2a621d4e1fbd
Kafka — Why does Kafka use a pull-based message consumer? — https://medium.com/codex/asynchronous-communication-why-does-kafka-use-a-pull-based-message-consumer-442c19a70f58
