Kafka跨集群同步方案全解析：从MirrorMaker 2.0到自研管道的选型指南

最近大半年，我陆续被拉进好几个项目，需求听起来都差不多：把Kafka里的数据同步到另一套Kafka。有的要跨机房灾备，源集群挂了备份集群能立刻顶上；有的是老集群退役，几千个topic要原样搬去新集群；还有的是好几套业务集群往同一个分析集群汇聚。一开始我以为统一用一个工具就能搞定，实际做下来才发现，同样叫Kafka到Kafka同步，场景不同，方案完全是另一回事。MirrorMaker 2.0、Kafka Connect自研连接器、自己写Consumer-Producer管道、商业复制工具，这些我都真刀真枪跑过，踩过不少坑。这篇就按我实际使用的顺序，把几种主流方案的适配场景、配置方法和坑位一次讲清楚，给准备做跨集群同步的伙伴当个参考。

1. 同步需求背后的真实场景：先把问题定义清楚

1.1 四种最常见的Kafka到Kafka同步场景

第一个场景是跨机房容灾。主集群部署在A机房，备份集群在B机房，平时业务只读写A集群，B集群默默接收同步数据。一旦A机房出故障，流量整体切换到B集群。这个场景的难点不在"消息能不能过去"，而在"消费进度能不能接得上"——如果B集群只有原始消息，消费者的offset却从零开始，切换后要么大量重复消费，要么积压到爆。很多团队第一次做灾备演练时，都会在切换这一步翻车。

第二个场景是集群搬迁或升级。老集群版本太旧、机器要退租，或者公司统一换一套新的Kafka环境，需要把全部topic连同数据一起搬过去，同时在搬迁窗口内尽量不停机。这种任务对方案的完整性要求最高：光搬消息不够，topic的副本因子、保留时间、清理策略、ACL、消费者组的offset，最好都能一起迁移，否则新集群就是个"裸壳"，业务切过去还是要重新配置一遍。

第三个场景是多集群数据汇聚。按团队或地域拆成几套Kafka之后，实时数仓和分析平台往往需要一份全量数据。这种同步通常是单向的，而且目标端经常要做"整形"：把某个topic改名、在消息里打上来源集群的标签、过滤掉内部测试数据。此时工具是否具备可编程的转换能力，比同步吞吐量更重要。

第四个场景是环境复制。把生产数据同步到测试或预发环境，供联调、压测、算法训练使用，一般还要顺带做脱敏。这类任务多半是阶段性的，同步链路跑一段时间就撤掉，不值得为它搭建一套常驻基础设施。

我把这四个场景整理成一张表，选型的时候可以直接对照：

1.2 动手之前必须拿到的五个约束答案

选型之前先回答这五个问题，很多纠结在源头就能解决。

第一，同步方向是单向还是双向。双向（active-active）意味着两个集群都接受业务写入，复制逻辑要能防止消息在集群之间来回传成死循环，还要处理同一key在两端都被更新的冲突。MirrorMaker 2.0支持双向复制，但默认的topic改名策略要理解清楚，否则会踩坑。

第二，延迟要求到底是多高。灾备场景通常能容忍分钟级延迟，实时分析链路则可能要求秒级。延迟要求直接影响一个关键参数：消费者组offset的同步间隔。间隔越短，切换时丢掉的进度越少，但内部topic的压力也越大。

第三，对重复和乱序的容忍度。绝大多数Kafka同步方案都不能保证完全精确一次，最多做到"至少一次"。如果下游消费逻辑没有幂等，设计同步方案时会非常痛苦，反过来，如果下游能按业务主键去重，方案选择空间就大很多。

第四，需要同步哪些附加对象。只同步消息最简单，同步topic配置和ACL稍微复杂一点，再把消费者组的offset一起同步，复杂度会直接上一个台阶。很多自研方案就是栽在最后这一点上。

第五，数据规模和网络拓扑。topic数量是几十个还是上万个，单条消息多大，跨机房带宽多少，RTT多高——这些决定了连接器开多少个task、要不要启用跨集群压缩、同步窗口怎么定。建议把量级先估出来，再谈方案，不要拿着两个topic的测试结果去设计万级topic的线上链路。

一句话记这个阶段的核心：数据复制只是手段，业务在切换那一刻能不能"接得住"才是目的。所有方案选择，最后都要回到这个目的上来。

