Kafka 主题到多个 kafka 主题调度程序（同一个集群）答案

【问题标题】：Kafka topic to multiple kafka topics dispatcher (same cluster)Kafka 主题到多个 kafka 主题调度程序（同一个集群）
【发布时间】：2020-04-12 07:54:32
【问题描述】：

我的用例如下：我有一个 kafka 主题 A，其中的消息“逻辑上”属于不同的“服务”，我不处理系统将消息发送到 A。

我想根据描述服务的一个列（格式为 CSV-风格，但没关系）。

服务集是静态的，我暂时不需要处理添加/删除。

我希望使用KafkaConnect 来执行此类任务，但令人惊讶的是，没有 Kafka 源/接收器（我找不到票，但它们已被拒绝）。

我见过MirrorMaker2，但对于我的（简单）用例来说，这看起来有点过头了。

我也知道KafkaStreams，但我宁愿不为此编写和维护代码。

我的问题是：有没有一种方法可以在不自己编写 kafka 消费者/生产者的情况下使用 kafka 原生工具来实现这个主题调度？

PS：如果有人认为MirrorMaker2 很适合我，我也很感兴趣，我不太了解这个工具。

【问题讨论】：

【解决方案1】：

据我所知，没有直接的方法可以根据传入的消息将传入的主题消息分支到主题列表。您需要编写自定义代码来实现这一点。

context.forward(key, value, To.child("所选主题"))

【讨论】：

【解决方案2】：

Mirror Maker 是用来做……镜像的。当您想要将一个集群从一个数据中心镜像到另一个具有相同主题的集群时，它非常有用。您的用例不同。

Kafka Connect 用于通过 Kafka 主题同步不同的系统（例如来自数据库的数据），但我也没有在这个用例中看到它。

我会为此使用 Kafka Streams 应用程序。

【讨论】：

【解决方案3】：

所有其他答案都是正确的，在撰写本文时，我确实在 Kafka 工具集中找到了任何“仅配置”解决方案。

最后的诀窍是使用Logstash，因为它的“kafka 输出插件”支持topic-id 参数中的 jinja 变量。

因此，一旦您在字段中提供了“目标主题名称”（比如 service_name），就这么简单：

output {
  kafka {
    id => "sink"
    codec => [...]
    bootstrap_servers => [...]
    topic_id => "%{[service_name]}"
    [...]
  }
}

【讨论】：