java - Kafka 消费者卡在 iterator.hasNext() 处(实际上是暂停和恢复)，即使主题中有大量消息可供使用

Question

我正在开发一个分布式解决方案，其中 2 个消费者在同一消费者组下的 2 个不同服务器上运行，并从具有 2 个分区和复制因子 3 的 3 台机器 Kafka 主题进行消费。在我的消费者类中(这是一个 Callable)，关键部分如下:

@Override
public Object call() throws Exception {
    ConsumerIterator<byte[], byte[]> it = stream.iterator();
    try {            
        while (it.hasNext()){
            byte[] message = it.next().message();
            // other code here
        }
    } catch (Throwable e) {
        e.printStackTrace();
    }
    log.error("Shutting down Thread: " + streamNumber + ", kafka consumer offline!!!");
}

我的消费者类还生成了 16 个其他线程来处理所消费的消息。当我在两台不同的服务器上启动两个消费者时，最初几分钟，它们似乎都在无缝地消费来自 Kafka 主题的消息(每个分区一个分区)。然而，在一段时间之后，每个消费者似乎都陷入了 while (it.hasNext()) 语句，即使每个分区中还有数千条消息需要消费。下面的屏幕截图显示了此时 Kafka 消费者偏移量的状态。

正如您所看到的，消费者远远落后于主题中可用消息的数量。从我的日志来看，当这个消耗线程暂停时，其他线程运行良好并完成其工作。从较长时间的运行来看，有趣的是，我还注意到消耗线程会暂停并在一段时间后恢复。然而，每次暂停，下一次消费的消息数量也会减少得离谱。例如，在我第一次启动两个消费者之后，每个消费者似乎都无缝地消费了大约 15,000 条消息，直到卡在流迭代器上，然后暂停了 20 - 25 分钟，消费了大约 5,000 条消息，然后再次暂停了大约 30 分钟，消费了还有 100 个，这样的情况还在继续。如果我停止消费者进程并重新启动，整个循环似乎会重复。

这些是我正在使用的消费者配置:

group.id=ct_job_backfill     
zookeeper.session.timeout.ms=1000
zookeeper.sync.time.ms=200
auto.commit.enable=true
auto.offset.reset=smallest
rebalance.max.retries=20
rebalance.backoff.ms=2000
topic.name=contentTaskProd

消费者服务器都是在 Linux 上运行的 32 线程 64 GB 计算机。

知道是什么原因造成的吗？提前致谢。如果您需要更多信息或有任何不清楚的地方，请告诉我。

UPDATE: I have tried increasing the number of partitions from 2 to 32, and inside each of my consumer server spawning 16 consumers threads each consuming from a partition. However, that doesn't seem to change the behaviour. I notice the same pause and resume cycle.

Answer 1

我也遇到过同样的问题。在浏览解决方案时，我遇到了 kafka 已报告的问题 https://issues.apache.org/jira/browse/KAFKA-2978 .

看起来他们已经在 0.9.0.1 版本中解决了。我将尝试用这个版本更新库。如果我能够用新的 jar 解决这个问题，我会更新。与此同时，你也可以尝试同样的方法。〜干杯