Kafka Ktable changelog（使用toStream（））在同时到达多个具有相同密钥的消息时丢失了一些Ktable更新

Question

我有一个输入流，我用它来创建一个KTABLE。然后使用toStream（）方法，使用ktable changelog创建一个输出流。问题是toStream（）方法创建的流不包含更新了KTABLE的输入流中的所有消息。下面是我的代码：

final KTable<String, event> KTable = inputStream.groupByKey().aggregate(() -> null,
      aggregateKtableMethod,
      storageConf);

KStream<String, event> outputStream = KTable.toStream();

我想这是因为Ktable的更新是通过一些批处理操作进行的，但我找不到任何与之相关的配置或文档。是这些丢失事件的原因吗？你知道如何更改配置以便我不会丢失任何消息吗？

Answer 1

我找到解决办法了。问题出在我用来创建ktable的“StorageConf”中，缓存是可以的。我只需要禁用它，函数是：

storageConf.withCachingDisabled();

final KTable<String, event> KTable = inputStream.groupByKey().aggregate(() -> null,
  aggregateKtableMethod,
  storageConf);

现在我在输出流中有了我的所有事件。