apache-kafka - 如何将一些消息从一个 kafka 主题复制到另一个 bash？

Question

请帮忙

我们有 2 个 kafka 主题。我想将 10 条消息从主题 1 开始复制到主题 2。

我尝试使用 kafka-console-consumer 和 kafka-console-producer

首先，我将 10 条消息从 topic1 保存到某个目录:

for (( i=1; i<=10; i++ )); do bin/kafka-console-consumer.sh --bootstrap-server 1.1.2.3:9092 --group CONSUMER1 --topic TOPIC1 --max-messages 1 > /tmp/_topic/$i.msg; done;

然后我尝试使用 kafka-console-producer 将其发送到 topic2:

for (( i=1; i<=10; i++ )); do  bin/kafka-console-producer.sh --broker-list 1.1.2.4:9092 --topic TOPIC2 < /tmp/_topic/$i.msg; done;

我遇到了错误 - 我的服务无法反序列化数据。
我的问题是:

我的解决方案行得通吗？

为什么我可以收到这个错误？

将消息从一个主题复制到另一个主题一次的最佳方法是什么？

更新:
我是如何解决这个问题的(感谢:Robin Moffatt):
我使用 kafka-mirror 和这个 jar :
https://github.com/opencore/mirrormaker_topic_rename
有了这个，我可以将消息从一个主题 kafka 复制到一个集群上的另一个主题

Answer 1

你可以用 kafkacat 来做到这一点:

kafkacat -b localhost:9092 -C -t source-topic -K: -e -o beginning -c10 | \
kafkacat -b localhost:9092 -P -t target-topic -K: 

|将第一个 kafkacat(即 -C 消费者)的输出重定向到第二个 kafkacat(即 -P 生产者)的输入

-c10意味着只消耗 10 条消息

-o beginning表示从主题的开头开始。

请注意，如果您有二进制数据(例如 Avro)，这将不起作用。要正确执行此操作，请使用 Replicator 或 MirrorMaker2 和 ByteArrayConverter 之类的东西。

引用: https://rmoff.net/2019/09/29/copying-data-between-kafka-clusters-with-kafkacat/