mirror maker2背景

通常情况下，我们都是使用一套kafka集群处理业务。但有些情况需要使用另一套kafka集群来进行数据同步和备份。在kafka早先版本的时候，kafka针对这种场景就有推出一个叫mirror maker的工具（mirror maker1，以下mm1即代表mirror maker1），用来同步两个kafka集群的数据。

最开始版本的mirror maker本质上就是一个消费者 + 生产者的程序。但它有诸多诸多不足，包括

1. 目标集群的Topic使用默认配置创建，但通常需要手动repartition。
2. acl和配置修改的时候不会自动同步，给多集群管理带来一些困难
3. 消息会被DefaultPartitioner打散到不同分区，即对一个topic ，目标集群的partition与源集群的partition不一致。
4. 任何配置修改，都会使得集群变得不稳定。比如比较常见的增加topic到whitelist。
5. 无法让源集群的producer或consumer直接使用目标集群的topic。
6. 不保证exactly-once，可能出现重复数据到情况
7. mm1支持的数据备份模式较简单，比如无法支持active <-> active互备
8. rebalance会导致延迟

因为存在这些问题，mirror maker难以在生产环境中使用。所以kafka2.4版本，推出一个新的mirror maker2（以下mm2即代表mirror maker2）。mirror maker2基于kafka connect工具，解决了上面说的大部分问题。

今天主要介绍mirror maker2的设计，主要功能和部署。

设计和功能

整体设计

mirror maker2是基于kafka connect框架进行开发的，可以简单地将mirror maker2视作几个source connector和sink connector的组合。包括：

• MirrorSourceConnector, MirrorSourceTask：用来进行同步数据的connector
• MirrorCheckpointConnector, MirrorCheckpointTask：用来同步辅助信息的connector，这里的辅助信息主要是consumer的offset
• MirrorHeartbeatConnector, MirrorHeartbeatTask：维持心跳的connector

不过虽然mirror maker2岁基于kafka connect框架，但它却做了一定的改造，可以单独部署一个mirror maker2集群，当然也可以部署在kafka connect单机或kafka connect集群环境上。这部分后面介绍部署的时候再介绍。

和mm1一样，在最简单的主从备份场景中，mm2建议部署在目标（target）集群，即从远端消费然后本地写入。如果部署在源集群端，那么出错的时候可能会出现丢数据的情况。

其整体架构如图：

内部topic设计

mm2会在kafka生成多个内部topic ，来存储源集群topic相关的状态和配置信息，以及维持心跳。主要有三个内部topic：

• hearbeat topic
• checkpoints topic
• offset sync topic

这几个内部topic都比较好理解，一看名字基本就知道是干嘛用的，值得一提的是这其中checkpoints和hearbeat功能都可以通过配置关闭。下面我们详细介绍下这几个topic的功能和数据格式。

heartbeat topic

在默认的配置中，源集群和目标集群都会有一个用于发送心跳的topic，consumer 客户端通过这个 topic，一方面可以确认当前的 connector 是否存活，另一方面确认源集群是否处于可用状态。

heartbeat topic的schema如下：

• target cluster：接收心跳集群
• source cluster：发送心跳的集群
• timestamp：时间戳

checkpoints topic

对应的connector（即MirrorCheckpointConnector）会定期向目标集群发送checkpoint信息，主要是consumer group提交的offset ，以及相关辅助信息。

checkpoints topic 的schema如下：

• consumer group id (String)
• topic (String) ：包含源集群和目标集群的 topic
• partition (int)
• upstream offset (int): 源集群指定consumer group已提交的offset（latest committed offset in source cluster）
• downstream offset (int): 目标集群已同步的offset（latest committed offset translated to target cluster）
• metadata (String)：partition元数据
• timestamp

mm2提供的另一个功能，consumer切换集群消费就是通过这个topic实现的。因为这个topic中存放了源集群consumer group的消费offset，在某些场景（比如源集群故障）下要切换consumer到目标集群，就可以通过这个topic获取消费offset然后继续消费。

offset sync

这个topic ，主要是在两个集群间同步topic partition的offset。这里的offset并不是consumer的offset，而是日志的offset。

它的 schema 如下：

• topic (String)：topic 名
• partition (int)
• upstream offset (int)：源集群的 offset
• downstream offset (int)：目标集群的 offset，和源集群的应该保持一致

config sync

mm2会将源集群的数据同步到目标集群，那么目标集群对应的topic的读写权限上怎样的呢？mm2约定了，目标集群对应的topic（源集群备份的那个）只有source和sink connector能够写入。为了实施此策略，MM2使用以下规则将 ACL 策略传播到下游主题：

• 若用户对源集群的topic有read的权限，那么对目标集群对应的topic也有read的权限
• 除了mm2，别的用户都不能写入目标集群对应的topic

同时会同步topic相关配置信息

acl

consumer切换集群

源集群的consumer group offset ，是存储在目标集群的checkpoint topic中，这点我们上面已经有说到过。要获取这些offset信息，可以使用MirrorClient#remoteConsumerOffsets这个 api，然后就能用 consumer#seek api 根据给出的offset消费。

这里顺便提供下大致代码，首先maven添加依赖：

    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>connect-mirror</artifactId>
      <version>2.4.0</version>
    </dependency>
    <dependency>
      <groupId>org.apache.kafka</groupId>
      <artifactId>connect-mirror-client</artifactId>
      <version>2.4.0</version>
    </dependency>

        MirrorMakerConfig mmConfig = new MirrorMakerConfig(mm2.getProp());
        MirrorClientConfig mmClientConfig = mmConfig.clientConfig("target-cluster");
        MirrorClient mmClient = new MirrorClient(mmClientConfig);

        Map<TopicPartition, OffsetAndMetadata> offsetMap =
                mmClient.remoteConsumerOffsets("my-consumer-group", "source-cluster", Duration.ofMinutes(1));

# 指定两个集群，以及对应的host
clusters = us-west, us-east
us-west.bootstrap.servers = host1:9092
us-east.bootstrap.servers = host2:9092

# 指定同步备份的topic & consumer group，支持正则
topics = .*
groups = .*
emit.checkpoints.interval.seconds = 10

# 指定复制链条，可以是双向的
us-west->us-east.enabled = true
# us-east->us-west.enabled = true  # 双向，符合条件的两个集群的topic会相互备份

# 可以自定义一些配置
us-west.offset.storage.topic = mm2-offsets
# 也可以指定是否启用哪个链条的hearbeat，默认是双向hearbeat的
us-west->us-east.emit.heartbeats.enabled = false

PUT /connectors/us-west-source/config HTTP/1.1
 
{
    "name": "us-west-source",
    "connector.class": "org.apache.kafka.connect.mirror.MirrorSourceConnector",
    "source.cluster.alias": "us-west",
    "target.cluster.alias": "us-east",
    "source.cluster.bootstrap.servers": "us-west-host1:9091",
    "topics": ".*"
}