Kafka 0.10.1.0消费者poll方法在集群宕机时挂起，如何限制重试或触发失败？

针对你遇到的Kafka集群不可用时poll()无限挂起的问题，在0.10.1.0这个版本里确实存在这种默认行为——消费者会不断重试连接集群，直到成功或被主动唤醒。下面给你两种可行的解决方案：

方案1：通过消费者配置限制重试与超时 #

虽然这个版本没有直接的"最大重试次数"配置，但可以通过组合几个参数让消费者在连接失败后抛出异常，而不是无限等待：

• 调整request.timeout.ms：这个参数控制每个请求的超时时间，默认是30000ms（30秒）。如果Kafka集群无响应，超过这个时间后消费者会抛出TimeoutException。你可以根据需求缩短这个值，比如设置为10000ms：

  props.put("request.timeout.ms", "10000");
  

• 缩短metadata.max.age.ms：默认是5分钟，这个参数控制元数据的过期时间。缩短它能让消费者更快检测到集群状态变化，比如设置为30000ms（30秒）：

  props.put("metadata.max.age.ms", "30000");
  

• 限制重试间隔上限：reconnect.backoff.max.ms控制重试连接的最大间隔，默认是1000ms。设置一个合理值，避免无限制的长间隔重试：

  props.put("reconnect.backoff.max.ms", "5000");
  

组合这些配置后，当Kafka集群长时间不可用时，消费者会在几次重试后触发超时异常，你可以捕获这个异常并处理离线逻辑。

方案2：代码层面主动触发终止 #

如果配置调整无法满足需求，你可以在代码中手动控制poll()的执行时长，通过consumer.wakeup()主动终止挂起的poll()：

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

// 启动定时任务，在指定时间后唤醒消费者
ScheduledExecutorService executor = Executors.newSingleThreadScheduledExecutor();
executor.schedule(() -> {
    if (!stopped) {
        consumer.wakeup();
    }
}, timeout + 5000, TimeUnit.MILLISECONDS); // 比poll超时多5秒，覆盖重试时间

try {
    while (!stopped) {
        try {
            records = consumer.poll(timeout);
            // 集群正常时重置定时任务
            executor.shutdownNow();
            executor = Executors.newSingleThreadScheduledExecutor();
            executor.schedule(() -> {
                if (!stopped) {
                    consumer.wakeup();
                }
            }, timeout + 5000, TimeUnit.MILLISECONDS);
            
            for (ConsumerRecord<String, byte[]> record : records) {
                process(record);
            }
            Thread.sleep(sleepTime);
        } catch (WakeupException e) {
            if (!stopped) {
                // 触发主动唤醒，说明Kafka可能离线，处理离线逻辑
                System.err.println("Kafka集群无响应，触发离线处理");
                // 可加入重试次数统计，达到次数后退出或报错
                break;
            }
        }
    }
} finally {
    executor.shutdown();
    consumer.close();
}

这个思路是：每次调用poll()前启动定时任务，如果poll()在指定时间内未返回（说明集群不可用），就调用wakeup()让poll()抛出WakeupException，然后在异常分支中处理离线情况。你还可以在异常分支中加入重试次数计数，达到指定次数后直接退出程序或抛出自定义异常。

需要注意的是，wakeup()是线程安全的，专门用于中断挂起的poll()操作，不会影响其他正常的消费者操作。

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