Как заставить потребителей работать в Kafka 0.8 API

Во-первых, я хочу упомянуть, что между Kafka 0.8.0, 0.8.1 и 0.8.2 есть несколько изменений API (главное переписывание и упрощение произошло для 0.9.0 и 0.10.0) - таким образом, ваш вопрос немного открыт, просто спрашивая 0.8.

Чтобы написать потребителя Java для 0.8.2.2, вам нужно включить зависимость:

Это для Scala 2.11 — доступны и другие версии Scala.

<dependency>
  <groupId>org.apache.kafka</groupId>
  <artifactId>kafka_2.11</artifactId>
  <version>0.8.2.2</version>
</dependency>

Не используйте kafka-clients в качестве идентификатора артефакта для версии 0.8.x.

Минимальный пример для потребителя, получающего <String,String> парных сообщений "ключ-значение" и выводящего их на stdout, выглядит следующим образом:

import java.util.HashMap;
import java.util.Map;
import java.util.Properties;

import kafka.consumer.Consumer;
import kafka.consumer.ConsumerConfig;
import kafka.consumer.ConsumerIterator;
import kafka.consumer.KafkaStream;
import kafka.javaapi.consumer.ConsumerConnector;

public class ConsumerExample {

    public static void main(String[] args) {
        Properties props = new Properties();
        props.put("zookeeper.connect", "localhost:2181");
        props.put("group.id", "myGroup");

        final String topic = "test";

        ConsumerConnector consumer = Consumer.createJavaConsumerConnector(new ConsumerConfig(props));

        Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
        topicCountMap.put(topic, new Integer(1)); // number of consumer threads

        KafkaStream<byte[], byte[]> stream = consumer.createMessageStreams(topicCountMap).get(topic).get(0);

        ConsumerIterator<byte[], byte[]> it = stream.iterator();

        // infinite loop
        while(it.hasNext()) {
            System.out.println(new String(it.next().message()));
        }

        // non-reachable code...
        consumer.shutdown();
    }
}

Полный пример — использование нескольких потребительских потоков, включая правильное завершение работы — можно найти здесь: https://cwiki.apache.org/confluence/display/KAFKA/Consumer+Group+Example

Чтобы проверить это, следуйте краткому руководству и отправляйте сообщения через консоль Kafka. -продюсер.