Kafka Streams绑定异常：prod-id-count-store无法被Streams实例访问

问题描述 #

使用Spring Cloud Stream Kafka Binder执行Kafka Streams交互式查询时，从InteractiveQueryService获取键值存储prod-id-count-store始终失败，即使直接使用官方示例代码也会抛出UnknownStateStoreException。

错误栈信息 #

java.lang.IllegalStateException: Error retrieving state store: prod-id-count-store
    at org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService.lambda$getQueryableStore$0(InteractiveQueryService.java:115) ~[spring-cloud-stream-binder-kafka-streams-3.2.5.jar:3.2.5]
    at org.springframework.retry.support.RetryTemplate.doExecute(RetryTemplate.java:329) ~[spring-retry-1.3.1.jar:na]
    at org.springframework.retry.support.RetryTemplate.execute(RetryTemplate.java:209) ~[spring-retry-1.3.1.jar:na]
    at org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService.getQueryableStore(InteractiveQueryService.java:88) ~[spring-cloud-stream-binder-kafka-streams-3.2.5.jar:3.2.5]
    at com.nowatel.teacher.service.KafkaStreamsInteractiveQueryApplication$InteractiveProductCountApplication.printProductCounts(KafkaStreamsInteractiveQueryApplication.java:71) ~[main/:na]
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke0(Native Method) ~[na:na]
    at java.base/jdk.internal.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:77) ~[na:na]
    at java.base/jdk.internal.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) ~[na:na]
    at java.base/java.lang.reflect.Method.invoke(Method.java:568) ~[na:na]
    at org.springframework.scheduling.support.ScheduledMethodRunnable.run(ScheduledMethodRunnable.java:84) ~[spring-context-5.3.23.jar:5.3.23]
    at org.springframework.scheduling.support.DelegatingErrorHandlingRunnable.run(DelegatingErrorHandlingRunnable.java:54) ~[spring-context-5.3.23.jar:5.3.23]
    at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:539) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.runAndReset$$$capture(FutureTask.java:305) ~[na:na]
    at java.base/java.util.concurrent.FutureTask.runAndReset(FutureTask.java) ~[na:na]
    at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:305) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) ~[na:na]
    at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) ~[na:na]
    at java.base/java.lang.Thread.run(Thread.java:833) ~[na:na]
Caused by: org.apache.kafka.streams.errors.UnknownStateStoreException: Store (prod-id-count-store) not available to Streams instance
    at org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService.getStateStoreFromKafkaStreams(InteractiveQueryService.java:126) ~[spring-cloud-stream-binder-kafka-streams-3.2.5.jar:3.2.5]
    at org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService.lambda$getQueryableStore$0(InteractiveQueryService.java:109) ~[spring-cloud-stream-binder-kafka-streams-3.2.5.jar:3.2.5]
    ... 17 common frames omitted

配置信息 #

application.properties #

spring.cloud.stream.kafka.streams.binder.configuration.commit.interval.ms=1000
spring.cloud.stream.bindings.process-out-0.destination=product-counts
spring.cloud.stream.bindings.process-in-0.destination=products
spring.application.name=kafka-streams-iq-basic-sample

build.gradle #

plugins {
    id 'org.springframework.boot' version "2.7.4"
    id 'io.spring.dependency-management' version "1.0.14.RELEASE"
    id 'java'
}

group = 'com.example'
version = '0.0.1-SNAPSHOT'
sourceCompatibility = javaVersion

repositories {
    mavenCentral()
}

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-webflux'
    implementation 'org.apache.kafka:kafka-streams'
    implementation 'org.springframework.cloud:spring-cloud-stream'
    implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka'
    implementation 'org.springframework.cloud:spring-cloud-stream-binder-kafka-streams'
    implementation 'org.springframework.kafka:spring-kafka'

    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testImplementation 'org.springframework.kafka:spring-kafka-test'
    testImplementation 'io.projectreactor:reactor-test'
}

dependencyManagement {
    imports {
        mavenBom "org.springframework.cloud:spring-cloud-dependencies:2021.0.4"
    }
}

tasks.named('test') {
    useJUnitPlatform()
}

环境版本 #

• Spring Boot: 2.7.4
• Spring Cloud: 2021.0.4
• Java: 17
• Kafka: kafka_2.13-3.2.3

解决方案 #

1. 确保状态存储名称完全匹配 #

检查Kafka Streams拓扑中定义的状态存储名称与查询时使用的prod-id-count-store完全一致，包括大小写、特殊字符，避免拼写错误。

2. 等待Kafka Streams实例完全启动 #

错误核心原因是Streams实例未完成初始化就执行查询，需监听KafkaStreamsStateChangedEvent，确认实例进入RUNNING状态后再执行查询：

@EventListener
public void handleKafkaStreamsStateChange(KafkaStreamsStateChangedEvent event) {
    if (event.newState() == KafkaStreams.State.RUNNING) {
        // 在此执行交互式查询或标记查询可用
    }
}

3. 验证依赖版本兼容性 #

Spring Cloud 2021.0.4对应的Spring Cloud Stream Kafka Streams Binder版本为3.2.x，需确保Kafka 3.2.3与该版本兼容。同时确认所有依赖均支持Java 17（Spring Boot 2.7.x原生支持Java 17，但部分老依赖可能存在兼容问题）。

4. 确认状态存储已正确注册到拓扑 #

检查拓扑代码中是否正确创建并关联状态存储：

// 创建状态存储
StoreBuilder<KeyValueStore<String, Long>> storeBuilder = Stores.keyValueStoreBuilder(
    Stores.persistentKeyValueStore("prod-id-count-store"),
    Serdes.String(),
    Serdes.Long());
streamsBuilder.addStateStore(storeBuilder);

// 确保处理器关联该存储
streamsBuilder.stream("products")
    .process(() -> new ProductCountProcessor(), "prod-id-count-store");

5. 补充必要的Kafka Streams配置 #

添加以下配置确保状态存储正常初始化：

spring.cloud.stream.kafka.streams.binder.configuration.application.id=${spring.application.name}
spring.cloud.stream.kafka.streams.binder.configuration.state.dir=/tmp/kafka-streams-state

确保application.id唯一，状态目录有读写权限。

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