Kafka Streams：如何以编程方式判断状态恢复完成？

当然可以通过编程方式检测Kafka Streams的状态恢复完成状态，以下是几种实用方案，可直接对接OpenShift的就绪探针逻辑：

1. 注册StateListener监听状态切换 #

Kafka Streams提供了StateListener接口，能监听应用的状态变化。当应用从REBALANCING状态切换到RUNNING状态时，就意味着状态恢复（包括本地状态存储的加载、分区分配完成）已经完成。

示例代码：

KafkaStreams streams = new KafkaStreams(topology, streamsConfig);

streams.setStateListener((newState, oldState) -> {
    // 仅在重平衡完成后标记就绪
    if (oldState == KafkaStreams.State.REBALANCING && newState == KafkaStreams.State.RUNNING) {
        // 这里触发OpenShift就绪探针的"Alive"标记逻辑
        markReadinessProbeAsAlive();
    }
    // 应用停止或进入关闭流程时，标记探针为非就绪
    else if (newState == KafkaStreams.State.NOT_RUNNING || newState == KafkaStreams.State.PENDING_SHUTDOWN) {
        markReadinessProbeAsUnavailable();
    }
});

streams.start();

2. 暴露HTTP端点对接OpenShift就绪探针 #

结合状态监听逻辑，在应用内启动一个轻量HTTP端点，根据当前Kafka Streams的状态返回对应HTTP状态码，直接让OpenShift的就绪探针调用该端点：

示例伪代码（基于简单HTTP服务器实现）：

// 启动一个监听8080端口的HTTP服务
HttpServer readinessServer = HttpServer.create(new InetSocketAddress(8080), 0);
readinessServer.createContext("/ready", exchange -> {
    KafkaStreams.State currentState = streams.state();
    // 仅当状态为RUNNING时返回200，否则返回503
    int responseCode = (currentState == KafkaStreams.State.RUNNING) ? 200 : 503;
    exchange.sendResponseHeaders(responseCode, 0);
    exchange.close();
});
readinessServer.start();

之后在OpenShift的部署配置中，将就绪探针的httpGet路径设为/ready，端口设为8080即可——当端点返回200时，OpenShift会自动标记应用为"Alive"。

3. 利用AdminClient查询消费者组状态 #

如果需要更精细的恢复进度监控，可以通过Kafka的AdminClient查询应用对应的消费者组状态，确认所有成员都完成了分区分配并进入稳定状态：

示例代码：

AdminClient adminClient = AdminClient.create(streamsConfig);
String consumerGroupId = streamsConfig.getString(StreamsConfig.APPLICATION_ID_CONFIG);

// 查询消费者组详情
DescribeConsumerGroupsResult groupResult = adminClient.describeConsumerGroups(Collections.singleton(consumerGroupId));
ConsumerGroupDescription groupDesc = groupResult.all().get().get(consumerGroupId);

// 检查所有消费者成员是否处于稳定状态且已分配分区
boolean isRestored = groupDesc.members().stream()
    .allMatch(member -> member.state().equals(ConsumerGroupMemberState.STABLE) 
        && !member.assignment().topicPartitions().isEmpty());

if (isRestored) {
    markReadinessProbeAsAlive();
}

注意事项 #

• REBALANCING状态不仅会在应用启动时出现，运行中发生分区重平衡时也会触发，需根据业务场景判断是否需要在每次重平衡完成后都更新探针状态。
• 若使用Spring Boot等框架，可以直接借助Actuator的自定义健康指示器实现上述逻辑，无需手动搭建HTTP服务。

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