apache-kafka - 验证 kafka 主题消息

Question

我正在与 kafka 合作，有人要求我验证发送给 Kafka 的消息，但我不喜欢我认为这就是为什么我希望有人能就此提出建议的解决方案。

我们有很多不受我们控制的生产者，所以他们可以发送任何格式的任何消息，我们最多可以发送8000万条记录，并且应该在2小时内处理。
我被要求:

验证格式(Json，因为它必须与 mongoDB 兼容)。

验证一些发送的字段。

重命名部分字段

最后 2 个请求将使用存储在 MongoDB 中的参数完成。
所有这些都应该在假设我们不是唯一一个制造消费者的情况下完成，因此应该对我们的服务进行“简单”调用以进行此验证。
有任何想法吗？

Answer 1

这通常通过 Kafka Streams 作业完成。

您有生产者发送事件的“原始”输入主题。然后 Streams 作业读取这些主题并将有效记录写入“干净”主题。在 Streams 中，您可以进行各种处理以检查记录或在需要时丰富它们。

您可能还想将坏记录写入死信队列主题，以便检查这些情况发生的原因。

然后，您的消费者可以读取干净的主题，以确保他们只能看到经过验证的数据。

此解决方案为记录增加了一些延迟，因为它们必须在到达消费者之前进行“处理”。您还希望在 Kafka 集群附近运行 Streams 作业，因为根据您要验证的程度，它可能需要摄取大量数据。

另见 Handling bad messages using Kafka's Streams API其中一些概念是详细的。