spring-integration - 为什么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在 ConsumerIterator 中正是这个意思来自 KafkaStream :

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

哪里channel只是BlockingQueue .我认为为什么我们在为那个 consumer-timeout 的消息检索期间被阻止并不奇怪.

请从其他找到max-messages <int-kafka:consumer-configuration> 上的选项.是1默认情况下。