Kafka High Level Consumer – Frequently Missing Details

Apache Kafka is distributed with two java consumers: The high level consumer is very convenient to work with, providing clean and simple abstractions. The simple consumer gives the developer much more control over partition assignment and access to the data, but requires significantly more effort to use correctly.

There is a very detailed example of how to use the high level consumer in the Kafka documentation. This example is a great starting point and you should read it before reading anything else in this blog post.

However, the example is missing few details regarding the high level consumer API – leading to developers to use the low level consumer, which results in more complex code and more bugs.

So lets look at a slightly different example:  I’m a big fan of micro-batches, a design pattern that allows adjusting the latency and throughput tradeoffs to match requirements. Suppose we want to do micro-batch processing on data from Kafka. This will require reading as many events from Kafka as possible in 200ms, and then send the events to processing. If the send was successful, we can trust our processing pipeline to reliably handle the data, but if we failed to send the data on, we’d like to keep reading the same events from Kafka rather than moving the offset forward.

In terms of the high level consumer, we want to turn off auto commit feature, and rather commit events when we are certain they were successfully sent to the next stage of processing. This also means that we need to limit how long we wait for the next event, or we’ll miss our micro-batch window.

Lets start with the basic high level consumer loop shown in Kafka wiki:

while (it.hasNext())
    System.out.println("Thread " + m_threadNumber + ": " + new String(it.next().message()));

 while ((System.currentTimeMillis() < batchEndTime) && hasNext()) { 
    events.add(it.next().message()) }

After we collected enough events, we can send the list  to processing. We want to only commit the Kafka offsets if we successfully sent the events, so the first thing we need is to set auto.commit.enable=false in the consumer configuration.

Then lets look at the commit code:

if (events.size() > 0) {
        getProcessor().processEventBatch(events);
        events.clear();
        consumer.commitOffsets();
}

The order in which we do things here is important. First, we send the events to the processor, then we clear the event list and last we commit the offset to Kafka.

If we fail to send events to processing, this will throw an exception. The event list will not get cleared, which means that we can retry sending the same events again and again. Its important to note that the high level consumer tracks the offset independently – this means that if we clear the channel, as long as we don’t close the consumer, we will not be able to read the same events again even if we did not commit the offsets. That is – committing the offset protects against consumer crashes, but not against losing the events while the consumer is still iterating.

We commit the offset to Kafka after clearing the events list, because if Kafka throws an exception while committing offsets – we don’t need to process the events again – they were already successfully processed. This order of actions protects us both against data loss and against some cases of data duplication.

Note that we do not use multi-threaded consumer. Rather we have a separate consumer for each processing thread. We do that because currently the high-level consumer commits offsets of all threads at once. Since each thread processes data independently, committing offsets for all threads can lead to data loss.

The last piece of the puzzle is making sure our loop doesn’t hang when there are no more events to consume. After all, we want to send events to processing every 250ms regardless of how many events we collected at that time. This requires first setting a rather low timeout on the consumer: consumer.timeout.ms = 10 and then converting the timeout exception to a boolean value that will allow us to use hasNext() in a loop:

boolean hasNext() {
    try {
      it.hasNext();
      return true;
    } catch (ConsumerTimeoutException e) {
      return false;
    }
  }

The timeout value can be tuned – lower values will use more CPU but will allow more accurate micro-batches.

To see a complete example of a high level consumer using this pattern, I’ll refer you to the Flume Kafka Source.

One of the best things about the high level consumer is how it takes care of partition rebalancing for you. If you have multiple consumers in the same consumer group reading from a topic, if a consumer crashes or is added, the partitions in the topic will automatically rebalance between the consumers in the group.

Lets take a look at how this works: Here’s an example (created using ConsumerOffsetChecker tool) of one topic (t1) and one consumer group (flume). Each of the three partitions of t1 is being read by a different machine running the flume consumer:

Group           Topic                          Pid Offset   logSize         Lag             Owner
flume           t1                             0   501720   68100210042     50037974        kafkacdh-1
flume           t1                             1   49914701 49914701        0               kafkacdh-2
flume           t1                             2   54218841 82733380        28514539        kafkacdh-3

If kafkacdh-1 crashed, another consumer will pick up the partition:

Group           Topic                          Pid Offset     logSize         Lag            Owner
flume           t1                             0   59669715   100210042       40540327       kafkacdh-2
flume           t1                             1   49914701   49914701        0              kafkacdh-2
flume           t1                             2   65796205   82733380        16937175       kafkacdh-3

Then I can start flume_kafkacdh-4 and see things rebalance again:

Group           Topic                          Pid Offset     logSize         Lag            Owner
flume           t1                             0   60669715   100210042       39540327       kafkacdh-2
flume           t1                             1   49914701   49914701        0              kafkacdh-3
flume           t1                             2   66829740   82733380        15903640       kafkacdh-4

Using the fully power of the Kafka high level consumer can save you a lot of work, low level headache, bugs, code complexity and data loss. I recommend learning what it can do, and using it, rather than the lower level consumer, if at all possible.
Of course, soon Kafka will release another type of consumer – and life will become even easier.