《Java 消息中间件·可靠消息与消费治理》13.3 跨集群复制

为什么需要第二个集群

单集群能解决的问题，不要用两个集群解决。跨集群复制引入的复杂度是实打实的：异步延迟、重复、位移不再一一对应、切换时消费者要重新定位。它通常只为三件事存在：

1. 容灾：主集群整体不可用时，备集群能接管。
2. 迁移：从旧集群搬到新集群，不能停机。
3. 就近读取：跨地域部署，让读流量落在本地。

这三件事对复制的要求并不相同。容灾要求低 RPO（能接受丢多少），迁移要求可校验（搬完之后能对数），就近读取要求低延迟但可以接受重复。选型前先写清楚是哪一种，否则会做出一个三边都不满足的方案。

Kafka 官方的跨集群工具是 MirrorMaker 2（MM2）。它不是「把 broker 连起来」，而是一个基于 Kafka Connect 的复制应用：读源集群、写目标集群。理解这一点，后面所有行为都能推出来。

MM2 的四个组件

这里有两个反直觉的点，都是本机实测确认的：

第一，复制消费者没有消费组。 日志里它的 groupId=null，在源集群执行 kafka-consumer-groups.sh --list 看不到 MM2。它不是靠消费组协调的，而是靠 Connect 自己保存的位移。所以不能用消费组 lag 来监控复制延迟——那一栏根本不存在。要么比两边的 log-end offset，要么在目标端量消息的端到端时间。

第二，offset-syncs 主题在源集群，checkpoints 主题在目标集群。 这是按「谁读谁写」放的：源连接器在源集群读消息时就地记下映射；检查点连接器在目标集群写，供切换后的消费者查询。

复制后的主题名

MM2 默认的 DefaultReplicationPolicy 会在目标主题前加源集群别名：

源集群：library.loan.created
目标集群：src.library.loan.created

这个前缀不是装饰。它让「同名的两个主题」在目标集群里可以共存（比如双向复制时），也让消费者一眼看出数据来源。代价是所有消费端代码、监控、权限都要多一层前缀映射。如果不想改，可以换 IdentityReplicationPolicy，但双向复制时会互相覆盖，一般不推荐。

本机实测：两个单节点 + MM2

环境：apache/kafka:4.3.0 两个单节点 KRaft，容器名 mq-rel13-src / mq-rel13-dst，放在自定义网络 mq-rel13-net 上，advertised listener 用容器名。MM2 跑在第三个容器里，入口改成 connect-mirror-maker.sh，堆上限 512 MB。三个进程合计约 1.5 GB 堆，本机 4 vCPU / 7.7 GiB 能跑，但不要在这个规模上做压测。

MM2 配置：

clusters = src, dst
src.bootstrap.servers = mq-rel13-src:9092
dst.bootstrap.servers = mq-rel13-dst:9092

src->dst.enabled = true
src->dst.topics = library[.]loan[.]created
src->dst.groups = legacy-reader

replication.factor = 1
tasks.max = 2
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
refresh.topics.interval.seconds = 5
refresh.groups.interval.seconds = 5

第一个坑：内部主题默认 RF=3

前七个 replication.factor 都设成 1 了，第一次启动仍然失败，日志每 100 ms 刷一条：

' mm2-offsets.dst.internal' topic creation failed due to 'Unable to create topic(s) 'mm2-offsets.dst.internal': Unable to replicate the partition 3 time(s): The target replication factor of 3 cannot be reached because only 1 broker(s) are registered ...'

原因是 MM2 内嵌了一个 Kafka Connect worker，它自己要建三类内部主题：mm2-offsets.<alias>.internal、mm2-status.<alias>.internal、mm2-configs.<alias>.internal。这三类主题的复制因子由 Connect 的 offset.storage.replication.factor、status.storage.replication.factor、config.storage.replication.factor 控制，默认 3，和 replication.factor 是两套配置。单节点实验必须把这三个也降到 1。

