Kafka 0.9.0 Changes For Developers

Kafka 0.9.0 brings with it a bevy of goodness. Many of those features are aimed at operations, like security. Some are developer features, like a new Consumer API. This post will focus on the new features of 0.9.0 for developers.

New Consumer API

The most notable change is a brand-new consumer API. Unfortunately, there isn’t a good way to distinguish between the 0.7.x/0.8.x consumer and the new 0.9.x consumer. We’ve started to call the 0.9.0 consumer the “new” consumer and the 0.7.x/0.8.x consumer the “old” consumer.

The new consumer is a great replacement for the SimpleConsumer and the old consumer. IMHO, it gives the best of the both worlds. It has the relative simplicity of the old consumer while adding the power user features of the SimpleConsumer.

HelloWorld Consumer

The JavaDocs for the new consumer have all of the sample code. You’ll want to look there for code samples.

For now, here’s what a very simple new consumer looks like:

KafkaConsumer<String, String> consumer;

public void createConsumer() {
  String topic = "hello_topic";

  Properties props = new Properties();
  // Configure initial location bootstrap servers
  props.put("bootstrap.servers", "broker1:9092");
  // Configure consumer group
  props.put("group.id", "group1");
  // Configure key and value deserializers
  props.put("key.deserializer", 
      "org.apache.kafka.common.serialization.StringDeserializer");
  props.put("value.deserializer", 
      "org.apache.kafka.common.serialization.StringDeserializer");

  // Create the consumer and subscribe to the topic
  consumer = new KafkaConsumer<String, String>(props);
  consumer.subscribe(Arrays.asList(topic));

  while (true) {
    // Poll for ConsumerRecords for a certain amount of time
    ConsumerRecords<String, String> records = consumer.poll(100);

    // Process the ConsumerRecords, if any, that came back
    for (ConsumerRecord<String, String> record : records) {
            String key = record.key();
            String value = record.value();

            // Do something with message
    }
  }
}

There are some notable changes here. One big one is the consumer.poll() instead of the old it.hasNext(). The poll will only block for a specified amount of time while the hasNext blocks until a message comes in.

Also missing is the cumbersome Map to subscribe to topics and the number of streams/threads to consume them with. There’s a much cleaner subscribe to a List of topics.

Old Versus New Consumer

As of right now, the new consumer doesn’t have as many production miles on it. The old consumer has vastly more usage. The old consumer isn’t available in the 0.9.0 JAR. However, old consumers can still access the 0.9.0 brokers.

The 0.9.0 consumer assumes all offsets are stored in Kafka. If you have an existing old consumer, you’ll need to move your offsets to Kafka from ZooKeeper by setting offsets.storage=kafka and dual.commit.enabled=true in your 0.8.2 consumer code. Once you’ve moved the offsets to Kafka, you can start using a new consumer at the same place. Another option is to save out the current old consumer’s offset location and start the new consumer at the same offset.

Advanced Consumers

The SimpleConsumer used to be the only way to do advanced consumption. The new consumer changes that. The new consumer can easily consume a single partition. It can manage and change its offset location too. This was a common ask in the community. Check out the JavaDocs to see these examples.

Kafka Connect

Another notable improvement for developers is not having to develop. With Kafka Connect, we get a framework for moving data in and out of Kafka. This way, we can configure data movement instead writing code.

The community has taken up this framework and are creating many new connectors. Confluent has a page keeping track of the ones that have been created so far.