Consumer offsets

The initial design used a queue data structure to store messages entirely in memory.
This setup is simple but unreliable—messages are lost if the broker crashes or restarts.

To improve reliability, messages are now stored on disk in files.
An in-memory index maps each message identifier to its position in the log file.
This allows the broker to quickly locate and retrieve messages from disk.

A more advanced solution uses an LSM tree structure to build an embedded database.
This includes MMTable—an in-memory buffer for incoming writes—and SSTable—a sorted file of key-value pairs.
SSTables enable efficient storage and retrieval of messages by key.

Multiple index types exist, such as hash tables and bit trees.
Kafka uses a hash table index.
ActiveMQ uses a bit tree index.

For clarity in explanation, the overall message storage system is referred to as a "queue."
Though the real system is more complex, this simplification helps focus on core design concepts.

There are two distinct message processing models.

First model: Messages are pulled from the head of the queue.
The consumer processes the message.
If successful, the message is deleted.
If not, it is reinserted at the head for retry.
This model eliminates messages once processed.

Second model: Messages are retrieved from any position in the queue.
The broker knows the index of the next message to retrieve.
It returns the message to the consumer and increments the index to point to the next message.
Messages remain in the queue; only the index advances.

The difference is significant.
In the first model, the broker manages message status for every consumer.
It must track message delivery and ensure no duplicates.
It waits for acknowledgments and then either deletes or requeues messages.

Systems like ActiveMQ and RabbitMQ use this approach.

In the second model, the broker has minimal responsibilities.
It retrieves a message at a specified index and returns it to the requesting consumer.
No message deletion occurs immediately.

Systems like Kafka and Kinesis use this model.

The first approach introduces broker complexity.
This complexity directly impacts performance—specifically, throughput.

Kafka achieves higher throughput than RabbitMQ due to this design choice.

This comparison does not claim Kafka is superior overall.
It only highlights a performance benefit of Kafka’s architecture.

Both models have trade-offs.

In the second model, messages are stored in a log file on disk.
This ensures durability and persistence.

If messages are not immediately deleted after processing, they accumulate in memory. This accumulation quickly fills available memory.

Disk offers more storage space and is significantly cheaper than memory.

Messages are not stored in a single log file. Instead, they are split into many small files called segments.

Each segment is deleted only after all messages within it have been fully processed.

Every message is assigned a monotonically increasing sequence number.

This number uniquely identifies each message within a segment.

This sequence number is known as an offset in Kafka.

Offsets begin at zero. For each new message, the previous offset is incremented and assigned to the message.

In Kinesis, this concept is called a sequence number.

Systems using this model are referred to as log-based messaging systems.

A mechanism is needed to track the offset for each consumer.

Two options exist:

• The broker tracks the offset for each consumer
• Consumers themselves track their own offsets

Typically, consumers are responsible for offset tracking.

This means the consumer decides when to increment the offset and how to store it.

The process of persisting the offset is called checkpointing.

Consumers must checkpoint periodically.

If a consumer fails and is replaced by a new instance, the new consumer retrieves the stored offset and resumes processing from where the previous instance left off.

Offsets are read infrequently.

Offsets are written frequently—after every individual message or after every few messages.

What storage is best for such frequent writes?

This question is deferred to future discussions on storage selection.

Currently, Kinesis allows DynamoDB as one option.

Kafka initially used ZooKeeper to store offsets.

ZooKeeper provides a key-value store.

However, ZooKeeper performs poorly under high write loads.

The current solution is to store offsets directly within the broker.

When a consumer retrieves a message, it has two checkpointing choices:

• Checkpoint before processing the message
• Checkpoint after processing the message

If checkpointing occurs before processing:

• A processing failure results in message loss
• This guarantees at most one delivery

If checkpointing occurs after processing:

• A failure during checkpointing means the offset is not updated
• The message is re-pulled from the broker
• This ensures at least one delivery

Because messages are not deleted, consumers can rewind to an earlier offset and re-process data.

For example, if a consumer code has a bug discovered after processing messages, the consumer can re-process old messages once the bug is fixed.

This capability is not available in systems where messages are deleted after consumption.

The offset mechanism enables reliable message replay.

This provides a higher level of reliability than simple acknowledgments and retries.