我如何确定在 Spring Cloud Stream API 中更新了 Keystore

Question

我使用 spring cloud stream api 的聚合函数从一个主题创建一个物化视图。 如下所示：

public void process(KStream<Object, Object> input){
input
  .peek((key, value) ->{...}
  .map((key, value) -> {...}
  .groupByKey()
  .windowedBy(TimeWindows.of(5000))
  .aggregate(Initializer, Aggregator, Materialized)

然后我查询我创建的 Statestore：

 ReadOnlyWindowStore<Object, Object> windowStore =
  queryService.getQueryableStoreType("test", QueryableStoreTypes.windowStore());

现在我的问题是，在 process 方法处理了新事件后，如何确定此 statestore 是否已更新？他们的某种事件是我可以听还是我可以创建一个？

Answer 1

你的程序是：

input
  .peek((key, value) ->{...}
  .map((key, value) -> {...}
  .groupByKey()
  .windowedBy(TimeWindows.of(5000))
  .aggregate(Initializer, Aggregator, Materialized)

实际上，最后一个aggregate() 返回一个KTable 对象。如果您通过Materialized 禁用缓存，您可以通过以下方式了解KTable 的每一次更新：

input
  .peek((key, value) ->{...}
  .map((key, value) -> {...}
  .groupByKey()
  .windowedBy(TimeWindows.of(5000))
  .aggregate(Initializer, Aggregator, Materialized) // disable caching via Materialized
  .toStream()
  .foreach(...) // react to every update to the KTable