2. MirrorMaker 2.0：官方复制工具的能力边界

2.1 MM2的原理：连接器、复制流与心跳

MirrorMaker 2.0是Kafka官方主推的跨集群复制方案，从Kafka 2.4开始引入，设计目标就是替代老的MirrorMaker 1.x。它不是一个独立运行的服务，而是建立在Kafka Connect框架之上的一组连接器，包括MirrorSourceConnector、MirrorCheckpointConnector、MirrorHeartbeatConnector这几个核心组件。你写一份配置文件，用自带的脚本启动，它会自动在参与同步的集群上创建这些连接器。

配置里用clusters定义参与同步的集群别名，用A->B.enabled = true定义复制方向。启动之后，源集群的每条新消息会被读取并写入目标集群，这个过程本质上还是"一个消费者加一个生产者"，但框架替你解决了offset存储、任务分发、重试和健康检查这些脏活。

MM2有一个非常关键的设计：topic重命名。默认情况下，从集群A同步到集群B的topic，到了B之后名字会变成A.orders（前缀就是源集群别名，分隔符默认是.）。这样做有两个目的：一是在双向复制时防止同一条消息被两个集群反复传抄形成死循环；二是目标集群上"本地topic"和"远程topic"可以共存而不互相干扰。如果你不需要这个前缀，可以配置replication.policy.class=org.apache.kafka.connect.mirror.IdentityReplicationPolicy，让目标集群保留原始topic名。

心跳和检查点机制是MM2区别于其它方案的核心。源集群会周期性写心跳消息，目标集群的对应连接器消费心跳，用来检测链路是否健康。消费者组的进度则靠MirrorCheckpointConnector周期性地从源集群采集，记录到目标集群的内部checkpoint topic里。所以MM2不只是搬消息，它提供了"让下游消费者在目标集群续跑"的可能性——这一点对灾备场景至关重要。

2.2 一份可以直接跑的MM2配置

先给一份最简配置，双向同步两个集群的全部topic：

# mm2.properties # 定义参与同步的集群别名 clusters = A, B # 两个集群的broker地址 A.bootstrap.servers = 10.0.1.11:9092,10.0.1.12:9092,10.0.1.13:9092 B.bootstrap.servers = 10.0.2.11:9092,10.0.2.12:9092,10.0.2.13:9092 # 开启两个方向的同步 A->B.enabled = true B->A.enabled = true # 同步哪些topic，支持逗号分隔或正则表达式，.* 表示全部 A->B.topics = .* B->A.topics = .* # 目标集群上新建topic的副本因子 replication.factor = 3 # 内部topic的副本数可以设低一些 checkpoints.topic.replication.factor = 1 heartbeats.topic.replication.factor = 1 offset-syncs.topic.replication.factor = 1 # 同步topic配置、消费者组offset、ACL sync.topic.configs.enabled = true sync.group.offsets.enabled = true sync.topic.acls.enabled = true # 消费者组进度同步间隔，默认60秒 sync.group.offsets.interval.ms = 60000 # topic列表刷新间隔，默认600秒 refresh.topics.interval.seconds = 600

启动方式很简单，在Kafka安装目录下直接跑：

bin/connect-mirror-maker.sh mm2.properties

生产环境我更建议用分布式模式而非这个单机脚本，也就是把MM2的连接器提交到一个connect-distributed集群上运行，这样能拿到高可用和任务自动均衡。启动后先做两个验证：第一，在目标集群执行kafka-topics.sh --list，应该能看到带源集群别名的topic，比如A.orders；第二，用kafka-console-consumer.sh消费A.orders，确认消息内容跟源端一致。

2.3 我在MM2上踩过的几个坑

先讲最大的坑：topic改名导致下游断裂。某次做多集群汇聚，目标集群的实时计算任务本来消费orders，MM2同步过来的数据却写在A.orders里，任务静默无数据，排查了半天才发现是topic名对不上。单向同步场景下，我通常直接用IdentityReplicationPolicy去掉前缀；如果必须保留前缀，就要求下游消费端配合改订阅的topic名，并且把变更提前通知到位。

第二个坑是checkpoint的offset不是实时的。默认60秒才同步一次消费者组进度，也就是说源集群消费组的最新进度，最坏情况下会比checkpoint记录的领先60秒。灾备切换后，目标端消费者会从checkpoint位置继续消费，最近一分钟的消息会被重新读一遍。如果业务无法容忍这批重复，就把sync.group.offsets.interval.ms调小到5秒或10秒，代价是内部checkpoint topic的写入量变大。这个权衡一定要在演练阶段就跟业务方讲明白，别等真正切换时才暴露。

