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

Я начинаю с последнего документа Kafka http://kafka.apache.org/documentation.html. Но я сталкиваюсь с некоторыми проблемами, когда я пытаюсь использовать новый API-интерфейс Consumer. Я выполнил следующие действия:

1. Добавить новую зависимость

<dependency>
    <groupId>org.apache.kafka</groupId>
    <artifactId>kafka-clients</artifactId>
    <version>0.8.2.1</version>
</dependency>

2. Добавить конфигурации

    Map<String, Object> config = new HashMap<String, Object>();
    config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
            "host:9092");
    config.put(ConsumerConfig.GROUP_ID_CONFIG, "test");
    config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer.class.getName());
    config.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
            StringDeserializer.class.getName());
    config.put(ConsumerConfig.PARTITION_ASSIGNMENT_STRATEGY, "range");

3. Использовать API KafkaConsumer

KafkaConsumer<String, String> consumer = new KafkaConsumer<String, String>(config);
consumer.subscribe("topic");

Однако, когда я пытаюсь опросить сообщение от брокера, я получил только нуль:

Map<String, ConsumerRecords<String, String>> records = consumer.poll(0);
if (records != null)
    process(records);
else
    System.err.println("null");

И затем я знаю, что неправильно с потребителем после того, как я проверил исходный код:

@Override
public Map<String, ConsumerRecords<K,V>> poll(long timeout) {
    // TODO Auto-generated method stub
    return null;
}

Что еще хуже, я не могу найти никакой другой полезной информации об API 0.8.2, так как все использование Kafka несовместимо с последней версией. Может ли кто-нибудь помочь мне? Большое спасибо.

Ответы

dependencies {
  compile group: 'org.apache.kafka', name: 'kafka-clients', version: '0.8.2.1'
  compile group: 'org.apache.kafka', name: 'kafka_2.11', version: '0.8.2.1'
}