Spring Boot应用中Kafka健康检查的实现与疑问

一、有没有必要给Kafka做健康检查？ #

必须有，理由很直接：

• 能提前揪出集群问题：比如Broker挂了、网络断了这些情况，总比业务消息发不出去/收不到才发现强
• 稳住业务链路：Kafka很多时候是业务的核心中间件，一旦出问题影响一大片，健康检查能快速定位故障
• 适配微服务运维：不管是K8s的存活/就绪探针，还是服务网格的流量调度，健康检查都是基础操作

二、用健康主题验证生产消费能力合理吗？ #

当然合理，而且刚好能补上KafkaAdmin.describeCluster()的短板：

• describeCluster()只能确认应用和Kafka集群通不通、Broker活着没，但没法验证实际发消息、收消息的链路是否真的正常——比如权限不够、主题配置错了、消费者组出问题这些情况，光查集群状态发现不了
• 用健康主题的方式更贴近真实业务场景：
  • 生产端：往专属健康主题发条测试消息，能验证生产者序列化、权限、Broker接收等环节是否正常
  • 消费端：从这个主题把测试消息收回来，能验证消费者反序列化、分组配置、分区分配等环节是否正常
• 几个要注意的点：
  • 健康主题单独建，搞成单分区、短保留时间（比如1小时），别占资源
  • 检查逻辑加超时控制，别让健康检查拖慢应用
  • 可以结合Spring Boot Actuator的自定义HealthIndicator，把结果放到/actuator/health端点里，方便监控

三、实际实现小建议 #

可以基于Spring Boot Actuator写个自定义健康检查类，示例代码如下：

@Component
public class KafkaProduceConsumeHealthIndicator implements HealthIndicator {

    private final KafkaTemplate<String, String> kafkaTemplate;
    private final Consumer<String, String> kafkaConsumer;
    private static final String HEALTH_TOPIC = "kafka-health-check";

    public KafkaProduceConsumeHealthIndicator(KafkaTemplate<String, String> kafkaTemplate,
                                              @Qualifier("healthCheckConsumer") Consumer<String, String> kafkaConsumer) {
        this.kafkaTemplate = kafkaTemplate;
        this.kafkaConsumer = kafkaConsumer;
    }

    @Override
    public Health health() {
        String testMsg = "health-check-" + System.currentTimeMillis();
        try {
            // 发送测试消息，超时5秒
            kafkaTemplate.send(HEALTH_TOPIC, testMsg).get(5, TimeUnit.SECONDS);
            
            // 订阅健康主题并拉取消息
            kafkaConsumer.subscribe(Collections.singleton(HEALTH_TOPIC));
            ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(5));
            
            // 检查是否收到自己发的测试消息
            boolean received = records.records(HEALTH_TOPIC).stream()
                    .anyMatch(record -> testMsg.equals(record.value()));
            
            if (received) {
                return Health.up().withDetail("detail", "Kafka生产消费链路正常").build();
            } else {
                return Health.down().withDetail("detail", "没收到健康检查测试消息").build();
            }
        } catch (Exception e) {
            return Health.down(e).withDetail("detail", "Kafka生产消费链路异常").build();
        } finally {
            kafkaConsumer.unsubscribe();
        }
    }
}

• 注意：单独给健康检查配个Consumer，别和业务消费者组混在一起，避免干扰业务
• 把这个检查和KafkaAdmin自带的健康检查（Spring Boot Actuator已经集成了）结合起来，先查连通性，再查生产消费，分层验证更靠谱

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