第三个坑是topic数量很大的时候，新增topic的发现不够及时。默认topic刷新间隔600秒，意味着新创建的topic最坏要10分钟后才会开始同步。我习惯把refresh.topics.interval.seconds固定到120到300秒，同时把新增topic纳入变更流程，而不是依赖自动发现。反过来也别设太短，否则源集群的metadata请求会被刷爆。

第四个坑是跨版本同步时配置同步不完全可靠。两端的Kafka版本差距过大时，topic配置和ACL的同步可能失败或者字段丢失。做集群迁移时尽量让两端版本保持一致，如果做不到，就只开数据复制，配置和ACL单独用脚本迁移。

还要提醒一点：MM2虽然叫"镜像"，但它不做消息去重，也不做冲突合并。双向复制时如果两个集群都写了同一个key，目标端看到的就是两个版本，后到的覆盖先到的，业务需要自己保证同一时间只有一个集群接受某个key的写请求。

提示：所有关于MM2的结论，都建议在你自己的环境里用演练验证一遍。我见过不止一次"配置看着没问题，一切换就出事"的情况，灾备链路最怕的就是没演练过。

3. Kafka Connect自定义Connector：精细化复制必选之路

3.1 Connect复制链路的核心机制

当同步需求不只是"原样搬消息"，而要附带脱敏、过滤、改名、按内容分流时，就该考虑自己写一个Source Connector。Kafka Connect本身是一个数据集成框架，它负责管理worker、任务的分布、offset的持久化，以及提供REST API和监控指标。你只需要实现两个核心类：一个SourceConnector负责定义配置和task分配，一个SourceTask负责真正干活。

自定义Source Connector的工作模式是这样的：框架周期性地调用你的poll()方法，你返回一批SourceRecord；框架收到这批记录后，负责把它们写入目标topic，并周期性提交offset。每个SourceRecord里最重要的两个字段是sourcePartition和sourceOffset——它们共同构成这条记录的"源位置标识"。框架会把你返回的最后一条记录的sourceOffset存入内部offset topic，下次重启时通过context.offsetStorageReader()把上次的位置交还给你。

理解这个机制之后，就明白自定义Connector和MM2的区别在哪：框架只帮你在目标端投递消息、保存"源读取位置"，但不会帮你同步消费者组的offset，不会生成心跳，也不会同步topic配置和ACL。你自己的任务是从源topic读取记录并组装成SourceRecord，仅此而已。

大多数场景下我强烈建议做"字节透传"，也就是以byte[]作为key和value直接复制，不做反序列化。一旦你把记录解析成JSON对象再序列化回去，原始消息里的字段顺序、时间戳格式、压缩方式都可能发生变化，下游解析反而出问题。只有需要真正读取字段做判断时，才做局部解析。

3.2 自定义Source Connector的最小实现

这里给出一个可运行的最小骨架。Connector端负责读配置、分任务：

