kafka 实战-记一次生产环境broker may not available

1 故障背景

10:09:34.948 ifpo.apache .kafka.clients .Networkclient 748[Producer clientId=producer-1] Connection to node 0(10.x.x.x:9093) could not be established. Broker may not be available.

故障细节

生产kafka 服务端集群全部宕机，但车端还在往我的数据网关服务上报信息，而此时生产 kafka 集群全部宕机，但是这种异常，网关服务不会走到send 语句的 catch 语句块，因为这只是warn 而非 error 日志。所以导致这些数据全部丢失，直至发现大屏数据缺失！垃圾百度云的机器宕机且无任何告警！

解决方案

• 百度云运维给出平台控制台的监控告警方案
• 对 kafka broker 端探活自动重启
• kakfa 生产者客户端，考虑 fail fast方案，由数据网关这个 Java 生产者客户端代码实现，当 kafka 服务端宕机时，使发送消息 API 报错并让 java 客户端服务宕机。

环境

kafka 2.5.1 版本-百度云BMR集群。

2 客户端捕获异常

若希望在Kafka服务端宕机时让Java客户端抛出异常并且让服务宕机，可设置一些Kafka生产者参数实现。

可设：

• retries
• delivery.timeout.ms

控制生产者在放弃之前可以重试发送消息的次数和总时间。

默认情况下，Kafka生产者可能会重试多次发送消息，如果Kafka服务端宕机，生产者将会尝试在delivery.timeout.ms设置的时间内重试，直到超时。如果在这段时间内服务没有恢复，生产者将抛出一个TimeoutException，这时候你的服务可以捕获这个异常并作出相应的处理，比如让服务宕机。

以下是一个设置重试和超时参数的示例：

Properties props = new Properties();
props.put("bootstrap.servers", "your_kafka_broker:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("acks", "all");
props.put("retries", 0);  // 设置重试次数为0，表示不重试
props.put("delivery.timeout.ms", 15000);  // 设置消息发送超时时间

KafkaProducer<String, String> producer = new KafkaProducer<>(props);

在这个例子中，如果Kafka服务端宕机，生产者在发送消息时会立即失败而不进行重试，并且会在15秒后超时。这样，send方法会抛出一个TimeoutException。

然后，在你的send方法中，你可以捕获这个异常，并采取措施比如停止服务：

public void send(String topic, String key, String msg){
    try {
        producer.send(new ProducerRecord<>(topic, key, msg)).get();
    } catch (InterruptedException | ExecutionException e) {
        log.error("kafka send error: {}", e.getMessage(), e);
        // 在这里处理异常，比如让服务宕机
        System.exit(1);  // 非零状态码表示异常退出
    }
}

生产环境直接System.exit(1)会突然停止服务，而非优雅关闭。

实际应用可能想要实现一些更优雅的关闭过程，如关闭资源、通知依赖服务、等待正在处理的任务完成等。此外，你应确保这样的异常处理逻辑符合你的服务的容错和高可用策略。