这也解释了为什么单节点上跑 MM2 的教程经常「配置看起来都对，就是起不来」：报错里的主题名是 mm2-offsets，不是业务主题，容易以为是 offset-syncs.topic.replication.factor 没生效。

第二个坑：正则里的点

src->dst.topics 是正则，不是通配符。library.loan.created 里的 . 会匹配任意字符；写成 library[.]loan[.]created 才能精确匹配。另一个写法是 library\\.loan\\.created，因为 properties 文件会先吃掉一层反斜杠。两种都行，前者更不容易看错。

复制延迟：只能测，不能猜

先把源集群的主题建成 3 分区，生产 820 条消息。MM2 起来后，目标集群的 src.library.loan.created 分区位移与源端完全一致：

源集群：
library.loan.created:0:520
library.loan.created:1:0
library.loan.created:2:300

目标集群：
src.library.loan.created:0:520
src.library.loan.created:1:0
src.library.loan.created:2:300

逐条端到端延迟

在目标端起一个消费者，从最新位置开始读，每读到一条就用当前时间减去记录的 CreateTime。源端以 10 ms 间隔连续生产 300 条，测量结果：

records: 300
all:   min=70 p50=701 p95=1099 max=1111 mean=689.0
last100: min=70 p50=359 p95=607 max=648 mean=355.2
first5: [1044, 1036, 1037, 1038, 1039]
last5:  [113, 103, 92, 81, 70]

这组数字不能直接读成「复制延迟 700 ms」。 控制台消费者是批量打印的：一次 poll() 拿到一批记录后连续输出，时间戳打在打印时刻，所以同一批里越早生产的记录看起来越慢。真正接近复制延迟的是每批的尾部，也就是最后几条的 70–113 ms。

要更准，得在消费端按记录到达时刻打点（例如用 pause/resume 或自己写 Consumer 在 poll() 返回时逐条记时间），而不是用控制台消费者。本节没有做到这一层，所以只给区间，不给精确值。

突发收敛

再发 500 条突发消息，然后在源容器里同时查两边总位移，直到相等：

src_total=820
converged dst_total=820 after_ms=1557 polls=1

polls=1 说明第一次探测时已经追平；1557 ms 里绝大部分是 docker exec 与 JVM 启动的开销。因此只能说「500 条突发在约 1.5 秒内追平」，不能给出更细的收敛曲线。 这个精度在本机（同机、同网络、单分区）已经到测量工具的下限。

延迟由什么决定

MM2 的复制延迟不是固定值，它由四段叠加：

1. 源端 poll() 的等待：consumer.poll.timeout.ms 默认 1000 ms，没数据时不会空转。
2. 攒批：目标端生产者的 linger.ms 与 batch.size。
3. 网络与目标端写入。
4. Connect 的位移提交频率（影响故障恢复后的重放量，不影响稳态延迟）。

所以「把 MM2 的延迟调到 50 ms」这种需求，本质是要求改小 poll 超时和 linger，代价是吞吐下降、请求数上升。跨集群复制永远是吞吐与延迟的交换，不存在只赚不赔的参数。

位移映射：最容易被想当然的地方

切换集群时，消费者不能拿源集群的位移直接去目标集群提交。原因是目标集群的位移可能整体偏移：目标主题不是空的时候、MM2 从中途开始复制的时候、或者主题被重建过的时候，source offset 100 对应的可能是 target offset 37。

MM2 用两层记录解决这件事：

• offset-syncs：源连接器在复制过程中写 (topic, partition, upstreamOffset, downstreamOffset)。
• checkpoints：检查点连接器把消费组已提交的源位移翻译成目标位移，写进 src.checkpoints.internal。

本机把这两层都解出来了。mm2-offset-syncs.dst.internal 在源集群，用 MM2 自带的 OffsetSync.deserializeRecord 解码：

[offset-sync] library.loan.created partition=0 upstream=0 downstream=0
[offset-sync] library.loan.created partition=0 upstream=19 downstream=19
[offset-sync] library.loan.created partition=0 upstream=101 downstream=101
[offset-sync] library.loan.created partition=0 upstream=202 downstream=202
[offset-sync] library.loan.created partition=0 upstream=303 downstream=303
[offset-sync] library.loan.created partition=0 upstream=404 downstream=404
[offset-sync] library.loan.created partition=0 upstream=505 downstream=505
[offset-sync] library.loan.created partition=0 upstream=519 downstream=519

因为目标主题在复制开始时是空的，映射是 1:1。这是特例，不是规律：只要目标端提前有数据，或者复制从中间开始，映射立刻错位。

一个对不上的 checkpoint

用 legacy-reader 组在源集群消费并提交到 offset 50，等检查点连接器写入后，用 Checkpoint.deserializeRecord 解码目标集群的 src.checkpoints.internal：

[checkpoint] group=legacy-reader topic=src.library.loan.created partition=0 upstream=50 downstream=1 metadata=
[checkpoint] group=legacy-reader topic=src.library.loan.created partition=2 upstream=0 downstream=0

upstream=50 翻译成了 downstream=1，而 offset-syncs 明明写着 1:1。 这组数字与上一节的映射不一致。我们还用 MM2 自带的 RemoteClusterUtils.translateOffsets 查了一次，返回 0 条记录：

[raw] translated_entries=0 keys=[]

也就是说，在本次实验里，checkpoint 既不可信，翻译 API 也没有返回结果。具体成因（检查点写入时机早于 offset-syncs 加载完、还是 4.3.0 的某个行为差异）没有在本机定位，这里不编解释。

但结论是确定的，而且比任何「正确示例」都有用：

1. 不要假设源位移等于目标位移。 1:1 只是空目标集群的特例。
2. 不要盲信某一条 checkpoint。 它是写入时刻的快照，消费组不再提交就不会重写。
3. 切换前必须自己校验。 拿几条已知消息，在目标端按翻译后的位移读出来，比对内容，比读任何文档都可靠。
4. 翻译 API 要演练。 translateOffsets 在本次实验里返回空，如果切换脚本直接依赖它，就会「以为没位移可切」而做出错误决策。

checkpoint 不是 JSON

网上很多文章说 .checkpoints.internal 里是 JSON，可以直接用控制台消费者读。4.3.0 不是：Checkpoint 用 org.apache.kafka.common.protocol.types.Schema 做二进制编解码，kafka-console-consumer 打出来是一串控制字符，od -c 能看到 \0 与长度前缀。

要读它，只能用 MM2 自带的解码入口：

Checkpoint cp = Checkpoint.deserializeRecord(record);
System.out.printf("group=%s topic=%s partition=%d upstream=%d downstream=%d%n",
        cp.consumerGroupId(), cp.topicPartition().topic(),
        cp.topicPartition().partition(), cp.upstreamOffset(), cp.downstreamOffset());

这段代码用的是 connect-mirror-client，属于 Kafka 发行包自带的类，不引入第三方客户端。切到目标集群消费时，RemoteClusterUtils.translateOffsets 也是同一个包的入口。

切换时到底要做什么

主备切换不是「把 DNS 改一下」。至少要处理四件事：

一、生产者。 生产者的 bootstrap.servers 指向新集群后，acks=all、幂等、事务的语义都要重新建立。事务生产者的状态不会随复制迁移：切到目标集群后必须重新 initTransactions，同名 transactional.id 在目标集群是全新状态，别指望延续源集群的 epoch，也别指望源集群里未完成的事务会在目标端出现。

二、消费者位移。 优先用 checkpoint 翻译后的位移，但必须按上一节的方法校验。校验不通过时，退回到「按业务时间/唯一键重放」，不要硬提交一个猜出来的位移。

