Kafka Streams Binding: Store (prod-id-count-store) not available to Streams instance

Question

I'm trying to run interactive queries on Kafka Streams using Spring Cloud Stream Kafka Binder. I stuck on retrieving key value store from InteractiveQueryService. I'm keep getting always the same error on my code and even on code from examples:

https://github.com/spring-cloud/spring-cloud-stream-samples/tree/main/kafka-streams-samples/kafka-streams-interactive-query-basic

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

Code is from example class:

https://github.com/spring-cloud/spring-cloud-stream-samples/blob/main/kafka-streams-samples/kafka-streams-interactive-query-basic/src/main/java/kafka/streams/product/tracker/KafkaStreamsInteractiveQueryApplication.java

My application.properties file:

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

The only difference is Spring Boot version, Spring Cloud version and Java 17.

My build.gradle file:

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()
}

Kafka version: kafka_2.13-3.2.3

Can you please tell what I'm doing wrong, because I've wasted 2 days and have no ideas.

Answer 1

The fix is in the main branch, but not 3.2.x yet. However, you should be able to workaround the issue by setting:

spring.cloud.stream.kafka.streams.binder.configuration.application.server=<server>:<port>