卡夫卡消费者挂在 Java 中的 .hasNext [英] Kafka Consumer hanging at .hasNext in java

问题描述

我在 Java 中有一个简单的 Kafka Consumer，代码如下

I have a simple Kafka Consumer in java with the following code

    public void run() {
        ConsumerIterator<byte[], byte[]> it = m_stream.iterator();
        while (it.hasNext()&& !done){
            try {
                System.out.println("Parsing data");
                byte[] data = it.next().message();
                System.out.println("Found data: "+data);
                values.add(data); // array list
            } catch (InvalidProtocolBufferException e) {
                e.printStackTrace();
            }
        }
        done = true;
    }

当一条消息被发布时，数据被成功读取，但是当它返回检查它时.hasNext()，它保持挂起并且永远不会回来.

When a message is posted, the data is read successfully, however when it goes back to checking it.hasNext(), it stays pending and never comes back.

是什么阻碍了这一切?

m_stream 是一个 KafkaStream 获取如下:

m_stream is a KafkaStream obtained as follows:

Map<String, Integer> topicCountMap = new HashMap<String, Integer>();
topicCountMap.put(topic, new Integer(a_numThreads));
Map<String, List<KafkaStream<byte[], byte[]>>> consumerMap = consumer.createMessageStreams(topicCountMap);
List<KafkaStream<byte[], byte[]>> streams = consumerMap.get(topic);
executor = Executors.newFixedThreadPool(a_numThreads);
for (final KafkaStream stream : streams) {
   // m_stream is one of these streams
}

推荐答案

解决方案是添加属性

consumer.timeout.ms"

"consumer.timeout.ms"

现在当达到超时时会抛出一个 ConsumerTimeoutException

Now when the timeout is reached a ConsumerTimeoutException is thrown

这篇关于卡夫卡消费者挂在 Java 中的 .hasNext的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！