Kafka生产者未提示所有Broker不可达的技术问询

问题场景 #

当Kafka集群所有Broker节点均不可达时，生产者send()回调仅返回通用超时错误：

Topic XXX not present in metadata after 60000 ms.

开启DEBUG日志后，能看到所有节点连接尝试失败的细节（如连接超时、DNS解析失败）：

DEBUG org.apache.kafka.clients.NetworkClient - Initialize connection to node node2.url:443 (id: 2 rack: null) for sending metadata request
DEBUG org.apache.kafka.clients.NetworkClient - Initiating connection to node node2.url:443 (id: 2 rack: null) using address node2.url:443/X.X.X.X:443
....
DEBUG org.apache.kafka.clients.NetworkClient - Disconnecting from node 2 due to socket connection setup timeout. The timeout value is 16024 ms.
DEBUG org.apache.kafka.clients.NetworkClient - Initialize connection to node node0.url:443 (id: 0 rack: null) for sending metadata request
DEBUG org.apache.kafka.clients.NetworkClient - Initiating connection to node node0.url:443 (id: 0 rack: null) using address node0.url:443/X.X.X.X:443
....
DEBUG org.apache.kafka.clients.NetworkClient - Disconnecting from node 0 due to socket connection setup timeout. The timeout value is 17408 ms.

但这些关键故障细节仅在日志中存在，无法通过回调直接传递给应用，导致故障排查效率低下。

解决方案 #

1. 自定义生产者拦截器捕获全节点不可达状态 #

实现ProducerInterceptor，在消息发送或回调阶段，通过Kafka客户端的连接状态判断所有Broker是否不可达，并抛出自定义异常或触发告警：

public class ConnectionMonitorInterceptor<K, V> implements ProducerInterceptor<K, V> {
    private KafkaProducer<K, V> producer;

    @Override
    public void configure(Map<String, ?> configs) {
        // 可通过反射或初始化时注入Producer实例
    }

    @Override
    public ProducerRecord<K, V> onSend(ProducerRecord<K, V> record) {
        checkAllBrokersReachability();
        return record;
    }

    @Override
    public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
        if (exception instanceof TimeoutException) {
            checkAllBrokersReachability();
        }
    }

    private void checkAllBrokersReachability() {
        boolean allUnreachable = producer.connectionStates().allNodes().stream()
                .allMatch(node -> !producer.connectionStates().isConnected(node.id()));
        if (allUnreachable) {
            throw new AllBrokersUnreachableException("所有Kafka Broker节点均不可达");
        }
    }

    @Override
    public void close() {}
}

在生产者配置中启用拦截器：

producer.interceptor.classes=com.your.package.ConnectionMonitorInterceptor

2. 监听NetworkClient连接状态变化 #

通过反射获取Kafka内部的NetworkClient实例，注册连接状态监听器，实时捕获节点断开事件并判断是否全节点不可达：

// 反射获取NetworkClient实例
Field networkClientField = KafkaProducer.class.getDeclaredField("networkClient");
networkClientField.setAccessible(true);
NetworkClient networkClient = (NetworkClient) networkClientField.get(producer);

// 注册连接状态监听器
networkClient.addListener(new NetworkClient.Listener() {
    @Override
    public void onConnectionStateChange(Node node, ConnectionState newState) {
        if (newState == ConnectionState.DISCONNECTED) {
            boolean allUnreachable = networkClient.cluster().nodes().stream()
                    .allMatch(n -> networkClient.connectionState(n) == ConnectionState.DISCONNECTED);
            if (allUnreachable) {
                // 触发应用告警或异常处理逻辑
                triggerBrokerDownAlert();
            }
        }
    }
});

注意：该方式依赖Kafka内部API，版本升级时需验证兼容性。

3. 主动预检查Broker可达性 #

在生产者初始化完成后，主动发起连接检查，提前发现全节点不可达问题：

// 获取集群节点信息
Cluster cluster = producer.partitionsFor("target-topic").get(0).cluster();
boolean allReachable = cluster.nodes().stream().allMatch(node -> {
    try (Socket socket = new Socket()) {
        socket.connect(new InetSocketAddress(node.host(), node.port()), 5000);
        return true;
    } catch (IOException e) {
        return false;
    }
});

if (!allReachable) {
    throw new AllBrokersUnreachableException("初始化时发现所有Kafka Broker节点均不可达");
}

该方式适合启动时的健康检查，但无法覆盖运行时的节点不可达场景。

4. 关联日志上下文补充异常细节 #

利用日志框架的MDC（映射诊断上下文），在发送请求时标记唯一请求ID，当收到通用超时异常时，查询对应ID的DEBUG日志，将连接失败细节补充到异常信息中：

String requestId = UUID.randomUUID().toString();
MDC.put("kafka-request-id", requestId);

producer.send(record, (metadata, exception) -> {
    if (exception != null && exception instanceof TimeoutException) {
        // 从日志采集系统中查询该requestId对应的连接失败日志
        String failureDetails = LogQueryUtil.getDebugLogsByRequestId(requestId);
        throw new WrappedKafkaException(
            exception.getMessage() + "，故障细节：" + failureDetails,
            exception
        );
    }
});

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