基于 ENV 的 Kafka 消息过滤答案

【问题标题】：Kafka Message Filtering Based on ENVs基于 ENV 的 Kafka 消息过滤
【发布时间】：2020-03-12 01:47:26
【问题描述】：

我有一个消费者应用程序部署在多个 ENV（开发、测试、阶段和预生产）上。他们都在消费同一个 Kafka 主题（意味着像同一个主题的多个消费者一样工作）。

我对所有 ENV（开发、测试、阶段和预生产）都有单独的生产者应用程序。在有效负载中生成消息时，它有一个字段来提及生产者的 ENV。

我们的要求是 - Dev ENV 的消费者应该只消费 Dev ENV 的生产者应用程序的消息。其他 ENV 也是如此。

我的问题是 - 我应该使用消费者端过滤吗？这能保证我们的要求吗？它将如何保证我们的要求？

提前致谢。

【问题讨论】：

标签： apache-kafka apache-kafka-streams

【解决方案1】：

对于如何处理此要求，您有多种选择。但是，我认为一般来说，为不同的环境设置一个主题并不是一个好主意。研究数据保护和访问权限这听起来不是一个好的设计。

无论如何，我看到了以下选项。

选项 1： 使用环境（dev, test, ...）作为topic的key，告诉consumer按key过滤。

选项 2： 编写生产者，将数据从每个环境发送到各个分区，并告诉每个环境的消费者只从特定分区读取。

但在实施选项 2 之前，我宁愿这样做 选项 3： 为每个环境设置一个主题，让生产者/消费者从不同的主题中写入/读取。

【讨论】：

谢谢。关于选项 1 的一个问题 - 如果消费者按键过滤，那么消息是否会被标记为未消费并让符合条件的消费者消费它？
这取决于消费者是否在同一个消费者组中。如果他们不在同一个消费者组中（我认为这会对您有所帮助），那么每个消费者都在自己管理偏移量，将过滤掉的消息留给其他消费者。
具有相同组 id 的消费者意味着他们在同一个组中。
是的，group.id 相同的消费者属于同一组。
非常感谢@mike。现在，我认为目前最简单的解决方案是根据不同的环境创建单独的消费者组。虽然不是一个好的解决方案......最好为每个环境设置不同的主题，我们目前无法做到这一点。

【解决方案2】：

我同意 mike 的观点，即跨环境使用单个主题不是一个好主意。

但是如果您要这样做，那么我建议您使用流处理器为您的消费者创建单独的主题。您可以在 Kafka Streams、ksqlDB 等中执行此操作。

ksqlDB 看起来像这样：

-- Declare stream over existing topic
CREATE STREAM FOO_ALL_ENVS WITH (KAFKA_TOPIC='my_source_topic', VALUE_FORMAT='AVRO'); 

-- Create derived stream & new topic populated with message just for DEV
-- You can explicitly provide the target Kafka topic name.
CREATE STREAM FOO_DEV WITH (KAFKA_TOPIC='foo_dev') AS SELECT * FROM FOO_ALL_ENVS WHERE ENV='DEV';

-- Create derived stream & new topic populated with message just for PROD
-- If you don't specify a Kafka topic name it will inherit from the 
-- stream name (i.e. `FOO_PROD`)
CREATE STREAM FOO_PROD AS SELECT * FROM FOO_ALL_ENVS WHERE ENV='PROD';
-- etc

现在您的生产者写入单个主题（如果必须），但您的消费者可以从特定于其环境的主题中消费。 ksqlDB 语句是连续查询，因此将处理源主题中的所有现有消息以及到达的每条新消息。

【讨论】：