一、MirrorMaker介绍

MirrorMaker是Kafka附带的一个用于在Kafka集群之间制作镜像数据的工具。该工具从源集群中消费并生产到目标群集。这种镜像的常见用例是在另一个数据中心提供副本。

 

图1. MirrorMaker

对于迁移的topic而言，topic名字一样， partition数量可以不一样，消息的offset会不一样。

二、注意事项

1）whitelist和blacklist支持正则表达式。比如需要包含两个topic可以这样写，--whitelist 'A|B' or --whitelist 'A,B' ，或者想迁移所有topic可以这样写 --whitelist '*'

2）注意在迁移之前创建好相关topic以及规划好partition数量。

3）老版本和新版本迁移主要考虑consumer和producer的兼容性

4）如果允许的话，建议将MirrorMaker部署在目标集群内，这是因为如果一旦发生网络分区，消费者与源集群断开连接比生产者与目标集群断开连接要安全。如果消费者断开连接，那么只是当前读取不到数据，但是数据仍然在源集群内，并不会丢失；而生产者断开连接，MirrorMaker便生产不了数据，如果MirrorMaker本身处理不当，可能会丢失数据。

5）开始之前配置好限流，防止影响原来集群的正常工作。

三、进度监控

1） 检测MirrorMaker提交到源集群的位移。我们可以使用kafka-consumer-groups来检测分区的最新位移以及MirrorMaker提交的位移，通过计算差值得到落后间隔。

2） 查看Ckafka的Topic相关监控，查看当前进度

四、使用方法（0.8.2版本）

运行指令：

> bin/kafka-mirror-maker.sh

--consumer.config consumer.properties

--producer.config producer.properties

--new.producer

--whitelist “my-topic1,my-topic2”

配置说明：

--consumer.config # 消费者配置，详情参考kafka consumer配置

--producer.config # 生产者配置，详情参考kafka producer配置

--whitelist #需要mirror的topic，支持Java正则表达式，例如'ABTestMsg,AppColdStartMsg’

--blacklist #不需要拷贝的topic，支持Java正则表达式

--num.producers #producer数量，默认为1

--num.streams #consumer数量，默认为1

--queue.size #consumer和producer之间缓存的queue size，默认10000

Config demo

mirror_maker_consumer.properties 文件

zookeeper.connect=ZK_IP:PORT

group.id=mirror_maker-group

fetch.message.max.bytes=1048576

auto.commit.enable=true

auto.offset.reset=smallest

auto.commit.interval.ms=1000

mirror_maker_producer.properties 文件

bootstrap.servers=broker_ip:port

acks=-1

block.on.buffer.full=true

linger.ms=10

batch.size=16384

retries=3

五、使用方法（0.10.2版本）

运行指令：

> bin/kafka-mirror-maker.sh

--consumer.config consumer.properties

--producer.config producer.properties

--whitelist “my-topic1,my-topic2”

Config demo

mirror_maker_producer.properties 文件

bootstrap.servers=broker_ip:port

acks=1

linger.ms=100

batch.size=16384

retries=3

mirror_maker_consumer.properties 文件

bootstrap.servers=broker_ip:port

group.id=mirror_maker-group

enable.auto.commit=true

auto.offset.reset=earliest

auto.commit.interval.ms=1000