kafka跨地区跨集群同步工具MirrorMaker2 —— 筑梦之路

MM2简介 

KIP-382: MirrorMaker 2.0 - Apache Kafka - Apache Software Foundation

有四种运行MM2的方法：

As a dedicated MirrorMaker cluster.（作为专用的MirrorMaker群集）
As a Connector in a distributed Connect cluster.（作为分布式Connect群集中的连接器）
As a standalone Connect worker.（作为独立的Connect工作者）
In legacy mode using existing MirrorMaker scripts.（在旧模式下，使用现有的MirrorMaker脚本。）

MM2集群模式

配置文件示例1：

name = event-center-connector
connector.class = org.apache.kafka.connect.mirror.MirrorSourceConnector
tasks.max = 2# 定义集群别名
clusters = event-center, event-center-new# 设置event-center集群的kafka地址列表
event-center.bootstrap.servers = source:9193
event-center.security.protocol=SASL_PLAINTEXT
event-center.sasl.mechanism=PLAIN
event-center.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-pwd";
# 设置event-center-new集群的kafka地址列表
event-center-new.bootstrap.servers = target:29092
event-center-new.security.protocol=SASL_PLAINTEXT
event-center-new.sasl.mechanism=PLAIN
event-center-new.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-pwd";
# 开启event-center集群向event-center-new集群同步
event-center->event-center-new.enabled = true
# 允许同步topic的正则
event-center->event-center-new.topics = projects.*
event-center->event-center-new.groups = .*# MM2内部同步机制使用的topic，replication数量设置
checkpoints.topic.replication.factor=1
heartbeats.topic.replication.factor=1
offset-syncs.topic.replication.factor=1
offset.storage.replication.factor=1
status.storage.replication.factor=1
config.storage.replication.factor=1# 自定义参数
# 是否同步源topic配置
sync.topic.configs.enabled=true
# 是否同步源event-centerCL信息
sync.topic.acls.enabled=true
sync.group.offsets.enabled=true
# 连接器是否发送心跳
emit.heartbeats.enabled=true
# 心跳间隔
emit.heartbeats.interval.seconds=5
# 是否发送检查点
emit.checkpoints.enabled=true
# 是否刷新topic列表
refresh.topics.enabled=true
# 刷新间隔
refresh.topics.interval.seconds=60
# 是否刷新消费者组id
refresh.groups.enabled=true
# 刷新间隔
refresh.groups.interval.seconds=60
# DefaultReplicationPolicy / CustomReplicationPolicy
replication.policy.class=org.apache.kafka.connect.mirror.CustomReplicationPolicy
# 远端创建新topic的replication数量设置
replication.factor=3

注意：

replication.policy.class 默认为：DefaultReplicationPolicy，这个策略会把同步至目标集群的topic都加上一个源集群别名的前缀，比如源集群别名为A，topic为：bi-log，该topic同步到目标集群后会变成：A.bi-log，为啥这么做呢，就是为了避免双向同步的场景出现死循环。

官方也给出了解释：

这是 MirrorMaker 2.0 中的默认行为，以避免在复杂的镜像拓扑中重写数据。 需要在复制流设计和主题管理方面小心自定义此项，以避免数据丢失。 可以通过对“replication.policy.class”使用自定义复制策略类来完成此操作.

参考资料：

【如何保持topic名称一致】

【Kafka】MM2同步Kafka集群时如何自定义复制策略(ReplicationPolicy)_kafka mm2同步的目标topic名不一样-CSDN博客

配置文件示例2：

#定义集群别名
clusters = A, B
A.bootstrap.servers = A_host1:9092, A_host2:9092, A_host3:9092 # 设置A集群的kafka地址列表
B.bootstrap.servers = B_host1:9092, B_host2:9092, B_host3:9092 # 设置B集群的kafka地址列表
A->B.enabled = true # 开启A集群向B集群同步
A->B.topics = .* # 允许同步topic的正则B->A.enabled = true # 开启B集群向A集群同步
B->A.topics = .* # 允许同步topic的正则#MM2内部同步机制使用的topic，replication数量设置
checkpoints.topic.replication.factor=1
heartbeats.topic.replication.factor=1
offset-syncs.topic.replication.factor=1offset.storage.replication.factor=1
status.storage.replication.factor=1
config.storage.replication.factor=1#自定义参数
sync.topic.configs.enabled=true #是否同步源topic配置信息
sync.topic.acls.enabled=true #是否同步源ACL信息
emit.heartbeats.enabled=true #连接器是否发送心跳
emit.heartbeats.interval.seconds=5 #心跳间隔
emit.checkpoints.enabled=true #是否发送检查点
refresh.topics.enabled=true #是否刷新topic列表
refresh.topics.interval.seconds=5 #刷新间隔
refresh.groups.enabled=true #是否刷新消费者组id
refresh.groups.interval.seconds=5 #刷新间隔
readahead.queue.capacity=500 #连接器消费者预读队列大小
replication.policy.class=org.apache.kafka.connect.mirror.DefaultReplicationPolicy #使用LegacyReplicationPolicy模仿MM1
heartbeats.topic.retention.ms=1 day #首次创建心跳主题时，设置心跳数据保留时长
checkpoints.topic.retention.ms=1 day #首次创建检查点主题时，设置检查点数据保留时长
offset.syncs.topic.retention.ms=max long #首次创建偏移量主题时，设置偏移量数据保留时长
replication.factor=2 #远端创建新topic的replication数量设置

Standalone模式运行

需要两个配置文件，一个是作为worker的kafka集群信息（worker.properties），另一个是同步数据的配置（connector.properties）

worker.properties配置文件

bootstrap.servers=worker:29092
security.protocol=PLAINTEXT
sasl.mechanism=PLAINkey.converter = org.apache.kafka.connect.converters.ByteArrayConverter
value.converter = org.apache.kafka.connect.converters.ByteArrayConverteroffset.storage.file.filename=/tmp/connect.offsets
offset.flush.interval.ms=10000

connector.properties配置文件

name = MirrorSourceConnector
topics = projects.*
groups = *
connector.class = org.apache.kafka.connect.mirror.MirrorSourceConnector
tasks.max = 1# source
# 这个配置会使同步之后的Topic都加上一个前缀，慎重
source.cluster.alias = old
source.cluster.bootstrap.servers = source:9193
source.cluster.security.protocol=SASL_PLAINTEXT
source.cluster.sasl.mechanism=PLAIN
source.cluster.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="admin" password="admin-pwd";
# target
target.cluster.alias = new
target.cluster.bootstrap.servers = target:29092
target.cluster.security.protocol=SASL_PLAINTEXT
target.cluster.sasl.mechanism=PLAIN
target.cluster.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="Admin" password="hMOPbmZE";# 是否同步源topic配置信息
sync.topic.configs.enabled=true
# 是否同步源ACL信息
sync.topic.acls.enabled=true
sync.group.offsets.enabled=true
# 连接器是否发送心跳
emit.heartbeats.enabled=true
# 心跳间隔
emit.heartbeats.interval.seconds=5
# 是否发送检查点
emit.checkpoints.enabled=true
# 是否刷新topic列表
refresh.topics.enabled=true
# 刷新间隔
refresh.topics.interval.seconds=30
# 是否刷新消费者组id
refresh.groups.enabled=true
# 刷新间隔
refresh.groups.interval.seconds=30
# 连接器消费者预读队列大小
# readahead.queue.capacity=500
# 使用自定义策略
replication.policy.class=org.apache.kafka.connect.mirror.CustomReplicationPolicy
replication.factor = 3

 启动命令

./connect-standalone.sh worker.properties connector.properties

参考资料：

kafka 异步双活方案 mirror maker2 深度解析 - 知乎

Apache Kafka MirrorMaker 2.0 指南 - Azure HDInsight | Microsoft Learn

简述 Kafka Mirror Maker v2 – K's Blog

Getting up to Speed with MirrorMaker 2 - Confluent

9.4. 使用 MirrorMaker 2.0 在 Kafka 集群间同步数据 Red Hat AMQ 2021.q3 | Red Hat Customer Portal

https://blog.51cto.com/u_14049791/5703235

KafkaMirrorMaker2.0架构与实战全解《三》 - 墨天轮

2.4. Kafka MirrorMaker 2 集群配置 Red Hat AMQ Streams 2.4 | Red Hat Customer Portal

kafka:MirrorMaker-V1(MM1)到MirrorMaker-V2(MM2)-腾讯云开发者社区-腾讯云

【Kafka】记录一次基于connect-mirror-maker做的Kafka集群迁移完整过程_kafka数据同步到另外一个kafka-CSDN博客

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.rhkb.cn/news/334722.html

如若内容造成侵权/违法违规/事实不符，请联系长河编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