三、重复。 MM2 是至少一次：源端已复制、位移未提交时崩溃，重启会重发。切换窗口内如果生产者还在往源集群写，这部分数据要么丢（不复制），要么在目标端重复（切回后补复制）。跨集群复制不提供恰好一次，业务幂等（第 2 章）仍然是必需的。

四、RPO 与 RTO。 异步复制的 RPO 等于切换瞬间还没复制过去的数据量，通常用「复制延迟 × 写入速率」估算；RTO 取决于切流量、校验位移、预热消费者的时间。这两个数字必须演练测出来，不能拍。

双活与吞吐：两个不要做的事

不要用 MM2 做「同步复制」。 它读一批、写一批，源端的 acks=all 只保证写到源集群，不保证写到目标集群。源集群确认成功的那一刻，目标集群可能还差几百毫秒的数据。任何要求「两个集群同时有」的业务，都得在应用层做双写或对账，不能指望 MM2。

不要让 MM2 和 broker 抢同一批 CPU。 本机把两个 broker 加一个 MM2 塞进 4 vCPU，已经能跑但毫无余量；生产上 MM2 应该是一套独立的 Connect 集群，有自己的堆、自己的磁盘（Connect 的位移主题）和自己的监控。它的故障不应该影响 broker。

吞吐上，tasks.max 决定并行度：每个 task 负责一部分分区，task 太少会成为瓶颈，task 太多则增加目标端的请求压力。真正先到瓶颈的通常是跨机房带宽和目标集群的写入能力，不是 MM2 的 CPU。跨机房链路上可以给复制生产者开压缩：

src->dst.producer.compression.type = zstd
src->dst.producer.linger.ms = 50

这两条是 MM2 的 per-connector 覆盖语法：<source>-><target>.producer.<客户端配置>。压缩省带宽但费 CPU，linger 提高吞吐但增加延迟，和 11.2 的取舍完全一样，只是这次约束从单机变成了链路。压缩与 linger 的实际收益未在本机执行（本实验是两个容器走同一条 bridge 网络，没有跨机房链路可省），配置语法以 MM2 的客户端覆盖规则为准。

一套可执行的切换检查表

1. 停写源集群（或确认双写窗口），记录停写时刻的源端 log-end offset。
2. 等复制追平：目标端 log-end 不小于停写时的源端值；追不平就不要切。
3. 翻译并校验位移：用 checkpoint 翻译，抽 3–5 条按目标位移读出来比对内容。
4. 预启动消费者：先用 assign 指定位移试读，不提交；确认无误再切正式消费组。
5. 切生产者：改 bootstrap，重建事务生产者，观察错误率与重试。
6. 保留回切路径：源集群只读保留一段时间，确认无回流需求再下线。
7. 记录 RPO/RTO 实测值：写进演练报告，下一次切换以此为基线。

本节判据

1. MM2 的复制消费者有消费组吗？ 答：没有，groupId=null；不能用消费组 lag 监控复制延迟。
2. 为什么单节点跑 MM2 会卡在 mm2-offsets.*.internal？ 答：MM2 内嵌 Connect worker，它的 offset/status/config 三类内部主题默认 RF=3，要单独配 *.storage.replication.factor=1。
3. 源位移能直接当目标位移用吗？ 答：不能。空目标集群才恰好 1:1；有历史数据或中途开始时必然错位，必须用 offset-syncs/checkpoint 翻译并校验。
4. 本次实验里 checkpoint 说明了什么？ 答：它写的是 50→1，与 offset-syncs 的 1:1 不一致，翻译 API 还返回空；说明 checkpoint 是快照、不可盲信，切换前必须自己验证。
5. MM2 提供恰好一次吗？ 答：不提供，是至少一次；崩溃重发和切换窗口都会产生重复，业务幂等不可省。

阅读导航：上一节 13.2 SASL/ACL 与传输加密 ｜ 下一节 14.1 故障注入：broker 宕机