Kafka MirrorMaker是Kafka官网提供的跨数据中心流数据同步方案，其实现原理是通过从Source集群消费消息，然后将消息生产到Target集群从而完成数据迁移操作。用户只需要通过简单的consumer配置和producer配置，启动MirrorMaker，即可实现实时数据同步。
本文章主要聚焦跑通Kafka MirrorMaker数据迁移流程。实验中的Source Kafka版本为2.12，基于本地机器搭建。Sink集群为火山引擎Kafka中间件
步骤1：本地Kafka创建测试Topic
以下我们将以名称为“testTopic”的Topic为例演示。
创建Topic命令：
kafka-topics.sh \
--create \
--zookeeper localhost:2181 \ #根据实际情况填写
--replication-factor 1 \
--partitions 1 \
--topic testTopic创建成功后可以通过以下命令对topic进行检查
bin/kafka-topics.sh \
--list \
--zookeeper localhost:2181 #根据实际情况填写执行如下：
步骤2：同步创建火山kafka Topic
创建Kafka实例后，在“Topic管理”页签下，创建同名Topic。注意分区数最好保持与原集群分区保持一致。
步骤3：下载SASL_SSL证书
在下载SASL_SSL证书前，先确认用于访问的用户是否已经存在：
如果未建立，请先创建用户。
确认完成后，在“实例管理”页签下下载SASL_SSL证书
关于使用SASL_SSL可参考：文档中心-火山引擎
步骤4：修改Mirror Maker 生产者/消费者配置
consumer生产者的配置（consumer.properties）一般在kafka目录下的config目录下。修改如下：
bootstrap.servers=localhost:9092 # 需要根据实际情况修改
group.id=test-consumer-group # 需要根据实际情况修改同样，producer消费者的配置（producer.properties）也在此config目录下，该文件有较大修改：
bootstrap.servers= SASL接入点（公网） # 需要根据实际情况修改接入点获取途径如下：
外网访问，需要添加SASL认证信息：
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="替换用户名" password="替换密码";
sasl.mechanism=PLAIN security.protocol=SASL_SSL
ssl.truststore.location=/xxx/Kafka.client.truststore.jks #根据实际情况替换证书路径
ssl.truststore.password=KafkaTrustStorePass ssl.endpoint.identification.algorithm=步骤5：启动MirrorMaker
文件配置修改完成后，通过下方命令启动MirrorMaker实现本地Kafka与火山Kafka联通
kafka-mirror-maker.sh \
--consumer.config ../config/consumer.properties \ #根据实际情况指定consumer.properteis
--producer.config ../config/producer.properties \ #根据实际情况指定
producer.properties --whitelist "testTopic"步骤6：启动Producer生产数据
我们启动生产者对测试Topic进行消息生产
kafka-console-producer.sh \
--broker-list localhost:9092 \ #根据实际情况填写
--topic testTopic
步骤7：数据同步结果检查
在火山引擎Kafka实例“消息查询”页签，我们可以查询testTopic最近的数据，发现是有数据写入的。此时数量上和我们写入的数量一致。
由于火山对下载的消息进行了 Base64 编码传输，因此很难确认消息是否正确性、完整性。
可以通过客户端消费如下（客户端下载与使用可参考：文档中心-火山引擎）：
经过检查，消息与发送端数据保持一致。
可能出现的问题：
（1）MirrorMakker启动报错：java.lang.OutOfMemoryError: Java heap space
解决方法：修改 /bin/kafka-run-class.sh，找到 Memory options处，默认设置是256M，将其修改为如下值：
if [ -z "$KAFKA_HEAP_OPTS" ]; then KAFKA_HEAP_OPTS="-Xmx1024M -Xms512M" fi保存退出。
（2）Error while fetching metadata with correlation
修改 config\server.properties,修改内容如下：
listeners=PLAINTEXT://localhost:9092
advertised.listeners=PLAINTEXT://localhost:9092重启kafka，启动程序进行入库操作
MirrorMaker的配置说明：
--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
详情见：Kafka mirroring (MirrorMaker) - Apache Kafka - Apache Software Foundation
其他注意事项：
1）whitelist和blacklist支持正则表达式。比如需要包含两个topic可以这样写，--whitelist 'A|B' or --whitelist 'A,B' ，或者想迁移所有topic可以这样写 --whitelist '*'
2）注意在迁移之前创建好相关topic以及规划好partition数量。
3）老版本和新版本迁移主要考虑consumer和producer的兼容性
4）如果允许的话，建议将MirrorMaker部署在目标集群内，这是因为如果一旦发生网络分区，消费者与源集群断开连接比生产者与目标集群断开连接要安全。如果消费者断开连接，那么只是当前读取不到数据，但是数据仍然在源集群内，并不会丢失；而生产者断开连接，MirrorMaker便生产不了数据，如果MirrorMaker本身处理不当，可能会丢失数据。
5）开始之前配置好限流，防止影响原来集群的正常工作。
