Kafka Consumer¶
An Apache Kafka® Consumer is a client application that subscribes to (reads and processes) events. This section provides an overview of the Kafka consumer and an introduction to the configuration settings for tuning. Confluent Platform includes the Java consumer that is shipped with Apache Kafka.
Concepts¶
The Kafka consumer works by issuing “fetch” requests to the brokers leading the partitions it wants to consume. The consumer offset is specified in the log with each request. The consumer receives back a chunk of log beginning from the offset position. The consumer has significant control over this position and can rewind it to re-consume data if desired.
Consumer groups¶
A consumer group is a set of consumers which cooperate to consume data from some topics. The partitions of all the topics are divided among the consumers in the group. As new group members arrive and old members leave, the partitions are re-assigned so that each member receives a proportional share of the partitions. This is known as rebalancing the group.
The main difference between the older “high-level” consumer and the new consumer is that the former depended on ZooKeeper for group management, while the latter uses a group protocol built into Kafka itself. In this protocol, one of the brokers is designated as the group’s coordinator and is responsible for managing the members of the group as well as their partition assignments.
The coordinator of each group is chosen from the leaders of the
internal offsets topic __consumer_offsets, which is used to store
committed offsets. Basically the group’s ID is hashed to one of the
partitions for this topic and the leader of that partition is selected
as the coordinator. In this way, management of consumer groups is
divided roughly equally across all the brokers in the cluster, which
allows the number of groups to scale by increasing the number of
brokers.
When the consumer starts up, it finds the coordinator for its group and sends a request to join the group. The coordinator then begins a group rebalance so that the new member is assigned its fair share of the group’s partitions. Every rebalance results in a new generation of the group.
Each member in the group must send heartbeats to the coordinator in order to remain a member of the group. If no heartbeat is received before expiration of the configured session timeout, then the coordinator will kick the member out of the group and reassign its partitions to another member.
Offset Management¶
After the consumer receives its assignment from
the coordinator, it must determine the initial position for each
assigned partition. When the group is first created, before any
messages have been consumed, the position is set according to a
configurable offset reset policy (auto.offset.reset). Typically,
consumption starts either at the earliest offset or the latest offset.
As a consumer in the group reads messages from the partitions assigned by the coordinator, it must commit the offsets corresponding to the messages it has read. If the consumer crashes or is shut down, its partitions will be re-assigned to another member, which will begin consumption from the last committed offset of each partition. If the consumer crashes before any offset has been committed, then the consumer which takes over its partitions will use the reset policy.
The offset commit policy is crucial to providing the message delivery guarantees needed by your application. By default, the consumer is configured to use an automatic commit policy, which triggers a commit on a periodic interval. The consumer also supports a commit API which can be used for manual offset management. Correct offset management is crucial because it affects delivery semantics.
By default, the consumer is configured
to auto-commit offsets. Using auto-commit gives you “at least once”
delivery: Kafka guarantees that no messages will be missed, but
duplicates are possible. Auto-commit basically
works as a cron with a period set through the
auto.commit.interval.ms configuration property. If the consumer
crashes, then after a restart or a rebalance, the position of all
partitions owned by the crashed consumer will be reset to the last
committed offset. When this happens, the last committed position may
be as old as the auto-commit interval itself. Any messages which have
arrived since the last commit will have to be read again.
Clearly if you want to reduce the window for duplicates, you can
reduce the auto-commit interval, but some users may want even finer
control over offsets. The consumer therefore supports a commit API
which gives you full control over offsets. Note that when you use the commit API directly, you should first
disable auto-commit in the configuration by setting the
enable.auto.commit property to false.
Each call to the commit API results in an offset commit request being sent to the broker. Using the synchronous API, the consumer is blocked until that request returns successfully. This may reduce overall throughput since the consumer might otherwise be able to process records while that commit is pending.
One way to deal with this is to
increase the amount of data that is returned when polling. The
consumer has a configuration setting fetch.min.bytes which
controls how much data is returned in each fetch. The broker will hold
on to the fetch until enough data is available (or
fetch.max.wait.ms expires). The tradeoff, however, is that this
also increases the amount of duplicates that have to be dealt with in
a worst-case failure.
A second option is to use asynchronous commits. Instead of waiting for the request to complete, the consumer can send the request and return immediately by using asynchronous commits.
So if it helps performance, why not always use async commits? The main reason is that the consumer does not retry the request if the commit fails. This is something that committing synchronously gives you for free; it will retry indefinitely until the commit succeeds or an unrecoverable error is encountered. The problem with asynchronous commits is dealing with commit ordering. By the time the consumer finds out that a commit has failed, you may already have processed the next batch of messages and even sent the next commit. In this case, a retry of the old commit could cause duplicate consumption.
Instead of complicating the consumer internals to try and handle this problem in a sane way, the API gives you a callback which is invoked when the commit either succeeds or fails. If you like, you can use this callback to retry the commit, but you will have to deal with the same reordering problem.
Offset commit failures are merely annoying if the following commits succeed since they won’t actually result in duplicate reads. However, if the last commit fails before a rebalance occurs or before the consumer is shut down, then offsets will be reset to the last commit and you will likely see duplicates. A common pattern is therefore to combine async commits in the poll loop with sync commits on rebalances or shut down. Committing on close is straightforward, but you need a way to hook into rebalances.
Each rebalance has two phases: partition revocation and partition assignment. The revocation method is always called before a rebalance and is the last chance to commit offsets before the partitions are re-asssigned. The assignment method is always called after the rebalance and can be used to set the initial position of the assigned partitions. In this case, the revocation hook is used to commit the current offsets synchronously.
In general, asynchronous commits should be considered less safe than synchronous commits. Consecutive commit failures before a crash will result in increased duplicate processing. You can mitigate this danger by adding logic to handle commit failures in the callback or by mixing occasional synchronous commits, but you shouldn’t add too much complexity unless testing shows it is necessary. If you need more reliability, synchronous commits are there for you, and you can still scale up by increasing the number of topic partitions and the number of consumers in the group. But if you just want to maximize throughput and you’re willing to accept some increase in the number of duplicates, then asynchronous commits may be a good option.
A somewhat obvious point, but one that’s worth making is that asynchronous commits only make sense for “at least once” message delivery. To get “at most once,” you need to know if the commit succeeded before consuming the message. This implies a synchronous commit unless you have the ability to “unread” a message after you find that the commit failed.
In the examples, we show several detailed examples of the commit API and discuss the tradeoffs in terms of performance and reliability.
When writing to an external system, the consumer’s position must be coordinated with what is stored as output. That is why the consumer stores its offset in the same place as its output. For example, a Kafka Connect connector populates data in HDFS along with the offsets of the data it reads so that it is guaranteed that either data and offsets are both updated, or neither is. A similar pattern is followed for many other data systems that require these stronger semantics, and for which the messages do not have a primary key to allow for deduplication.
This is how Kafka supports exactly-once processing in Kafka Streams, and the transactional producer or consumer can be used generally to provide exactly-once delivery when transferring and processing data between Kafka topics. Otherwise, Kafka guarantees at-least-once delivery by default, and you can implement at-most-once delivery by disabling retries on the producer and committing offsets in the consumer prior to processing a batch of messages.
Tip
Consumers can fetch/consume from out-of-sync follower replicas if using a fetch-from-follower configuration. See Multi-Region Clusters to learn more.
Kafka Consumer Configuration¶
The full list of configuration settings are available in Kafka Consumer Configurations for Confluent Platform. Several of the key configuration settings and how they affect the consumer’s behavior are highlighted below.
Core configuration¶
The only required setting is
bootstrap.servers, but you should set a client.id
since this allows you to easily correlate requests on the broker with
the client instance which made it. Typically, all consumers within the
same group will share the same client ID in order to enforce
client quotas.
Group Configuration¶
You should always configure group.id unless
you are using the simple assignment API and you don’t need to store
offsets in Kafka.
You can control the session timeout by overriding the
session.timeout.ms value. The default is 10 seconds in the C/C++ and Java
clients, but you can increase the time to avoid excessive rebalancing, for example
due to poor network connectivity or long GC pauses.
The main drawback to using a larger session timeout is that it will take longer for the coordinator to detect when a consumer instance has crashed, which means it will also take longer for another consumer in the group to take over its partitions. For normal shutdowns, however, the consumer sends an explicit request to the coordinator to leave the group which triggers an immediate rebalance.
The other setting which affects rebalance behavior is
heartbeat.interval.ms. This controls how often the consumer will
send heartbeats to the coordinator. It is also the way that the
consumer detects when a rebalance is needed, so a lower heartbeat
interval will generally mean faster rebalancing. The default setting is
three seconds. For larger groups, it may be wise to increase this
setting.
Another property that could affect excessive rebalancing is max.poll.interval.ms. This
property specifies the maximum time allowed time between calls to the consumers poll method
(Consume method in .NET) before the consumer process is assumed to have failed.
The default is 300 seconds and can be safely increased if your application
requires more time to process messages. If you are using the Java consumer, you can also
adjust max.poll.records to tune the number of records that are handled on every
loop iteration.
Offset Management¶
The two main settings affecting offset
management are whether auto-commit is enabled and the offset reset
policy. First, if you set enable.auto.commit (which is the
default), then the consumer will automatically commit offsets
periodically at the interval set by auto.commit.interval.ms. The
default is 5 seconds.
Second, use auto.offset.reset to define the behavior of the
consumer when there is no committed position (which would be the case
when the group is first initialized) or when an offset is out of
range. You can choose either to reset the position to the “earliest”
offset or the “latest” offset (the default). You can also select
“none” if you would rather set the initial offset yourself and you are
willing to handle out of range errors manually.
Message handling¶
While the Java consumer does all IO and processing in the foreground thread, librdkafka-based clients (C/C++, Python, Go and C#) use a background thread. The main consequence of this is that polling is totally safe when used from multiple threads. You can use this to parallelize message handling in multiple threads. From a high level, poll is taking messages off of a queue which is filled in the background.
Another consequence of using a background thread is that all heartbeats and rebalancing are executed in the background. The benefit of this is that you don’t need to worry about message handling causing the consumer to “miss” a rebalance. The drawback, however, is that the background thread will continue heartbeating even if your message processor dies. If this happens, then the consumer will continue to hold on to its partitions and the read lag will continue to build until the process is shut down.
Although the clients have taken different approaches internally, they are not as far apart as they seem. To provide the same abstraction in the Java client, you could place a queue in between the poll loop and the message processors. The poll loop would fill the queue and the processors would pull messages off of it.
Kafka Consumer Group Command Tool¶
Kafka includes an admin utility for viewing the status of consumer groups.
List Groups¶
To get a list of the active groups in the cluster, you can use the
kafka-consumer-groups utility included in the Kafka distribution. On
a large cluster, this may take a while since it collects
the list by inspecting each broker in the cluster.
bin/kafka-consumer-groups --bootstrap-server host:9092 --list
Describe Group¶
The utility kafka-consumer-groups can also be used to collect
information on a current group. For example, to see the current
assignments for the foo group, use the following command:
bin/kafka-consumer-groups --bootstrap-server host:9092 --describe --group foo
If you happen to invoke this while a rebalance is in progress, the command will report an error. Retry again and you should see the assignments for all the members in the current generation.