Consuming messages

Configuring consumer properties

Many configuration settings exist for the consumer that control aspects of its behavior. The following settings are some of the most important ones.

Many more configuration settings are available, but read the Apache Kafka documentation before you start to experiment with them.

Consumer groups

A consumer group is a group of consumers that cooperates to consume messages from one or more topics. The consumers in a group all use the same value for the group.id configuration. If you need more than one consumer to handle your workload, you can run multiple consumers in the same consumer group. Even if you need one consumer only, it's usual to also specify a value for group.id.

Each consumer group has a server in the cluster that is called the coordinator responsible for assigning partitions to the consumers in the group. This responsibility is spread across the servers in the cluster to even the load. The assignment of partitions to consumers can change at every group rebalance.

When a consumer joins a consumer group, it discovers the coordinator for the group. The consumer then tells the coordinator that it wants to join the group and the coordinator starts a rebalance of the partitions across the group to include the new member.

The messages from a single partition are processed by only one consumer in each group. This mechanism ensures that the messages on each partition are processed in order. See the following diagram for an example where a topic contains three partitions and a consumer group, which is consuming that topic, contains two consumers. One consumer in the group is assigned two partitions, and the other consumer is assigned one partition.

When one of the following changes takes place in a consumer group, the group rebalances by shifting the assignment of partitions to the group members to accommodate the change:

• A consumer joins the group.
• A consumer leaves the group.
• A consumer is considered as no longer live by the coordinator.
• New partitions are added to an existing topic.

If you have a consumer group that is rebalanced, be aware that any consumer that has left the group has its commits that are rejected until it rejoins the group. In this case, the consumer needs to rejoin the group, where it might be assigned a different partition to the one it was previously consuming from.

Consumer liveness

Kafka automatically detects failed consumers so that it can reassign partitions to working consumers. It uses two mechanisms: polling and heartbeating.

If the batch of messages that are returned from Consumer.poll(...) is large or the processing is time-consuming, the delay before poll() is called again can be significant or unpredictable. In some cases, it's necessary to configure a long maximum polling interval so that consumers do not get removed from their groups just because message processing is taking a while. If this mechanism is the only one available, the time it takes to detect a failed consumer is also long.

To make consumer liveness easier to handle, background heartbeating was added in Kafka 0.10.1. The group coordinator expects group members to send it regular heartbeats to indicate that they remain active. A background heartbeat thread runs in the consumer and sends regular heartbeats to the coordinator. If the coordinator does not receive a heartbeat from a group member within the session timeout, the coordinator removes the member from the group and starts a rebalance of the group. The session timeout can be much shorter than the maximum polling interval, so that the time it takes to detect a failed consumer can be short, even if message processing takes a long time.

You can configure the maximum polling interval by using the max.poll.interval.ms property and the session timeout by using the session.timeout.ms property. You don't need to use these settings unless it takes more than 5 minutes to process a batch of messages.

Managing offsets

For each consumer group, Kafka maintains the committed offset for each partition that is consumed. When a consumer processes a message, it doesn't remove it from the partition. Instead, it just updates its current offset by using a process that is called committing the offset.

Event Streams retains committed offset information for 7 days.

Controlling the speed of message consumption

If you have problems with message handling that is caused by message flooding, you can set a consumer option to control the speed of message consumption. Use fetch.max.bytes and max.poll.records to control how much data a call to poll() can return.

Handling consumer rebalancing

When consumers are added to or removed from a group, a group rebalance takes place, and consumers are not able to consume messages. This results in all the consumers in a consumer group to be unavailable for a short period.

If you are notified with the "on partitions revoked" callback, use a ConsumerRebalanceListener to manually commit offsets (if you are not using auto-commit) and to pause further processing until notified of the successful rebalance by using the "on partition assigned" callback.

Code snippets

These code snippets are at a high level to illustrate the concepts involved. For complete examples, see the Event Streams samples in GitHub.

To connect a consumer to Event Streams, you need to create service credentials. For information about how to get these credentials, see Connecting to Event Streams.

In the consumer code, you first need to build the set of configuration properties. All connections to Event Streams are secured by using TLS and user-password authentication, so you need at least these properties. Replace BOOTSTRAP_ENDPOINTS, USER, and PASSWORD with those from your own service credentials:

Properties props = new Properties();
 props.put("bootstrap.servers", BOOTSTRAP_ENDPOINTS);
 props.put("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"USER\" password=\"PASSWORD\";");
 props.put("security.protocol", "SASL_SSL");
 props.put("sasl.mechanism", "PLAIN");
 props.put("ssl.protocol", "TLSv1.2");
 props.put("ssl.enabled.protocols", "TLSv1.2");
 props.put("ssl.endpoint.identification.algorithm", "HTTPS");

To consume messages, you also need to specify deserializers for the keys and values, as in the following example.

 props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
 props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");

These deserializers must match the serializers used by the producers.

Then, use a KafkaConsumer to consume messages, where each message is represented by a ConsumerRecord. The most common way to consume messages is to put the consumer in a consumer group by setting the group ID, and then call subscribe() for a list of topics. The consumer is assigned some partitions to consume, although if more consumers exist in the group than partitions in the topic, the consumer might not be assigned any partitions. Next, call poll() in a loop, receiving a batch of messages to process, where each message is represented by a ConsumerRecord.

props.put("group.id", "G1");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));  // T1 is the topic name
while (true) {
   ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
   for (ConsumerRecord<String, String> record : records)
     System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
}

This consumer loop runs forever but it can be interrupted from another thread by calling Consumer.wakeup() to achieve a tidy shutdown.

To commit offsets manually, it's first necessary to set the enable.auto.commit configuration to false. Then, use either Consumer.commmitSync() or Consumer.commitAsync() to update the consumer's committed offset periodically. For simplicity, this example processes the records for each partition and commits the last offset separately.

props.put("group.id", "G1");
props.put("enable.auto.commit", "false");
Consumer<String, String> consumer = new KafkaConsumer<>(props);
consumer.subscribe(Arrays.asList("T1"));
try {
  while (true) {
    ConsumerRecords<String, String> records = consumer.poll(Duration.ofMillis(100));
    for (TopicPartition tp : records.partitions()) {
      List<ConsumerRecord<String, String>> partRecords = records.records(tp);
      long lastOffset = 0;
      for (ConsumerRecord<String, String> record : partRecords) {
        System.out.printf("offset = %d, key = %s, value = %s%n", record.offset(), record.key(), record.value());
        lastOffset = record.offset();
      }
      // having processed all the records in the above loop, we commit the partition's offset to 1 more than the last offset 
      consumer.commitSync(Collections.singletonMap(tp, new OffsetAndMetadata(lastOffset + 1)));
    }
  }
}
finally {
  consumer.close();
}

Exception handling

Any robust application that uses the Kafka client needs to handle exceptions for certain expected situations. In some cases, the exceptions are not thrown directly because some methods are asynchronous and deliver their results by using a Future or a callback. Check out example code in GitHub that shows complete examples.

Handle the following list of exceptions in your code: