Kafka异地双活深度讲解 - Mirrormaker V2

总结：Apache Kafka Mirrormaker V1的解决方案在提供企业管理的灾难恢复方面存在局限性。MM V2（KIP-382）针对MM V1 进行了扩展，并修复了MM V1的局限性，使其能够动态修改配置，并且能够将Topic在群集之间保持同步，同时尽可能地降低触发Rebalance的情况以提高性能。此外，Active-Active群集和Disaster Recover在MM V2中已经属于开箱即用（Out-of-the-box）功能。

MirrorMaker2架构

01

MM V2的核心架构基于Kafka Connect框架，可以抽象的理解为是一个Kafka Connect 里的Source Connector和Sink Connector的组合体。MM V2建议采用与MM V1 中一致的Remote consume和Local produce的部署模式。因此在最简单的，也就是Source – Target 复制场景下，MM V2服务是部署在Target数据中心的。同时MM V2的 Connect框架所需的Primary集群与Target 的Kafka集群是共用的。

（点击查看大图）

为什么不使用Kafka MM V1 来实现Kafka的跨集群复制？

02

MM V1 中Topic的命名问题

基于MM V1的Kafka集群复制通常需要将Source的Topic名和Target的Topic名保持相同。这样Topic命名过程会导致在Active – Active双活时造成无限的消息循环。

在MM V2中，会通过增加预先配置的前缀（例如，群集别名是DCX或DCY）到Target的Topic的命名来解决这个问题。

例如，在Active - Active 场景下复制两个数据中心DCX，DCY的两个Kafka群集，MM V2会过滤掉前缀中带有目标群集名称的任何Topic。而Consumer可以订阅模糊匹配的多个Topic，例如，"* TopicA" 从源群集中消费，并在故障转移后自动继续从目标群集消费。

（点击查看大图）

主备Consumer Offset 管理

在MM V1中，Source集群的Topic Partition Offset和目标群集上Topic Partition Offset 几乎不可能相同。因此，Consumer提交到Source集群的Committed Offset 在Target集群中是不可用的。如果Consumer切换到目标群集，也就不能简单地使用原有的Committed Offset来继续消费。一种处理办法是依赖Kafka对消息时间戳的支持，但是这个解决办法不够完美，因为涉及到了猜测时间和重复消费的问题。

（点击查看大图）

MM V2的实现则完全不同，它采用了2个内部Topic来跟踪源和目标的offset mapping。同时也自动的完成了Source 的consumer_offsets与target 的 consumer_offsets之间的转换。技术实现上是通过Target集群中的 "offset_sync " topic对Source的Consumer offset与Target的Consumer Offset 进行了映射处理。对依赖 "__consumer_offsets" Topic来跟踪进度的Consumer Groups，MM V2会将Consumer offset映射到Target群集中的 "__checkpoint" Topic中，这是一个Log Compacted Topic。同时定期向源集群查询来自所有Consumer Group已提交的offset，并向Target集群的"__checkpoint" Topic发送消息来完成Offset管理的动作。

（点击查看大图）

因此，在MM V2 中，通过使用 "__checkpoint" Topic，Consumer在故障迁移时时可以直接确定（使用MM V2 API）需要开始消费的目标集群offset。

减少MirrorMaker集群数量

传统上，MM V1 群集与目标群集共用。因此，在使用Remote Consume和Local Produce模式之后，每个目标集群都会有一个镜像集群。

（点击查看大图）

于大型的数据中心，这样会明显的增加运营成本。而理想的情况是，每一对儿数据中心原则上应该只有一个MM集群。

在MM V2的实现下，Kafka Connect框架会假定Source Connetor从外部源读取并写入Kafka，而Sink Connector从Kafka读取并写入外部接收器。因此每个Target数据中心只需要一个Connect集群，在该对数据中心上复制的所有Kafka集群都可以由一个MM V2集群处理。

（点击查看大图）

更加灵活的白/黑名单控制

