Spring 和 Kafka 流 - 如何使用查询 API

Question

我是 kafka 和 kafka 流的新手。我有一个与 kafka 生产者、消费者、KStream 和 KTable 一起工作的基本 Spring 服务。现在，我想检查我的 KTable 记录，所以为了实现它，我正在尝试使用 Kafka Query API。

这可以通过以下方式实现（没有 Spring 集成）：

KafkaStreams streams = new KafkaStreams(topology, config);
// Get access to the custom store
MyReadableCustomStore<String,String> store = streams.store("the-custom-store", new MyCustomStoreType<String,String>());
// Query the store
String value = store.read("key");

现在，我尝试使用基于 Spring 的 InteractiveQueryService 进行查询。但我在 Spring 引导中遇到了一些依赖问题。

在 Spring 中使用 kafka 查询 API 的最佳方法是什么？

我的服务中的 Spring kafka 配置如下所示：

@Bean("streamsBuilder")
public StreamsBuilderFactoryBean recordsStreamBuilderFactoryBean() {
    Map<String, Object> config = new HashMap<>();
    // set some properties
    return new StreamsBuilderFactoryBean(new KafkaStreamsConfiguration(config));
}

你能建议吗？

Answer 1

这是一个 Spring Boot 应用程序，用于展示如何...

@SpringBootApplication
@EnableKafkaStreams
public class So58918956Application {

    public static void main(String[] args) {
        SpringApplication.run(So58918956Application.class, args);
    }

    @Bean
    public CountDownLatch latch(StreamsBuilderFactoryBean streamsBuilderFB) {
        CountDownLatch latch = new CountDownLatch(1);
        streamsBuilderFB.setStateListener((newState, oldState) -> {
            if (State.RUNNING.equals(newState)) {
                latch.countDown();
            }
        });
        return latch;
    }

    @Bean
    public KTable<String, String> table(StreamsBuilder streamsBuilder) {
        Serde<String> serde = Serdes.String();
        KTable<String, String> table = streamsBuilder.table("so58918956",
                Consumed.with(serde, serde)
                        .withOffsetResetPolicy(AutoOffsetReset.EARLIEST), 
                Materialized.as("the-custom-store"));
        return table;
    }

    @Bean
    public ApplicationRunner runner(StreamsBuilderFactoryBean streamsBuilderFB,
            KafkaTemplate<String, String> template, KTable<String, String> table) {

        return args -> {
            template.send("so58918956", "key", "value");
            latch(streamsBuilderFB).await(10, TimeUnit.SECONDS);
            ReadOnlyKeyValueStore<String, String> store = streamsBuilderFB.getKafkaStreams().store(
                    table.queryableStoreName(),
                    QueryableStoreTypes.keyValueStore());
            System.out.println(store.get("key"));
        };
    }

    @Bean
    public NewTopic topic() {
        return TopicBuilder.name("so58918956").partitions(1).replicas(1).build();
    }

}