MirrorMaker队列名称不一致解决

Posted 2023-05-09

参考技术A Kakfa MirrorMaker是Kafka 官方提供的跨数据中心的流数据同步方案，其实现原理是通过从Source Cluster消费消息，然后将消息生产到Target Cluster，即普通的消息生产和消费。用户只要通过简单的consumer配置和producer配置，启动Mirror，就可以实现准实时的数据同步。

简单的说，这个工具就是启动consumer消费旧集群中topic的数据，然后启动producer直接发送到新集群中的topic。

MirrorMaker同步队列时要求topic的名字必须一样。

经过在github中进行搜索，发现开源项目：mirrormaker_topic_rename已经满足项目需求 https://github.com/opencore/mirrormaker_topic_rename

mvn clean package

上传mmchangetopic-1.0-SNAPSHOT.jar至目录：/kafka/libs

配置文件目录：/kafka/config

相关配置文件：consumer.properties和producer.properties

consumer.properties => 源集群配置[A]

producer.properties => 目标集群配置[B]

消息流向：消息写入A集群后会被复制到目标集群B中。

./kafka-mirror-maker.sh --consumer.config ../config/consumer.properties --producer.config ../config/producer.properties --whitelist .* --message.handler com.opencore.RenameTopicHandler --message.handler.args 'a,b'

nohup ./kafka-mirror-maker.sh --consumer.config ../config/consumer.properties --producer.config ../config/producer.properties --whitelist .* --message.handler com.opencore.RenameTopicHandler --message.handler.args 'a,b' &

nohup bin/kafka-mirror-maker.sh --consumer.config MirrorMaker/consumer.properties --producer.config MirrorMaker/producer.properties --whitelist ' test' &

nohup ./bin/kafka-mirror-maker.sh --new.consumer --consumer.config config/mirror-consumer.properties --num.streams 40 --producer.config config/mirror-producer.properties --whitelist 'ABTestMsg|AppColdStartMsg|BackPayMsg|WebMsg|GoldOpenMsg|BoCaiMsg' &

kafka-mirror-maker --consumer.config consumer.properties --producer.config producer.properties --whitelist test_.* --message.handler com.opencore.RenameTopicHandler --message.handler.args test_source,test_target;test_source2,test_target2

bin/kafka-mirror-maker.sh --consumer.config consumer.properties --producer.config producer.properties --whitelist “my-topic1,my-topic2”

如需要调试，查看debug日志 可以修改/kafka/config/tools-log4j.properties

https://blog.csdn.net/gdutliuyun827/article/details/55263245

bootstrap.servers=10.255.30.152:9092

group.id=mirrorGroupTest

auto.offset.reset=earliest

partition.assignment.strategy=org.apache.kafka.clients.consumer.RoundRobinAssignor

request.timeout.ms=50000

heartbeat.interval.ms=30000

session.timeout.ms=40000

max.poll.records=20000

receive.buffer.bytes=524288

max.partition.fetch.bytes=5248576

bootstrap.servers=10.255.30.132:9092

compression.type=none

batch.size=16384

retries=3

acks=1

producer.type=sync

如何将自定义 jar 文件添加/复制到 Kafka MirrorMaker 中？

【中文标题】如何将自定义 jar 文件添加/复制到 Kafka MirrorMaker 中？

【英文标题】：How to add / copy a custom jar file into KafkaMirrorMaker?

【发布时间】：2022-01-03 10:03:12

【问题描述】：

我正在开发一些自定义主题名称映射，并从中生成了一个 jar 文件。

由于我使用的是 MirrorMaker v1，因此这些变量也已添加到 KafkaMirrorMaker yaml 文件中的 KAFKA_MIRRORMAKER_MESSAGE_HANDLER 和 KAFKA_MIRRORMAKER_MESSAGE_HANDLER_ARGS。

但我不知道如何将此自定义 jar 文件物理添加到 KafkaMirrorMaker pod 中。我检查了 KafkaMirrorMaker 的 CRD，但还没有找到任何线索。

那么，有没有办法让 KafkaMirrorMaker 下载一些文件/工件并将 jar 文件包含到类路径中，以便自定义 MessageHandler 可以找到它？

helm install 命令用于部署 Mirror Maker。我目前使用的KafkaMirrorMaker的apiVersion：kafka.strimzi.io/v1beta2

【问题讨论】：

如何运行 KafkaMirrorMaker？ thisbin/kafka-run-class.sh kafka.tools.MirrorMaker ... 之类的东西与我们分享您的 Dockerfile

我们没有 Dockerfile。我们只有 helm yaml 文件。是否可以只修改这个文件来指示Mirror Maker下载jar文件？

我认为不可能。 Java 应用程序需要更复杂的方法来加载外部 jar。如果它部署在 kubernetes 上，则应该存在 Dockerfile。如果你找到它，添加任何文件（jar）会很容易

【参考方案1】：

基于strimzi 标签，我假设您使用的是Strimzi 的Mirror Maker v1？要添加您自己的 JAR，您需要构建一个自定义容器映像。

您可以修改 Strimzi 项目源代码并从头开始构建所有内容（您可以将您的 JAR 作为依赖项添加到 `docker-images/kafka/... 中的 3rd 方库）。但这在构建整个项目时相当复杂。

最简单的方法是编写自己的 Dockerfile 并使用现有的 Strimzi 镜像作为基础镜像。例如：

FROM quay.io/strimzi/kafka:0.26.0-kafka-3.0.0
USER root:root
COPY ./my-jar.jar /opt/kafka/libs/my-jar.jar
USER 1001

您可以构建此 Dockerfile 并将其推送到您自己的 Docker 注册表（Docker Hub、Quay，无论您使用什么）。您应该根据您使用的 Strimzi 版本和您使用的 Kafka 版本确保 FROM 使用正确的图像。

一旦你有了它，你必须告诉 Strimzi 使用这个图像。您可以使用KafkaMirrorMaker 自定义资源中的.spec.image 选项来执行此操作。或者您可以在 Strimzi Cluster Operator 部署中更改环境变量 STRIMZI_KAFKA_MIRROR_MAKER_IMAGES 并更新应该在那里使用的图像。

【讨论】：

谢谢。我将图像托管在我们的内部服务器中。如何配置 imagePullSecret？

嗯...我尝试设置 STRIMZI_IMAGE_PULL_SECRETS 环境变量，但不知何故它没有反映在部署对象中。

啊...好吧。环境变量必须在 Strimzi Cluster 操作员的部署中设置，而不是在 MirrorMaker CR 实例中设置。谢谢你的帮助！ :)