MM V1会将白名单和黑名单与正则表达式或Topic列表结合使用。但这些都是静态配置的。也就是说，当创建一个与白名单匹配的新Topic时，会在Target集群上创建新Topic，并自动进行复制。但是，白名单本身更新时，它需要重启。每次列表更改时重新启动MM V1都会在造成数据堆积，从而导致重启后的复制吞吐风暴。在MM V2中，可以使用REST API动态更改Topic列表和正则表达式的配置，不需要重启服务。

为什么不直接用Kafka Connect来实现Kafka的跨集群复制？

03

Kafka Connect框架的Kafka重依赖问题

Kafka Connect框架需要有一个Kafka集群来存储状态，在Connect中叫“Primary”集群。在典型的Kafka Connect配置中，Source Connector将数据从外部源写入Kafka集群，而Sink Connector从Kafka集群读取数据并写入外部存储库。在Source – Target 复制场景下，Connect的Primary集群是我们的Target Kafka集群。

如果我们只是采用Kafka Source和Connect连接器并将它们串联起来实现kafka的灾备，那么数据先写入Primary Kafka 集群然后再读取出来。MM V2 则是从Source直接传递给Sink 从而避免了这种不必要的数据复制。

同时，在Active – Active场景下，没有必要为每个Kafka集群建一个Primary群集。

Rebalance的频繁触发

MirrorMaker2中使用的Kafka Connect框架原生采用了Kafka的High Level Consumer从Kafka读取数据。High Level Consumer会自动地使Consumer Group中消费的分区在整个组中自动平衡。每次Topic的元数据发生更改时，例如改分区计数，或更改Connect Worker的数量等等，会触发Connect rebalance。频繁的重新平衡会导致阻塞，并且严重的影响Target集群的吞吐。

在MM V2中，我们使用了Low Level Consumer 去Consume给定的分区列表，因此可以避免由于Topic的分区数更改而触发的Rebalance动作。因此，对Topic和分区数的任何更改都不会导致完全的重新平衡。但是，需要注意的是，由Connect集群本身（例如添加更多Worker Node等）的更改触发的重新平衡是无法避免的。在大多数情况下，这些变化比Topic的变化次数要少的多。

MM V2目前的一些局限性及未来改进

04

跨集群有且只有一次的消息复制

Kafka提供有且只有一次（EOS）的消息处理，但该特性仅是针对某一个具体的Kafka集群，而在跨集群的场景下并不适用。因此跨群集复制无法直接利用这个特性。也就是说，当前的MM2在源和目标集群之间复制数据时只能提供至少一次语义，下游可能存在重复记录。

来看一下跨集群复制上在哪个环节会出现数据重复。如果我们将复制的动作看做一个从Source集群Consume并Produce到Target集群的流，那么消费Source端的Consumer会写Source端的"__consumer_offsets" Topic来跟踪这个Consumer的状态。同时会把数据写入下游的对应Topic中

（点击查看大图）

这两个“Write”操作不能做成原子事务，因为它们跨越两个不同的集群，总是有可能在其中一个失败时导致数据重复。

如何才能做到跨集群的有且只有一次的消息处理？其实和其他流数据处理系统一样，在MM V2中，我们有一个"__checkpoint" Topic是在Target集群上的，它是用来来跟踪Source的Consumer状态。因此，MM V2可以通过这个内部Topic与目标Topic处于同一事务中来提供EOS语义。这个功能即将在MM V2的下一次迭代中推出。

高吞吐的对等复制

在复制Kafka集群的场景中，如果源和目标在相同Topic，相同Partition数，相同的分区方案，相同压缩，相同serde的情况下，我们称之为对等镜像。在这种情况下，理想情况是将一批记录作为字节流读取并将其写入而不进行任何处理。这样我们可以绕过Consume，解析，序列化，Produce这个流程。对等复制可以比传统方法提供更高的吞吐。这是MM V2即将推出的另一项功能。

参考文章：

【A look inside Kafka Mirrormaker 2】：

https://blog.cloudera.com/a-look-inside-kafka-mirrormaker-2/

【Kafka - KIP 382 】：

https://cwiki.apache.org/confluence/display/KAFKA/KIP-382%3A+MirrorMaker+2.0

【KIP 382 Repo】：

https://github.com/apache/kafka/pull/6295