java - 当卡夫卡宕机时，卡夫卡消费者挂起投票

Question

我一直在研究 Zookeeper 和 Kafka 的基本设置以了解如何使用它，但我在与消费者打交道时遇到了麻烦。当 Kafka 不可用时，对 poll() 方法的调用将挂起，直到它重新联机。

卡夫卡版本:0.10.1.0

我的代码是这样的:

KafkaConsumer<String, byte[]> consumer = new KafkaConsumer<>(props);
consumer.subscribe(topics);

while (!stopped) {
    // If by any reason Kafka is not available this call will hang
    // until Kafka is back online.
    records = consumer.poll(timeout);

    for (ConsumerRecord<String, byte[]> record : records) {
        process(record);
    }

    Thread.sleep(sleepTime);
}

我读到，当我调用 poll() 时，消费者将无限期地尝试连接到 Kafka，直到它恢复在线或直到 consumer.wakeup()被称为。

我希望代码在 Kafka 不在线时表现不同。 在从不存在的 kafka 进行轮询时，是否有任何方法可以限制消费者重试或使其失败？

Answer 1

不幸的是，这仍然是一个问题。许多消费者方法可能会因各种情况而挂起。

有一个 Kafka 改进提案正在进行中，KIP-266 , 为 Consumer 方法添加超时以避免挂起。

据我所知，从另一个线程调用 wakeup() 是最好的解决方法

---

编辑:从 Kafka 2.0.0 开始，所有消费者调用都可以接受超时。这允许在经纪人倒闭的情况下恢复控制。