使用 Kafka Streams 进行简单分类

Question

我目前正在尝试找到一种直接且高效的方法来使用 Kafka Streams 对记录进行分类。
所有记录都至少包含一个id 和一个failed 属性。
（id 只是一个字符串，failed 是布尔值）

这个想法是在一开始将所有传入的记录分类为“消息”。
一旦传入记录之一设置了失败字段，这应该在某处“持久化”，并且记录应该被归类为“失败”。

从现在开始，具有相同id 的传入记录也应归类为“失败”，无论是否设置了failed 属性。

我正在考虑使用 Kafka Streams 的内部状态存储（连同交互式查询功能）或每次输入记录时都会查询的外部数据库。我认为 Kafka 本身的状态存储听起来像更轻量级的解决方案。

这里有一个小概念草图，希望能帮助理解这个问题。

有人知道如何以正确的方式解决这个问题吗？

谢谢 一切顺利 - 蒂姆

Answer 1

你的方法对我来说听起来不错。不要认为你需要 IQ 功能。只需定义一个自定义 Transformer 并将键值存储附加到它。在处理过程中，如果您收到带有failed=true 的消息，则将 ID 放入存储中。对于每条带有failed=false 的传入消息，您还需要检查存储以检查是否存在具有相同 ID 的先前失败消息。

要保留失败的消息，您只需将流分成两部分（可能使用branch() 并将failed 消息写入一个特殊主题。