为什么即使有消息，Kafka Consumer 也会阻塞直到超时？

Question

我正在使用带有 kafka 的高级客户端（遗留代码）。

但是，我观察到即使主题上有很多可用的消息，消费者仍然会被阻塞，直到 zk-connection-timeout 时间过去。

为什么会这样？我该如何解决这个问题？我感兴趣 - 是否找到消息？走得更远。

<int-kafka:zookeeper-connect id="zookeeperConnect"
    zk-connect="#{kafkaConfig['zooKeeperUrl']}" zk-connection-timeout="5000"
    zk-session-timeout="5000" zk-sync-time="2000" />

<int-kafka:consumer-context id="consumerContext" consumer-timeout="5000" zookeeper-connect="zookeeperConnect">

Answer 1

好吧，当您使用多个主题、配置时，当每个 KafkaStream 被阻塞并参与相同的顺序迭代过程时，这是高级消费者实现的一个已知问题。

更多信息请参见Slow consumer throuput when using 2 consumer-configuration。

更新

consumer-timeout 在来自KafkaStream 的ConsumerIterator 中就是这个意思：

currentDataChunk = (FetchedDataChunk)this.channel().poll((long)this.consumerTimeoutMs, TimeUnit.MILLISECONDS);

channel 就是 BlockingQueue。我认为这并不奇怪为什么我们在consumer-timeout 的消息检索期间被阻止。

请在<int-kafka:consumer-configuration> 上找到max-messages 选项。默认为1。