public class KafkaCopyConnector extends SourceConnector { private static final ConfigDef CONFIG_DEF = new ConfigDef() .define("src.topic", Type.STRING, Importance.HIGH, "源topic") .define("dst.topic", Type.STRING, Importance.HIGH, "目标topic") .define("batch.size", Type.INT, 500, Importance.MEDIUM, "poll批次大小"); private Map<String, String> settings; @Override public void start(Map<String, String> props) { this.settings = props; } @Override public Class<? extends Task> taskClass() { return KafkaCopyTask.class; } @Override public List<Map<String, String>> taskConfigs(int maxTasks) { // 简化处理：每个task拿到相同配置 return Collections.nCopies(maxTasks, settings); } @Override public ConfigDef config() { return CONFIG_DEF; } @Override public void stop() {} @Override public String version() { return "1.0.0"; } }

Task端是最容易写错的地方。核心逻辑是：先订阅源topic，然后用context.offsetStorageReader()恢复上次的位置，再开始poll循环：

public class KafkaCopyTask extends SourceTask { private KafkaConsumer<byte[], byte[]> consumer; private String dstTopic; @Override public void start(Map<String, String> props) { Properties consumerProps = new Properties(); consumerProps.put("bootstrap.servers", props.get("src.bootstrap.servers")); consumerProps.put("group.id", "connect-copy-" + props.get("src.topic")); consumerProps.put("enable.auto.commit", "false"); consumerProps.put("key.deserializer", ByteArrayDeserializer.class.getName()); consumerProps.put("value.deserializer", ByteArrayDeserializer.class.getName()); consumer = new KafkaConsumer<>(consumerProps); dstTopic = props.get("dst.topic"); // 关键：用框架保存的offset恢复读取位置 List<PartitionInfo> infos = consumer.partitionsFor(props.get("src.topic")); for (PartitionInfo info : infos) { TopicPartition tp = new TopicPartition(info.topic(), info.partition()); Map<String, ?> sourcePartition = Collections.singletonMap("topic", info.topic()); Map<String, ?> offset = context.offsetStorageReader().offset(sourcePartition); if (offset == null) { consumer.seekToBeginning(Collections.singleton(tp)); } else { consumer.seek(tp, (Long) offset.get("offset") + 1); } } // 生产环境建议用assign手动分配分区，避免多个task抢同一批topic } @Override public List<SourceRecord> poll() throws InterruptedException { ConsumerRecords<byte[], byte[]> records = consumer.poll(Duration.ofMillis(200)); List<SourceRecord> result = new ArrayList<>(); for (ConsumerRecord<byte[], byte[]> record : records) { // 这里可以按需做过滤、改名、打标、脱敏 result.add(new SourceRecord( Collections.singletonMap("topic", record.topic()), Collections.singletonMap("offset", record.offset()), dstTopic, null, record.key(), null, record.value())); } return result; } @Override public void stop() { if (consumer != null) { consumer.close(); } } }

这个代码里最关键的是start()里的offset恢复逻辑。如果你图省事，用consumer自己的group offset来恢复，一旦Connect框架和consumer的offset管理不一致，重启后就会出现重复消费或者跳过消息。我在项目里见过不止一次这样的BUG，症状就是"重启之后数据错位"。

3.3 与MM2相比的优点和代价

自定义Connector的优点是控制力强。你可以在poll循环里做任意处理：不想要的记录直接不放进result里就实现了过滤；需要分流时按内容修改SourceRecord的target topic；需要打标签时在SourceRecord的headers里加字段。数据整形需求越复杂，这条路越值。

代价也很明确。第一，MM2开箱即有的消费者组offset同步、心跳检测、ACL同步，自定义Connector全都没有，需要自己实现或干脆放弃。第二，Connect框架的offset提交是周期性的（默认offset.flush.interval.ms），极端情况下会有少量重复，下游要做好幂等。第三，自己写的代码要自己负责重试、异常处理和监控告警，出错时没有厂商兜底。

所以我的判断是：如果需求里带了"改名、脱敏、过滤"这类整形动作，Connect自定义Connector是首选；但如果核心诉求是"灾备切换时消费者能续跑"，那还是把MM2或商业方案放在前面，因为它们把最难的那部分offset衔接做掉了。

4. 自研Consumer-Producer管道：什么时候才需要自己造轮子

4.1 自研管道的典型动机

说实话，自研管道最大的优势不是技术，而是省事。有些同步任务就是一次性或者短期的：把一个topic的历史数据搬到新集群，跑几天就结束；把生产环境数据同步到测试集群做一轮压测，压完就停。为这种事去部署一套Kafka Connect集群，维护成本明显不划算。

另一个动机是业务逻辑太重。比如要根据消息内容做路由，或者把多条消息做关联拼接，用Connect的SMT写起来十分别扭，直接写一个consumer程序反而顺手。还有的团队对Consumer/Producer API很熟悉，但没怎么碰过Connect，上手自研管道比学新框架更快。

需要说清楚的是：自研方案通常只解决"消息搬运"这一件事。什么topic配置同步、ACL同步、消费者组offset衔接，统统要自己写，而且大概率写不完整。所以它适合低复杂度、短周期、能容忍重复的场景；跨机房灾备这种严肃场合，我不建议走这条路。

4.2 一个可落地的管道骨架

下面这个Python管道是我用得比较顺手的骨架，逻辑很直白：消费源topic，原样发送到目标topic，等这一批全部发送成功后再提交offset。

from kafka import KafkaConsumer, KafkaProducer SRC_BOOTSTRAP = "10.0.1.11:9092,10.0.1.12:9092,10.0.1.13:9092" DST_BOOTSTRAP = "10.0.2.11:9092,10.0.2.12:9092,10.0.2.13:9092" SRC_TOPIC = "orders" DST_TOPIC = "orders_dst" consumer = KafkaConsumer( SRC_TOPIC, bootstrap_servers=SRC_BOOTSTRAP, group_id="sync-orders-001", enable_auto_commit=False, auto_offset_reset="earliest", ) producer = KafkaProducer( bootstrap_servers=DST_BOOTSTRAP, acks="all", enable_idempotence=True, ) while True: batch = consumer.poll(timeout_ms=1000) if not batch: continue for tp, records in batch.items(): futures = [] for record in records: future = producer.send( DST_TOPIC, key=record.key, value=record.value, headers=[ ("src_topic", tp.topic.encode()), ("src_partition", str(tp.partition).encode()), ("src_offset", str(record.offset).encode()), ], ) futures.append(future) # 关键：这一批全部发送成功后才提交offset for future in futures: future.get(timeout=30) consumer.commit()

这个骨架有几个刻意设计的点。第一，enable_auto_commit=False手动管理offset，避免"生产还没成功就提交消费进度"。第二，先收集这一批所有send的future，逐个get()成功之后再commit()；中途只要有一个失败，程序就抛异常退出，重启后会从最近一次提交的位置重新消费，源端会产生重复，但绝不丢数据。第三，在header里记录源topic、partition、offset，下游做排查和幂等都有依据。

还有一个细节：key保持原样传递，这样目标端按key做分区哈希时，大体上能延续源端的分区关系。如果源端和目标端的分区数不一致，这个映射就不保证一一对应了，下游不要对分区关系做任何假设。

4.3 自研管道最容易栽的三个跟头

第一个跟头是只send不等待就commit。send()是异步的，返回的是future而不是发送结果；如果代码写成"提交所有send之后就commit"，一旦producer在后台发送失败，这批消息就永久丢了——因为消费进度已经提交，重启后不会再读这些消息。这是自研管道里最致命、也最容易犯的错误。我的习惯是宁可慢一点，也必须先future.get(timeout=30)再commit。

第二个跟头是以为自研管道能"在目标端续跑消费进度"。事实上你在commit的是源集群消费者组的offset，跟目标集群消费者的进度完全没有关系。如果想让目标端消费者从某个位置继续，需要自己写offset映射逻辑，这个工作量往往被严重低估。

第三个跟头是对消息做全量反序列化。为了处理业务字段而把每条消息都解析成JSON，一旦遇到脏数据，整个管道就可能挂掉。我的做法是只对需要判断的字段做解析，其余内容一律按字节透传；解析失败的记录单独丢到dead letter topic，而不是在管道里原地抛异常。

注意：自研管道跑起来容易，但维护是长期的。建议至少把消费lag、生产失败数、目标topic写入速率这几个指标接到监控上，别等出问题再回头看日志。

5. 商业复制工具与云托管方案：花钱能买到什么

5.1 商业Replicator到底在解决什么

部分商业Kafka发行版会附带一个名为Replicator的复制组件，它底层同样基于Kafka Connect，但做了大量生产级加固。跟开源MM2相比，它能提供的增量价值主要在三块：更低的端到端延迟、更完整的offset转换机制、以及更省心的运维支持。

延迟这块，商业方案通常针对复制链路做过专门的吞吐优化，在跨机房高延迟网络下表现更稳定。offset转换这块则是它的核心卖点：不仅同步消费者组的进度，还能在目标端做位置映射，让消费者切换时几乎感受不到进度跳动。运维上，它一般自带图形化的复制链路监控、故障告警和原厂支持，这对于团队人手紧张的场景确实有吸引力。

但要注意，商业Replicator不是开箱即用的独立产品，它依赖商业Kafka发行版的整体环境，许可证成本不低。如果公司本身没在使用对应的商业发行版，为了一个同步任务引入它，ROI往往不高。

5.2 云环境提供的集群链接功能

使用云上托管Kafka服务时，常见的做法是利用平台自带的集群链接功能。它的好处是彻底免运维：不需要自己部署Connect集群，不用写连接器代码，在控制台上建立链接、选择要同步的topic、配置同步方向即可。计费一般按同步的数据量来算，链路闲置时的成本很低。

这类功能也支持双向同步，并且平台会处理重复消息的标识和部分冲突策略。对于已经把业务迁到云上、或者正在做混合云建设的团队来说，这确实是最省事的选项。

它的短板是绑定平台。链接配置、监控、告警都在云厂商的控制台里，数据链路跟平台的生态强耦合。将来如果要把业务迁回自建环境，这套同步链路基本要推翻重来。另外，云上集群链接对跨云网络质量要求较高，网络抖动会直接反映在同步延迟上。

5.3 买商业方案前要算清楚的三本账

第一本账是许可证费用对自运维人力成本。一套MM2出问题，排查和优化的工时可能相当可观；但如果只有一两条同步链路，开源方案完全够用，买商业方案的钱不一定划算。

第二本账是功能匹配度。如果只是单向同步几个topic，商业方案的offset转换、图形化监控、SLA保障这些卖点你根本用不上。反过来，如果业务核心链路依赖跨机房容灾，切换时的数据一致性直接关系营收，那这笔钱可能花得值得。

第三本账是锁定风险。托管方案把数据链路绑在平台里，未来要迁出会非常痛苦。签约前先确认周边生态是否支持你想要的监控、告警、审计方式，别等链路跑起来才发现告警接不进公司现有的监控体系。

我个人的态度是：能用开源方案解决的问题，优先开源；商业方案的价值不在于"功能多"，而在于"有人兜底"。团队小、核心链路重要、预算允许的时候，花钱买省心是合理的；否则还是先把开源方案用熟练更实在。

6. 选型对照与实测经验

6.1 一张表看懂方案差异

把前面几种方案放到同一张表里，差异一目了然：

这里说的"延迟参考"是我在普通内网环境下的量级感受，不是benchmark。真正选型时还是要用你自己的数据量和网络环境跑一轮压测，重点观察两点：目标端消息可见延迟，以及消费者组进度同步的最坏延迟。

6.2 实测下来的延迟与稳定性参考

我用三节点源集群、三节点目标集群，topic设12个分区，单条消息约2KB，消息速率约每秒2万条，在万兆内网环境下做过一轮对比。MM2数据可见延迟大概在1到2秒，消费者组进度同步跟随sync.group.offsets.interval.ms，调到5秒后基本稳定。Connect自定义Connector在默认配置下端到端延迟约几百毫秒，但poll批次调大时延迟会出现明显的波动，属于正常现象。自研管道在"发送全部成功再提交"的保守模式下，单进程大概能跑到每秒1到2万条，瓶颈主要在producer的批量参数和max.in.flight.requests.per.connection设置，调整后还有提升空间。

稳定性方面，MM2和Connect方案都依赖Kafka Connect的rebalance机制，topic数量多、任务数大的时候，重启一个worker会导致所有任务重新分配，短则几十秒、长则几分钟的"空窗"。自研管道的稳定性完全取决于你写的代码和网络质量，没有框架层面的容错。这些都是真实运维中会碰到的问题，提前有预期，故障时才不会慌。

6.3 我的最终建议与使用体会

按场景给一个我的选择习惯：跨机房容灾且要求切换后消费组能衔接的，优先MM2，把checkpoint间隔调小，配合业务端幂等；只是持续搬运几个topic、下游能容忍重复的，用Connect自定义Connector，顺带把改名脱敏一起做了；一次性搬迁、跑几天就撤的，自研Consumer-Producer管道，注意batch send和commit的顺序；多集群网状复制、需要集中管理配置和ACL的，MM2是性价比最高的起点；团队小、预算允许、业务核心链路非常重要的，考虑商业复制方案或云托管链接，买的是SLA和运维保障。

我最近做的一次跨机房容灾演练，最终就是MM2加自定义SMT的组合：用IdentityReplicationPolicy去掉topic前缀，把checkpoint间隔调到5秒，再在下游消费端做了幂等兜底。演练时切换流量，目标端消费者只重复了最近几秒的消息，业务方确认可以接受。整个过程最有价值的部分反而是演练本身——把"切换后offset对不上"的问题在正式风险暴露之前就发现并解决掉了。

最后再分享一个小技巧：不管用哪种方案，都建议在目标端对同步过来的topic设置3副本并保证min.insync.replicas不小于2，同时在同步时把"源集群别名+partition+offset"作为header附加到每条消息上。这样下游做去重、排查问题时都有明确依据。这个习惯在跨集群数据链路出故障时，能帮你省下大量排查时间。