KafkaStreams interactive queries with Spring-Kafka

Question

I am writing a Spring Boot app with Spring Kafka. As my app is focused on Kafka Streams and I need to use interactive queries and query my state stores I am wondering: is there any particular way to access kafka streams state stores with Spring Kafka?

From what I've seen there is some support in Spring Cloud Stream Binder Kafka Streams for interactive queries but I cannot find anything about them in Spring Kafka. Am I missing something or there is no support for it in Spring-Kafka?

And if so - is there anything in particular I should have in mind while creating my own version of org.springframework.cloud.stream.binder.kafka.streams.InteractiveQueryService ?

It seemed a little too much to include Spring Cloud Streams when all I use is only Kafka and Kafka Streams but if interactive query support provided there would be difficult to implement on my own maybe it is recommended to include it anyway? I would appreciate any advise.

AFAIK, spring-kafka is just for producer/consumer API, not Kafka Streams — OneCricketeer
Note: spring-cloud-stream-binder-kafka-streams includes spring-kafka, so there is no reason you couldn't use them together — OneCricketeer
spring-kafka supports Kafka Stream. docs.spring.io/spring-kafka/docs/2.2.3.RELEASE/reference/html/…. — Steephen
It is not difficult to implement your own InteractiveQueryService by accessing the underlying KafkaStreams object if you want to avoid using Spring Cloud Stream for this. Spring Cloud Stream provides idiomatic usage of accessing the state store, but if you don't have the requirement of going to a micro services based model, then I understand that you might want to use spring-kaka directly. — sobychacko
Yes, if are not using Spring cloud Stream, in that case, you need to programmatically access the state store and interact with it. Thats where using the binder might help you there. — sobychacko

FrancescoM FrancescoM · Accepted Answer · 2019-11-05T11:36:29

Well if you use Spring-Kafka and Kafka-Streams, i suggest you to define your KafkaStreams as @Bean so spring will take care of the lifecycle of the object, having something like

@Configuration
public class KafkaConfig{
  @Bean
  KafkaStreams ingestAndAggregateStream(KafkaProperties kafkaProps){
  Topology t....
  KafkaStreams stream = new KafkaStreams(t,kafkaProps.buildStreamsProperties())
  stream.start();
  return stream;
  }
}

Once you do that you can autowire the stream in the controller and use it to retrieve and query the LOCAL KVStore

@RestController
public class StreamingController {

  private final KafkaStreams ingestAndAggregateStream;
  public StreamingController(KafkaStreams ingestAndAggregateStream){
    this.ingestAndAggregateStream = ingestAndAggregateStream;
  }

  @RequestMapping(value = "/queryStore", method = RequestMethod.GET)
  @ResponseBody
  public Value getKVStoreValue(@RequestParam String key){
    ReadOnlyKeyValueStore<String, Value> store = ingestAndAggregateStream
        .store("KV_STORE_NAME",
            QueryableStoreTypes.<String, Value>keyValueStore());
    return store.get(key);
  }
}

KafkaStreams interactive queries with Spring-Kafka

1 Answers