KafkaConsumer在KafkaServer上出错（版本0.9.0.1）

Question

我试图用Kafka客户端库（0.9.0.1）测试生产者和消费者。代理（0.9.0.1）正在服务器上运行，我已经测试了KafkaProducer，没有问题。但当我测试KafkanConsumer进行轮询时，代理会发出一条错误消息。

[2016-03-18 13:44:19,129]错误关闭/172.26.132.149的套接字，因为错误（kafka.network.处理器）kafka.common.KafkaException：错误的请求类型10在kafka.api.Request estKeys$. desializerForKey（Request estKeys.scala:57）在kafka.network.Request estChannel$Request.（Request estChannel.scala:53）在kafka.network.Processor.read（SocketServer.scala:353）在kafka.network.Processor.run（SocketServer.scala:245）

消费者测试代码如下。

class ConsumerRunner implements Runnable{
    private KafkaConsumer<String,String> consumer;
    private String topic;
    public ConsumerRunner(String topic,Properties props){
        consumer = new KafkaConsumer<String,String>(props);
        this.topic = topic;
        consumer.subscribe(Arrays.asList(this.topic));
    }
    public void run() {
        while(true){
            ConsumerRecords<String,String> records = consumer.poll(10000);
            for (ConsumerRecord<String, String> record : records)
                 System.out.printf("offset = %d, key = %s, value = %s", record.offset(), record.key(), record.value());
        }
    }

}

我猜轮询请求包含错误的请求类型键，但当我检查Kafka核心源时，我重新确认请求类型键“10”定义为“GroupCoordinatory”。我在Kafka身上发现了可疑代码。网络请求通道。scala'

   val requestObj =
      if ( RequestKeys.keyToNameAndDeserializerMap.contains(requestId))
        RequestKeys.deserializerForKey(requestId)(buffer)

      else
        null

测试使用者还显示错误消息

lient.clientEOFException： null atlient.java:320NetworkRorg.apache.kafka.clients.consumer.internals.FromReadableChannel（NetworkRlient.poll）atlient.java:213NetworkRorg.apache.kafka.clients.consumer.internals.from（NetworkRlient.poll）atlient.java:193KafkaCorg.apache.kafka.clients.consumer.internals.（KafkaClient.poll）atlient.java:163KafkaCorg.apache.kafka.clients.consumer.internals.（KafkaCoordinator.ensure）atorg.apache.kafka.common.network.Selector.poll（Selector.java:286）atorg.apache.kafka.clients.NetworkClient.poll（NetworkClient.java:256）atorg.apache.kafka.clients.consumer.internals.消费者网络Cjava.io.投票（消费者网络Corg.apache.kafka.common.network.）ateceive.read消费者网络Ceceive.java:83（消费者网络Corg.apache.kafka.common.network.）ateceive.read消费者网络Ceceive.java:71（消费者网络Corg.apache.kafka.common.network.）athannel.receive消费者网络hannel.java:153（消费者网络Corg.apache.kafka.common.network.）athannel.readAbstractChannel.java:134协调器已知（Abstract协调器. java： 180）at org. apache. kafka. clients

有人有主意吗？是我的问题吗？还是其他人？请帮帮我。谢谢你。

Answer 1

Minkoo公司

不确定您只是试图为0.9 Kafka代码创建一个消费者，还是您的kafka消息有导致此问题的特定内容，您可以分享更多详细信息吗？

但是，如果您只是想为0.9编写一个Kafka消费者，那么在Kafka0.9中就有了新的消费者API。如果您愿意使用新的消费者API，请查看此示例https://github.com/sdpatil/KafkaAPIClient/blob/master/src/main/java/com/spnotes/kafka/simple/Consumer.java用于样品。

苏尼尔