在Quarkus中使用Kafka Streams的疑问：正确性与状态检测

一、用法正确性判断 #

你的代码整体思路没问题，用KTable结合本地状态存储实现类数据库查询是Kafka Streams的典型用法，但有两个细节需要优化：

• 状态存储获取时机：streams.start()后直接拿keyValueStore不太稳妥，因为Streams启动后需要时间完成状态初始化和恢复，这时候直接获取大概率会抛出InvalidStateStoreException。建议通过setStateListener监听状态变为RUNNING后再获取存储，或者每次查询时动态调用streams.store(...)（更稳妥）。
• 发送与查询的一致性：你用Quarkus的Emitter发消息到topic，这部分是独立的生产者逻辑。KTable是基于topic日志异步构建状态的，所以发完消息立刻查询可能拿到旧值。如果要强一致性，建议用Kafka Streams的KStream来处理写入，或者接受异步同步的延迟（毕竟这是流处理的特性）。

二、如何检测Kafka Streams的异常状态 #

Kafka Streams提供了多种机制来监控状态和错误：

• 状态监听器：通过streams.setStateListener((newState, oldState) -> { ... })监听状态变化，比如从RUNNING跳到ERROR或NOT_RUNNING时，就可以触发告警或者执行恢复逻辑。
• 全局异常处理：配置StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG为自定义处理器或者官方的LogAndFailExceptionHandler，捕获处理过程中的反序列化等异常；还可以用streams.setUncaughtExceptionHandler处理未捕获的全局异常，避免线程崩溃。
• 指标监控：Quarkus集成了Micrometer，Kafka Streams会暴露一堆指标（比如kafka_streams_state、kafka_streams_task_count），用Prometheus这类工具监控这些指标，状态异常时就能及时发现。

三、关闭Kafka后读值不报错的原因 #

KTable的本地状态默认是持久化在磁盘的RocksDB里，Kafka集群宕机后，Streams会进入ERROR状态，但本地缓存依然能访问，所以keyValueStore.get(key)读的是本地数据，不会报错。但这时候状态不会再从Kafka同步新数据，发送的消息也发不出去，属于离线只读状态。

如果想在Kafka不可用时让读操作报错，可以在查询前先检查Streams状态：

private String getValue(String key) {
    if (streams.state() != KafkaStreams.State.RUNNING) {
        throw new StatusServiceException("Kafka Streams未处于运行状态，无法查询");
    }
    try {
        return streams.store(StoreQueryParameters.fromNameAndType(
            configurationProperties.kafkaStream().storeName(), 
            QueryableStoreTypes.keyValueStore()
        )).get(key);
    } catch (InvalidStateStoreException e) {
        throw new StatusServiceException(e);
    }
}

四、优化后的代码示例 #

初始化代码（修正状态监听与异常处理） #

public void start() {
    Properties properties = new Properties();
    properties.put(StreamsConfig.APPLICATION_ID_CONFIG, configurationProperties.kafkaStream().applicationId());
    properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServer);
    properties.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    properties.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
    // 配置反序列化异常处理器，遇到异常直接终止，避免脏数据
    properties.put(StreamsConfig.DEFAULT_DESERIALIZATION_EXCEPTION_HANDLER_CLASS_CONFIG, LogAndFailExceptionHandler.class);

    kTable = builder.table(topic, Materialized.as(configurationProperties.kafkaStream().storeName()));

    streams = new KafkaStreams(builder.build(), properties);

    // 监听状态变化
    streams.setStateListener((newState, oldState) -> {
        if (newState == KafkaStreams.State.ERROR) {
            System.err.println("Kafka Streams进入错误状态");
            // 这里可以加告警逻辑，比如发邮件、调用告警接口
        }
    });

    // 处理未捕获的全局异常
    streams.setUncaughtExceptionHandler((thread, throwable) -> {
        System.err.println("Kafka Streams线程[" + thread.getName() + "]出现未捕获异常");
        throwable.printStackTrace();
        // 可选：尝试重启Streams或者触发服务降级
    });

    streams.start();
}

读取代码（增加状态检查与动态获取存储） #

private String getValue(String key) {
    KafkaStreams.State currentState = streams.state();
    if (currentState != KafkaStreams.State.RUNNING) {
        throw new StatusServiceException("Kafka Streams当前状态为" + currentState + "，无法执行查询");
    }
    try {
        // 每次查询动态获取存储，避免状态变化导致存储失效
        KeyValueStore<String, String> store = streams.store(
            StoreQueryParameters.fromNameAndType(
                configurationProperties.kafkaStream().storeName(), 
                QueryableStoreTypes.keyValueStore()
            )
        );
        return store.get(key);
    } catch (InvalidStateStoreException e) {
        throw new StatusServiceException(e);
    }
}

public int getNumber() {
    final String numberStr = getValue("number");
    int number = numberStr != null && !numberStr.isEmpty() ? Integer.parseInt(numberStr) : 0;
    int newNumber = number + 1;
    emitter.send(Record.of("number", String.valueOf(newNumber)));
    return newNumber;
}

五、参考方向 #

• Quarkus官方文档的Kafka Streams章节，重点看状态管理和错误处理部分。
• Kafka Streams官方指南的「Interactive Queries」章节，详细讲了如何正确使用本地状态存储做查询。
• Quarkus提供的Kafka Streams示例项目，里面有状态监听、查询等常见场景的实现。

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