如何将Kafka Streams状态存储恢复状态纳入Spring Boot健康检查？

核心思路 #

默认的Spring Boot Actuator Kafka Streams健康检查仅校验Kafka Streams的全局状态，但当流线程处于STARTING或PARTITIONS_ASSIGNED状态（此时状态存储正在恢复）时，全局状态可能仍显示为RUNNING，导致健康检查误判为Up。我们可以通过自定义HealthIndicator，深入检查每个流线程的状态，确保只有所有线程都进入RUNNING状态时，健康检查才返回Up。

实现步骤 #

1. 创建自定义健康指示器 #

编写一个实现HealthIndicator接口的类，注入Kafka Streams实例（或Spring Cloud Stream的StreamsBuilderFactoryBean），遍历所有流线程的状态进行校验：

import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.processor.internals.StreamThread;
import org.springframework.boot.actuate.health.Health;
import org.springframework.boot.actuate.health.HealthIndicator;
import org.springframework.stereotype.Component;

@Component
public class CustomKafkaStreamsHealthIndicator implements HealthIndicator {

    private final KafkaStreams kafkaStreams;

    // 若使用Spring Cloud Stream，替换为注入StreamsBuilderFactoryBean获取实例
    // public CustomKafkaStreamsHealthIndicator(StreamsBuilderFactoryBean streamsFactoryBean) {
    //     this.kafkaStreams = streamsFactoryBean.getKafkaStreams();
    // }

    public CustomKafkaStreamsHealthIndicator(KafkaStreams kafkaStreams) {
        this.kafkaStreams = kafkaStreams;
    }

    @Override
    public Health health() {
        if (kafkaStreams == null || kafkaStreams.state() != KafkaStreams.State.RUNNING) {
            return Health.down()
                    .withDetail("streamsState", kafkaStreams != null ? kafkaStreams.state() : "UNINITIALIZED")
                    .build();
        }

        // 遍历检查所有流线程的状态
        for (StreamThread thread : kafkaStreams.localThreadsMetadata()) {
            StreamThread.State threadState = thread.state();
            if (threadState != StreamThread.State.RUNNING) {
                return Health.outOfService()
                        .withDetail("streamThreadId", thread.threadId())
                        .withDetail("threadState", threadState)
                        .withDetail("message", "State store is recovering, not ready for requests")
                        .build();
            }
        }

        return Health.up().withDetail("allStreamThreads", "RUNNING").build();
    }
}

2. 禁用默认健康检查（可选） #

如果希望完全替换默认的Kafka Streams健康检查，可在application.properties中禁用默认指示器：

management.health.kafka-streams.enabled=false

3. 验证效果 #

启动应用后，当流线程处于STARTING或PARTITIONS_ASSIGNED状态时，访问/actuator/health端点会返回OUT_OF_SERVICE状态；只有所有线程进入RUNNING状态后，才返回UP。负载均衡器可根据此状态避免将请求路由至未就绪的实例。

关键说明 #

• 线程状态STARTING表示流线程正在初始化并恢复状态存储；PARTITIONS_ASSIGNED表示线程已分配分区，但状态存储恢复尚未完成，此时无法正常访问状态存储。
• 自定义指示器会优先检查全局状态，再逐个校验线程状态，确保覆盖所有未就绪场景。

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