《当微服务遇上Ribbon：一场负载均衡的华丽舞会》

《当微服务遇上Ribbon：一场负载均衡的华丽舞会》

在微服务的厨房里，如何确保每一道服务都恰到好处？揭秘Spring Cloud Ribbon如何像大厨一样精心调配资源，让负载均衡变得像烹饪艺术一样简单！ 文章目录 Spring Cloud Ribbon 详解1. 引言微服务架构中的负载均衡需求Spring Cloud Rib…
阅读更多...
迅睿 CMS 中开启【ionCube 扩展】的方法

迅睿 CMS 中开启【ionCube 扩展】的方法

有时候我们想要某种功能时会到迅睿 CMS 插件市场中找现有的插件，但会有些担心插件是否适合自己的需求。于是迅睿 CMS 考虑到这一层推出了【申请试用】，可以让用户申请试用 30 天，不过试用是有条件的，条件如下： php 版…
阅读更多...
Python---Matplotlib（2万字总结）【从入门到掌握】

Python---Matplotlib（2万字总结）【从入门到掌握】

数据可视化 在完成了对数据的透视之后，可以将数据透视的结果通过可视化的方式呈现出来，简单的说，就是将数据变成漂亮的图表，因为人类对颜色和形状会更加敏感，然后再进一步解读数据背后隐藏的价值。在之前的文章中已经…
阅读更多...
链表mark

链表mark

什么是链表，链表是一种通过指针串联在一起的线性结构，每一个节点由两部分组成，一个是数据域一个是指针域（存放指向下一个节点的指针），最后一个节点的指针域指向null（空指针的意思）。…
阅读更多...
PCIe协议之-DLLP详解

PCIe协议之-DLLP详解

✨前言： 🌟数据链路层的功能 数据链路层将从物理层中获得报文， 并将其传递给事务层； 同时接收事务层的报文， 并将其转发到物理层; 核心的功能有以下三点 1.保证TLP在 PCIe 链路中的正确传递; 2.数据链路层使用了容错…
阅读更多...
「Python Socket超能力：网络世界的隐形斗篷！」

「Python Socket超能力：网络世界的隐形斗篷！」

Hi，我是阿佑，今天将带领大家揭开Python Socket编程的神秘面纱，赋予我们的网络应用隐形斗篷般的超能力！ 深入探讨Socket编程的革命性力量，教你如何用Python的Socket模块来构建强大的网络应用。从简单的HTTP服务器到复杂…
阅读更多...
你的手机是如何控制你的手表之广播篇

你的手机是如何控制你的手表之广播篇

前言 要让手机能够控制手表，第一步当然要让手机能够“看见”手表，人类作为上帝视角，我们是能够通过眼睛直接看见手机和手表的，但要让手机“看见”手表，就需要一端把自己的信息通过电磁波的形式发往空中，另…
阅读更多...
力扣225. 用队列实现栈

力扣225. 用队列实现栈

Problem: 225. 用队列实现栈 文章目录 题目描述：思路Code 题目描述： 思路 1.对一个queue模拟栈的操作，同时用一个int类型的变量topElem记录每次每次队列队尾的元素（也即是模拟stack中的stack的栈顶元素）； 2…
阅读更多...
dubbo复习：（10）使用tripple协议进行通信

dubbo复习：（10）使用tripple协议进行通信

一、pom.xml <?xml version"1.0" encoding"UTF-8"?> <project xmlns"http://maven.apache.org/POM/4.0.0"xmlns:xsi"http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation"http://maven.apache.org/POM/4.…
阅读更多...
es安装错误Exception in thread “main“ java.nio.file.NoSuchFileException解决方案

es安装错误Exception in thread “main“ java.nio.file.NoSuchFileException解决方案

docker 启动es出现一下错误的解决方案 Exception in thread “main” java.nio.file.NoSuchFileException: /usr/share/elasticsearch/config/jvm.options Exception in thread "main" java.nio.file.NoSuchFileException: /usr/share/elasticsearch/config/jvm.op…
阅读更多...
最简单的方式解决android studio 模拟器无法联网的问题

最简单的方式解决android studio 模拟器无法联网的问题

最简单的方式解决android studio 模拟器无法联网的问题 看了网上很多解决android studio内置模拟器无法联网的问题&#xff0c;基本上都是在模拟器手机上配置dns&#xff0c;个人试了多种办法也连不上网&#xff0c;现在给出一种&#xff0c;仅需要在命令行操作的解决安卓模拟…
阅读更多...
曲线拟合工具软件(免费)

