卡夫卡的藏书阁【Book28】- Kafka - MirroMaker

“In man's struggle against the world, bet on the world.”
― Franz Kafka

Kafka ＭirrorＭaker 是 Kafka 提供的一个工具去让你可以将来源丛集 ( Cluster ) 的资料同步到目标丛集，基本上的做法很单纯就是利用消费者去消费来源丛集指定的Topic，再使用生产者去新增到目标丛集，藉此达到资料同步的目的。比较常见的使用场景是灾害备援，将资料镜像复制到另外一个备份的资料中心当备援，或是将资料镜像後提供给不同消费者去使用，这样可以提高不少的吞吐量和容错度。

目前环境：
1个 zookeeper: localhost:2181
3个 broker: localhost:9092, :9093, :9094

准备工作：
新建一个 zookeeper 使用 localhost:2182
新建3个 broker 使用 localhost:9192, 9193, 9194

$ cd /usr/local/etc/kafka
$ cp zookeeper.properties zookeeper-2.properties
$ vim zookeeper-2.properties

修改内容如下：
dataDir=/usr/local/var/lib/zookeeper-2
clientPort=2182

• 修改 mirror-server-1.properties

broker.id=200
listeners=PLAINTEXT://localhost:9192
log.dirs=/usr/local/var/lib/kafka-logs/broker200
zookeeper.connect=localhost:2182

• 修改 mirror-server-2.properties

broker.id=201
listeners=PLAINTEXT://localhost:9193
log.dirs=/usr/local/var/lib/kafka-logs/broker201
zookeeper.connect=localhost:2182

• 修改 mirror-server-3.properties

broker.id=202
listeners=PLAINTEXT://localhost:9194
log.dirs=/usr/local/var/lib/kafka-logs/broker202
zookeeper.connect=localhost:2182

• 启动三台 Broker

$ kafka-server-start /usr/local/etc/kafka/mirror-server-1.properties
$ kafka-server-start /usr/local/etc/kafka/mirror-server-2.properties
$ kafka-server-start /usr/local/etc/kafka/mirror-server-3.properties

查看 Broker 详细资料，可以看到 Broker200 的 cluster.id 是跟原本broker1是不一样的，因为是连到不同Zookeeper

$ grep --color 'log.dirs' mirror-server-1.properties
log.dirs=/usr/local/var/lib/kafka-logs/broker200
$ cd /usr/local/var/lib/kafka-logs/broker200
$ cat meta.properties
#
#Fri Sep 03 12:08:56 CST 2021
broker.id=200
version=0
cluster.id=jwCSNMp4RLGf9aQg__VFXw

准备mirror-consumer.properties和mirror-producer.properties

$ cp consumer.properties mirror-consumer.properties

修改内容如下：
bootstrap.servers=localhost:9192,localhost:9193,locaslhost:9194
group.id=test-MirrorMaker-group
exclude.internal.topics=true
mirror.topics.whitelist=app_log
client.id=mirror_maker_consumer

$ cp producer.properties mirror-producer.properties

修改内容如下：
bootstrap.servers=localhost:9192,localhost:9193,localhost:9194
acks=1
batch.size=100
client.id=mirror_maker_producer

在目标Cluster建立跟来源Cluster一样的Topic
另外，如果你的消费者在目标丛集是平行并发去消费的话，那很重要的话是，你必须要记得要新增建立跟目标丛集一样数量的分区 ( partition )，举例来说，如果你的主题名称为 keyin-logs 的主题拥有六个分区在目标丛集，要确保你的丛集也有六个分区，如果你的目标丛集未预先建立好分区，那 mirrormaker 会尝试去建议主题，目前丛集会依据设定档的 num.partitions 参数和 num.replicas 参数去建立分区，但这个数量可能就不会符合使用者希望的分区数量

$ kafka-topics --create --zookeeper 127.0.0.1:2182 --replication-factor 2 --partitions 3 --topic topicWithThreeBroker

在哪里使用MirrorMaker

通常建议在目标丛集上使用 MirrorMaker

以下为指令 MirrorMaker 的相关参数

$ kafka-mirror-maker

可选参数                                   说明
------                                   -----------
--abort.on.send.failure <String: Stop    喷错失败的时候退出 mirror maker
  the entire mirror maker when a send    (预设值: true)
  failure occurs>
--consumer.config <String: config file>  消费者设定档
--consumer.rebalance.listener <String:   消费者重新平衡监听器
  A custom rebalance listener of type   
  ConsumerRebalanceListener>            
--message.handler <String: A custom      处理生产者跟消费者间的讯息处理器
  message handler of type                
  MirrorMakerMessageHandler>             
--message.handler.args <String:          Mirror Maker 使用的讯息处理器
  Arguments passed to message handler   
  constructor.>
--num.streams <Integer: Number of        消费者串流的数量
  threads>                                 (预设值: 1)
--offset.commit.interval.ms <Integer:    提交偏移量的间隔时间
  offset commit interval in                (预设值: 60000毫秒)
  millisecond>
--producer.config <String: config file>  Embedded producer config.
--rebalance.listener.args <String:       Arguments used by custom rebalance
  Arguments passed to custom rebalance     listener for mirror maker consumer.
  listener constructor as a string.>
--version                                显示 Kafka 的版本
--whitelist <String: Java regex          指定要进行镜像的 topic，这个参数支援 Java 的正则表达式，可以使用`--whitelist 'topicA|topicB`，或是`--whitelist '*'`
  (String)>