java - 将数据存储到 GlobalKTable

Question

我无法理解如何将数据保存到 globalKTable。我尝试使用

在 Java 中创建 GlobalKTable


GlobalKTable globalTable = builder.globalTable(inputTopic, Materialized.>as("global-store"));


在此之后我无法找到任何直接将数据存储到此 GlobalKTable 的示例。它在任何地方都与连接或流一起使用。

Answer 1

由于每个 Stream 应用程序实例加载整个 GlobalKTable 数据，它默认禁用记录到 changelog 主题，并且 GlobalKTable 使用输入主题作为恢复过程的更改日志源(以实现容错)，因此更改全局存储没有意义，您不能这样做。

您可以使用 ProcessorContext.getStateStore("global-store") 获得对 Processor API 中全局状态的只读访问权限，而无需将存储添加到 Processor.

实现此目的的一种方法是将要进行的更新推送到 inputTopic，以便它更新所有应用程序实例中的每个 GlobalKTable。