通过代码/api重置Kafka流应用程序

Question

我想知道用Kafka Streams执行这种操作的最佳方法是什么。

我有一个 Kafka 流和一个 KGlobal 表，让我们说产品 （1.000.000 消息） 和类别逻辑表 （10 msg）。每当新消息到达主题类别LogicBlobTable时，我需要重新处理所有将新到达的消息应用于产品的产品，并且输出将转到第三个主题。

我在考虑使用Kafka . tools . streams reset逻辑，并以某种方式挂钩我的代码，以停止kafkaStream，运行reset并再次启动流。

第二种选择是没有kafka流，而只有两个消费者和一个生产者。通过这种方式，我可以使用方法消费者.seekToBegining（集合.emptyList（））;

Answer 1

在这种情况下，重置KafkaStreams应用程序将导致大量重复输出。假设流中有10条记录，表中有5条记录，在处理时会生成3条输出记录。现在，将第6条记录添加到表中，并重新读取整个流。因此，您将把前3条输出记录重新发送到输出主题，如果某些记录也连接到新添加的第6条表记录，则可能还会发送其他输出记录。这似乎不是你想要的。

我猜你需要手动使用KafkaConsumer/KafkaProducer。