如何配置Kafka Java Consumer在Broker不可用时停止重试并崩溃

当Broker不可用时，Kafka Java Consumer会无限重试连接（至少我未等到它停止）。我希望在Broker不可用时，poll()方法抛出异常，或通过其他方式让应用直接崩溃，这看似易配置但我未找到相关信息。以下是示例代码，该代码会持续返回0条记录并无限重试连接：

package org.example;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.List;
import java.util.Properties;

public class Consumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9094");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");
        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(List.of("game.journal"));
            while (true) {
                ConsumerRecords<String, String> recs = consumer.poll(Duration.ofMillis(100));
                for (ConsumerRecord<String, String> rec : recs) {
                    System.out.printf("Recieved %s: %s", rec.key(), rec.value());
                }
            }
        }
    }
}

实现方案 #

1. 调整Consumer核心配置，触发连接超时异常 #

通过配置限制元数据获取、重连的时间和次数，让Consumer在无法连接Broker时主动抛出异常：

• metadata.fetch.timeout.ms：设置元数据获取超时时间（比如5000，即5秒），超时后无法获取Broker元数据会抛出TimeoutException
• reconnect.backoff.ms：设置初始重连间隔（比如1000，即1秒）
• reconnect.backoff.max.ms：设置最大重连间隔（比如3000，即3秒），避免无限拉长重试周期

将这些配置添加到Properties中：

props.put(ConsumerConfig.METADATA_FETCH_TIMEOUT_MS_CONFIG, 5000);
props.put(ConsumerConfig.RECONNECT_BACKOFF_MS_CONFIG, 1000);
props.put(ConsumerConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG, 3000);

2. 代码中添加空轮询计数逻辑，主动触发崩溃 #

即使配置了超时参数，poll()可能仍会返回空记录而不抛出异常。可以统计连续空轮询的次数，达到阈值时主动抛出异常终止应用：

修改后的完整代码：

package org.example;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.time.Duration;
import java.util.List;
import java.util.Properties;

public class Consumer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9094");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-group");
        // 添加超时与重连配置
        props.put(ConsumerConfig.METADATA_FETCH_TIMEOUT_MS_CONFIG, 5000);
        props.put(ConsumerConfig.RECONNECT_BACKOFF_MS_CONFIG, 1000);
        props.put(ConsumerConfig.RECONNECT_BACKOFF_MAX_MS_CONFIG, 3000);

        int emptyPollCount = 0;
        final int MAX_EMPTY_POLLS = 10; // 连续10次空轮询触发崩溃

        try (KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props)) {
            consumer.subscribe(List.of("game.journal"));
            while (true) {
                ConsumerRecords<String, String> recs = consumer.poll(Duration.ofMillis(100));
                if (recs.isEmpty()) {
                    emptyPollCount++;
                    if (emptyPollCount >= MAX_EMPTY_POLLS) {
                        throw new RuntimeException("Broker不可用：连续" + MAX_EMPTY_POLLS + "次轮询无记录，触发应用崩溃");
                    }
                } else {
                    emptyPollCount = 0; // 重置计数
                    for (ConsumerRecord<String, String> rec : recs) {
                        System.out.printf("Received %s: %s%n", rec.key(), rec.value());
                    }
                }
            }
        } catch (Exception e) {
            System.err.println("Consumer异常终止：" + e.getMessage());
            System.exit(1); // 主动退出应用进程
        }
    }
}

3. 启动阶段主动检测Broker连接 #

在Consumer订阅主题后，主动尝试获取Broker元数据，失败则直接抛出异常终止应用：

// 订阅主题后添加检测逻辑
consumer.subscribe(List.of("game.journal"));
try {
    consumer.listTopics(Duration.ofMillis(5000)); // 5秒内获取主题列表，失败则抛出异常
} catch (Exception e) {
    throw new RuntimeException("启动失败：无法连接到Broker", e);
}

内容的提问来源于stack exchange，提问作者artem malinko