Kafka节点故障后出现LEADER_NOT_AVAILABLE，应用无法自动恢复连接

先梳理下你的场景：3个Kafka 1.0.0节点，MissionControlTopic是4分区3副本，节点0偶尔失联后，其余两个节点能正常接管分区，但多数生产者、消费者会持续报LEADER_NOT_AVAILABLE；就算节点0恢复、leader重新平衡后，应用还是报错，只有重启才能恢复——但本地复现却能自动恢复，而且你已经排查过advertised.*配置没明显问题。

结合你提供的生产者、消费者代码，我觉得核心问题出在客户端元数据刷新不及时+生产者重试策略过于保守，咱们一步步说：

问题根源分析 #

1. 生产者重试配置为0：你把RETRIES_CONFIG设成了0，意味着一旦发送请求遇到LEADER_NOT_AVAILABLE，生产者直接失败、不会重试。而Kafka客户端刷新元数据需要时间（默认5分钟刷新一次），如果故障期间元数据没更新，生产者就会一直用旧的leader信息，哪怕后续leader恢复了也不会主动重试。
2. 客户端元数据缓存过期时间太长：默认metadata.max.age.ms是300000ms（5分钟），在生产环境中，这个周期太长了——节点故障、leader切换后，客户端要等很久才会主动拉取最新的集群元数据，导致一直用旧的leader地址。
3. Kafka 1.0.0版本的局限性：这个版本比较老旧，存在一些元数据同步的已知bug，比如leader切换后客户端元数据更新不及时的问题，这也是本地复现正常但生产环境出问题的可能原因之一（生产环境网络延迟、集群压力更大，放大了这个bug）。

针对性解决方案 #

1. 调整生产者配置，让它能自动重试并及时刷新元数据 #

修改生产者的重试和元数据相关配置，确保遇到故障时不会直接放弃，而是等待元数据更新后重试：

Properties properties = new Properties();
properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaUrl);
properties.put(ProducerConfig.ACKS_CONFIG, "all");
properties.put(ProducerConfig.RETRIES_CONFIG, Integer.MAX_VALUE); // 改为无限重试（可根据业务调整）
properties.put(ProducerConfig.RETRY_BACKOFF_MS_CONFIG, 1000); // 每次重试间隔1秒，避免频繁请求
properties.put(ProducerConfig.LINGER_MS_CONFIG, 10);
properties.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, 10000);
properties.put(ProducerConfig.METADATA_MAX_AGE_MS_CONFIG, 30000); // 缩短元数据过期时间到30秒
properties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.getCanonicalName());
properties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, GenericEventSerializer.getCanonicalName());

2. 优化消费者配置，加快元数据刷新 #

消费者同样需要缩短元数据过期时间，同时适当延长请求超时，给集群足够的时间恢复leader：

Properties props = new Properties();
props.put("enable.auto.commit", false);
props.put("bootstrap.servers", kafkaHostString);
props.put("group.id", consumerGroupId);
props.put("request.timeout.ms", 30000); // 延长请求超时到30秒
props.put("session.timeout.ms", 10000);
props.put("max.poll.records", 10000);
props.put("batch.size", 6400000);
props.put("metadata.max.age.ms", 30000); // 同样缩短元数据过期时间

Consumer<String, GenericEvent> consumer = new KafkaConsumer<>(props, new StringDeserializer(), new GenericEventDeserializer());
consumer.subscribe(Collections.singleton(topic));

另外，在消费者的异常处理中，可以主动触发元数据刷新，遇到LEADER_NOT_AVAILABLE时不用直接崩溃：

try {
    records = consumer.poll(pollTimeout);
    consumer.commitSync();
} catch (KafkaException e) {
    if (e.getMessage() != null && e.getMessage().contains("LEADER_NOT_AVAILABLE")) {
        // 调用poll(0)触发元数据刷新
        consumer.poll(0);
        LOGGER.warn("Leader not available, refreshing metadata", e);
    } else {
        LOGGER.error("Kafka consumer error", e);
    }
}

3. 长远建议：升级Kafka版本 #

Kafka 1.0.0是2017年的版本，后续的2.0+版本修复了很多元数据同步和客户端可靠性的问题，如果业务允许，建议逐步升级到较新的稳定版本（比如2.8.x或3.x），从根源上避免这类旧版本的bug。

为什么本地复现能自动恢复？ #

本地环境通常集群压力小、网络延迟低，元数据刷新的延迟可以忽略不计，而且故障恢复速度快，客户端能很快获取到新的leader信息。但生产环境中，集群压力大、网络可能有波动，元数据刷新的延迟被放大，就会出现必须重启应用才能恢复的情况。

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