曲线拟合工具软件(免费)

曲线拟合是数据处理中经常用到的数值方法，本质是使用某一个模型（方程或者方程组）将一系列离散的数据拟合成平滑的曲线或者曲面，数值求解出对应的函数参数，大家可以利用MATLAB的曲线拟合工具箱也可以使用第三方的拟合软件，今天我们介绍Welsim免费的曲线拟合软件 1、MATLA…
阅读更多...
牛！华为《Linux 面试笔记大全》太赞了，完整版PDF 开放下载！

牛！华为《Linux 面试笔记大全》太赞了，完整版PDF 开放下载！

在QQ和微信社群中&#xff0c;我注意到许多人都在寻找一份全面的Linux学习资料。因此&#xff0c;我在这里为大家整理和分类了相关的信息&#xff0c;可以看作是对重点内容的梳理和归纳。 这份《Linux面试笔记》主要分为三大部分&#xff1a;基础篇-进阶篇-高级篇 本书笔记针…
阅读更多...
生成式AI的GPU网络技术架构

生成式AI的GPU网络技术架构

生成式AI的GPU网络 引言&#xff1a;超大规模企业竞相部署拥有64K GPU的大型集群&#xff0c;以支撑各种生成式AI训练需求。尽管庞大Transformer模型与数据集需数千GPU&#xff0c;但实现GPU间任意非阻塞连接或显冗余。如何高效利用资源&#xff0c;成为业界关注焦点。 张量并…
阅读更多...
Linux系统之GoAccess实时Web日志分析工具的基本使用

Linux系统之GoAccess实时Web日志分析工具的基本使用

Linux系统之GoAccess实时Web日志分析工具的基本使用 一、GoAccess介绍1.1 GoAccess简介1.2 GoAccess功能1.3 Web日志格式 二、本地环境介绍2.1 本地环境规划2.2 本次实践介绍 三、检查本地环境3.1 检查本地操作系统版本3.2 检查系统内核版本3.3 检查系统镜像源3.4 更新软件列表…
阅读更多...
宝塔下应该用 Memcached 还是 Redis？

宝塔下应该用 Memcached 还是 Redis？

明月最近在跟几个使用宝塔面板的客户运维的时候发现不少站长不知道如何选择 Memcached 和 Redis&#xff0c;甚至都说不清楚 Memcached 或者 Redis 具体是用来干啥的&#xff1f;甚至还碰到过一个站长 Memcached 和 Redis 都安装了&#xff0c;但一个都没有用&#xff0c;就那么…
阅读更多...
遇到了导师放养，该怎么坚持？

遇到了导师放养，该怎么坚持？

最近收到学生读者的留言&#xff0c;抱怨科研的困难。导师忙碌且学生众多&#xff0c;自己只是众多学生之一&#xff0c;常常处于放养状态。除了每周的组会外&#xff0c;几乎无法接触到导师。在这种状态下&#xff0c;缺乏方向和动力&#xff0c;非常担心无法顺利毕业&#xf…
阅读更多...
进程互斥经典问题（读写者问题、理发店问题）

进程互斥经典问题（读写者问题、理发店问题）

目录 读写者问题 问题描述 问题分析 进程互斥问题三部曲 读者写者算法实现 一、找进程——确定进程关系 二、找主营业务 三、找同步约束 a.互斥 b.资源 c.配额 理发店问题 问题描述 问题分析 进程互斥问题三部曲 理发店问题算法实现 一、找进程——确定进程…
阅读更多...
光伏智慧化运营解决方案的应用和价值

光伏智慧化运营解决方案的应用和价值

在社会对新能源需求的不断扩大&#xff0c;光伏已经成为了可再生能源的重要组成部分&#xff0c;随着光伏电站数量和规模的不断扩大&#xff0c;相关企业和用户都就开始关注如何能够高效精准的进行电站管理&#xff0c;对此&#xff0c;鹧鸪云提出了光伏智慧化运营解决方案&…
阅读更多...
解读makefile中的.PHONY

解读makefile中的.PHONY

在 Makefile 中&#xff0c;.PHONY 是一个特殊的目标&#xff0c;用于声明伪目标&#xff08;phony target&#xff09;。伪目标是指并不代表实际构建结果的目标&#xff0c;而是用来触发特定动作或命令的标识。通常情况下&#xff0c;.PHONY 会被用来声明一组需要执行的动作&a…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023