Spring-Cloud-Stream 仅在应用程序启动时实现的状态存储完全填充并准备就绪后处理 kafka 消息

Question

参考这个解决方案，我的Spring云流应用程序.yml文件具有以下配置：

#application.yml

spring.cloud.stream.bindings.input:
  destination: my-topic-name
  contentType: application/json
  consumer:
    useNativeDecoding: false
spring.cloud.stream.kafka.streams.bindings.input:
  consumer:
    keySerde: org.apache.kafka.common.serialization.Serdes$StringSerde
    valueSerde: org.apache.kafka.common.serialization.Serdes$StringSerde
    materializedAs: my-store

在我的主应用程序中，类注释为@EnableB

ReadOnlyKeyValueStore<Object, String> store;

 input.process(() -> new Processor<Object, Product>() {

                @Override
                public void init(ProcessorContext processorContext) {
                    store = (ReadOnlyKeyValueStore) processorContext.getStateStore("my-store");


                }

                @Override
                public void process(Object key, Object value) {
                    //find the key
                    store.get(key);
                }

                @Override
                public void close() {
                    if (state != null) {
                        state.close();
                    }
                }
            }, "my-store");

问题是，在应用程序第一次启动时，状态存储尚未完全填充，也尚未准备就绪（例如空状态存储），但消息仍然更早到达，并且正在由Kafka流拓扑处理，结果出乎意料。

我们如何确保在第一次启动应用程序时，在@StreamListener中定义的任何流处理拓扑开始处理传入消息之前，所有或特定(如果可能，用户可以定义)已被指示使用application.yml文件中的materializedAs进行具体化的状态存储已完全填充并准备就绪。我们可以强制消息的流处理一直等到应用程序第一次启动时状态存储被完全填充吗？

我试图通过修改其中一个 spring-cloud-stream 示例来复制该问题，并将修改后的版本推送到此处。有关此的更多详细信息讨论也可以在这里找到

Answer 1

您似乎需要使用GlobalKTable（状态存储）加入流，而不是使用进程接口。这里是Kstream join GlobalKTable。请尝试一下。