Kafka Consumer Design

An Apache Kafka® consumer works by issuing “fetch” requests to the brokers leading the partitions that a client wants to consume from. A consumer specifies the log offset with each request and receives back a chunk of the log beginning from that offset position. As a result, the consumer has significant control over this position and can specify the offset to re-consume data if needed.

Push versus pull

Kafka follows a traditional design, shared by most messaging systems, in that data is pushed by the producer to the broker and pulled from the broker by the consumer. Other log-centric systems, such as Scribe and Apache Flume, follow a push-based model where data is pushed to the consumers.

The goal of any messaging system to fully utilize the consumer with the correct transfer rate. Getting this rate right is harder than it seems.

There are pros and cons to pushing data to the consumer or a consumer that pulls data:

  • A disadvantage of a push-based system is that is has difficulty dealing with diverse consumers because the broker would control the rate at which data is pushed.
  • Another disadvantage of a push system is that a consumer can get overwhelmed when its rate of consumption falls below the rate of production (similar to a denial of service attack).
  • An advantage of a pull-based system is that enables the consumer to catch up when it can, if it falls behind production. The consumer can indicate it is overwhelmed and catch up can be implemented with some kind of backoff protocol.
  • Another advantage of a pull-based system is that enables aggressive batching of data sent to the consumer. In contrast, a push-based system must either send a request immediately or accumulate more data and then send it later without knowledge of whether the downstream consumer will be able to process it immediately. If tuned for low latency, this will result in sending a single message at a time only for the transfer to end up being buffered anyway, which is wasteful. A pull-based design fixes this by having the consumer pull all available messages after their current position in the log (or up to some configurable maximum size). So one gets optimal batching without introducing unnecessary latency.
  • A disadvantage of a naive pull-based system is that if the broker has no data the consumer may end up polling in a tight loop, effectively busy-waiting for data to arrive. To avoid this Kafka provides parameters in the pull requests that enable the consumer request to block in a “long poll”, waiting until data arrives. The request can optionally wait until a given number of bytes is available to ensure large transfer sizes.

Another possible design is one that is pull, end-to-end. The producer writes to a local log, and brokers pull from the log, with consumers pulling from the brokers. This is an interesting architecture, but doesn’t scale when the system has thousands of producers.

Track consumer position

Tracking what has been consumed is another key performance point of a messaging system. Most messaging systems keep metadata about what messages have been consumed, on the server side. That is, as a message is handed to a consumer, the server either locally records that fact immediately or may wait for acknowledgement from the consumer. Since the data structures used for storage in many messaging systems scale poorly, this is a pragmatic choice–since the broker knows what is consumed it can immediately delete it, keeping its data size small.

However, getting the broker and consumer to agree on what has been consumed can be challenging. If the broker records a message as consumed immediately after it is handed out over the network and the consumer fails to process the message because it crashes or the request times out, that message will be lost. To solve this problem, many messaging systems add an acknowledgement feature which means that messages are not marked as consumed when sent, but only marked as sent. The broker waits for a specific acknowledgement from the consumer to record the message as consumed. This strategy fixes the problem of losing messages, but creates new problems. First, if the consumer processes the message but fails before it can send an acknowledgement then the message will be consumed twice. The second problem is performance related. The broker must keep multiple states about every message. The broker locks it so it is not given out a second time, and marks it as permanently consumed so that it can be removed, and problems such as what do with messages that are sent but never acknowledged must be dealt with.

Kafka handles this differently. An Kafka topic is divided into a set of ordered partitions, and consumer-design are grouped by a group ID. Each partition is consumed by exactly one consumer within each subscribing consumer group at any given time. This means that the position of a consumer in each partition is just a single integer, the offset of the next message to consume. This means consumer state is very small, and this state can be periodically checkpointed. This makes message acknowledgements very cheap.

A side benefit of this is that a consumer can deliberately rewind to an old offset and re-consume data. This violates the common contract of a queue but turns out to be an essential feature for many consumers. For example, if a bug is found in the consumer code after consuming some messages, the consumer can re-consume those messages the bug is fixed.

Consumer position illustrated

When a consumer group is first initialized, the consumers typically begin reading from either the earliest or latest offset in each partition. The messages in each partition log are then read sequentially. As the consumer makes progress, it commits the offsets of messages it has successfully processed. For example, in the image that follow, the consumer’s position is at offset 6 and its last committed offset is at offset 1.

../../_images/consumer_position_image.png

When a partition gets reassigned to another consumer in the group, the initial position is set to the last committed offset. If the consumer in the example above suddenly crashed, then the group member taking over the partition would begin consumption from offset 1. In that case, it would have to reprocess the messages up to the crashed consumer’s position of 6.

The diagram also shows two other significant positions in the log. The log end offset is the offset of the last message written to the log. The high watermark is the offset of the last message that was successfully copied to all of the log’s replicas. From the perspective of the consumer, the main thing to know is that you can only read up to the high watermark. This prevents the consumer from reading unreplicated data which could later be lost. [1]

[1]This section adapted from https://www.confluent.io/blog/tutorial-getting-started-with-the-new-apache-Kafka-0-9-consumer-client/ by Jason Gustafson

Learn more

Note

This website includes content developed at the Apache Software Foundation under the terms of the Apache License v2.