如何在Spring Cloud Stream Kafka Streams应用程序中使用StateStoreBuilder添加StateStore

Question

原生 Kafka API 允许我们使用 StreamsBuilder 创建和添加状态存储：

    final StreamsBuilder builder = new StreamsBuilder();
    ...
    final StoreBuilder<WindowStore<String, Long>> dedupStoreBuilder = Stores.windowStoreBuilder(
            Stores.persistentWindowStore(storeName,
                                         retentionPeriod,
                                         windowSize,
                                         false
            ),
            Serdes.String(),
            Serdes.Long());

    builder.addStateStore(dedupStoreBuilder);

我想使用Spring Cloud Streams做同样的事情，但无法弄清楚访问StreamsBuilder以添加商店的方法。

我已尝试检索文档中所述的StreamsBuilderFactoryBean，希望可以从中获取StreamsBuilder对象，但该bean似乎不可用：

@EnableBinding(KafkaStreamsProcessor::class)
class FraudKafkaStreamsConfiguration(private val context: ApplicationContext) {

    @StreamListener
    @SendTo("output")
    fun process(@Input("input") input: KStream<String, TransferEmitted>): KStream<String, TransferEmitted> {

        val streamsBuilderFactoryBean = context.getBean("&stream-builder-process", StreamsBuilderFactoryBean::class.java)
        ...
        return xxx

    }

}

org.springframework.beans.factory.异常：没有名为流构建器进程的bean可用

无论如何，我什至不确定这是正确的方法。那么，我们如何以编程方式创建 StateStore？

Answer 1

由于我的 Scs 版本（Fishtown SR3），我没有看到记录的过程，但好消息是，自日耳曼敦以来，可以以声明方式创建状态存储：

const val DEDUP_STORE = "dedup-store"

@EnableBinding(KafkaStreamsProcessor::class)
class FraudKafkaStreamsConfiguration {

    @KafkaStreamsStateStore(name = DEDUP_STORE, type = KafkaStreamsStateStoreProperties.StoreType.KEYVALUE)
    @StreamListener
    @SendTo("output")
    fun process(@Input("input") input: KStream<String, TransferEmitted>): KStream<String, TransferEmitted> {
        return input.transform(TransformerSupplier { DeduplicationTransformer() }, DEDUP_STORE)

    }

}