KTable在Spring Boot应用程序中不返回数据，但可以查询

Question

我有一个使用Kafka流的Spring Boot应用程序。我有一个包含一些金融货币报价的KTable，它是这样创建的：

@Bean(name = "indicativeQuotes")
public KTable<String, Quote> quoteKTable(StreamsBuilder streamsBuilder) {
    return streamsBuilder.table(quoteTopicName,
            Materialized.<String,Quote,KeyValueStore<Bytes,byte[]>>as("quoteTable")
                    .withKeySerde(Serdes.String())
                    .withValueSerde(new JsonSerde<>(Quote.class)));
}

我@autowire这个bean在另一个组件中，并用以下代码测试它：

@Autowired
private KTable<String, Quote> indicativeQuotes;

@PostConstruct
private void postConstruct() {
    doPrint();
}

public void doPrint() {
        ReadOnlyKeyValueStore<String, Quote> store = streamsBuilderFactoryBean.getKafkaStreams().store("quoteTable", QueryableStoreTypes.keyValueStore());
        store.all().forEachRemaining(keyValue -> log.info("Key: " + keyValue.key + " Value: " + keyValue.value));
        indicativeQuotes.foreach((k,v) -> log.info(k));}

当通过store查询时，代码记录了正确的值，但它在foreach（）中没有输出任何内容，就好像like表是空的一样。我还尝试了print（）和其他选项--所有选项都不输出任何内容，没有任何例外。

我的用例是，我有一个预定的Quartz作业，它应该在触发时将KTable的当前状态写入Kafka主题，如下所示：

@Override
protected void executeInternal(JobExecutionContext jobExecutionContext) throws JobExecutionException {
    TriggerKey triggerKey = jobExecutionContext.getTrigger().getKey();
    log.info("Job was triggered by: {}", triggerKey.getName());

    indicativeQuotes.filter((key, value) -> key.equals(triggerKey.getName()))
            .mapValues(quoteToCourseFixedMapper)
            .toStream()
            .peek((instrument, course)-> log.info("Sending courses for instrument: {}, {}", instrument, course))
            .to(quoteEventTopicName);
}

但我认为这段代码不起作用，因为它不是处理拓扑的一部分，而且我不能只是按需从Ktable中获取数据。我在这里有点困惑，当然我可以在事件触发时通过store查询数据，但是也许对于这样的用例有更好的模式？基本上，我感兴趣的是，是否有可能合并这个触发的作业事件作为处理管道的一部分。

Answer 1

如果只想将更新发布到另一个主题，请将KTable转换为KStream，并使用to（）函数。

KTable ktable = ...;
KStream ksteram = ktable.toStream();
kstream.to("topic", Produces.with(keySerde, valueSerde))

主题将包含该表的更改日志。

显然，由于一些生命周期相关的概念，您不能仅仅注入(@autowire)kstream/ktable。您应该尽可能保持与KafkaStreams相关的代码内联。