apache-kafka - 验证 kafka 主题消息

Question

我正在使用 kafka，并且有人要求我对发送给 Kafka 的消息进行验证，但我不喜欢我认为的解决方案，这就是为什么我希望有人能在这方面给我建议。

我们有许多不受我们控制的生产者，所以他们可以发送任何格式的任何消息，我们可以发送多达 8000 万条记录，它们应该在 2 小时内处理完毕。我被要求：

• 验证格式（Json，因为它必须与 mongoDB 兼容）。

• 验证发送的一些字段。

• 重命名一些字段

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

标签： apache-kafka

Answer 1

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

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

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

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

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

另请参阅使用 Kafka 的 Streams API 处理错误消息，其中详细介绍了其中一些概念。