9.4. 使用 MirrorMaker 2.0 在 Kafka 集群间同步数据

流程

1. 在文本编辑器中打开示例属性文件，或创建一个新属性文件，再编辑 文件以包含连接信息和每个 Kafka 集群的复制流。

   以下示例显示了连接两个集群 cluster-1 和 cluster -2 （双向）的配置。集群名称可以通过 clusters 属性进行配置。

   clusters=cluster-1,cluster-2 1
   
   cluster-1.bootstrap.servers=CLUSTER-NAME-kafka-bootstrap-PROJECT-NAME:443 2
   cluster-1.security.protocol=SSL 3
   cluster-1.ssl.truststore.password=TRUSTSTORE-NAME
   cluster-1.ssl.truststore.location=PATH-TO-TRUSTSTORE/truststore.cluster-1.jks
   cluster-1.ssl.keystore.password=KEYSTORE-NAME
   cluster-1.ssl.keystore.location=PATH-TO-KEYSTORE/user.cluster-1.p12_
   
   cluster-2.bootstrap.servers=CLUSTER-NAME-kafka-bootstrap-<my-project>:443 4
   cluster-2.security.protocol=SSL 5
   cluster-2.ssl.truststore.password=TRUSTSTORE-NAME
   cluster-2.ssl.truststore.location=PATH-TO-TRUSTSTORE/truststore.cluster-2.jks_
   cluster-2.ssl.keystore.password=KEYSTORE-NAME
   cluster-2.ssl.keystore.location=PATH-TO-KEYSTORE/user.cluster-2.p12_
   
   cluster-1->cluster-2.enabled=true 6
   cluster-1->cluster-2.topics=.* 7
   cluster-2->cluster-1.enabled=true 8
   cluster-2->cluster-1B->C.topics=.* 9
   
   replication.policy.separator=- 10
   sync.topic.acls.enabled=false 11
   refresh.topics.interval.seconds=60 12
   refresh.groups.interval.seconds=60 13

   1

   每个 Kafka 集群都使用其别名识别。

   2

   cluster-1 的连接信息，使用 bootstrap 地址和 端口 443。两个集群都使用端口 443 来使用 OpenShift Route 连接到 Kafka。

   3

   ssl. 属性为 cluster-1 定义 TLS 配置。

   4

   cluster-2 的连接信息。

   5

   ssl. 属性为 cluster-2 定义 TLS 配置。

   6

   从 cluster-1 集群 启用到 cluster-2 集群的 复制流。

   7

   将 cluster-1 集群中的所有主题复制到 cluster-2 集群。

   8

   从 cluster-2 集群 启用到 cluster-1 集群的 复制流。

   9

   将特定主题从 cluster-2 集群复制到 cluster-1 集群。

   10

   定义用于重命名远程主题的分隔符。

   11

   启用后，ACL 将应用到同步的主题。默认值为 false。

   12

   检查要同步的新主题之间的期间。

   13

   检查要同步的新消费者组之间的期间。

2. 选项：如果需要，添加可覆盖远程主题自动重命名的策略。该主题不会用源集群的名称来附加名称，而是保留其原始名称。

   此可选设置用于主动/被动备份和数据迁移。

   replication.policy.class=io.strimzi.kafka.connect.mirror.IdentityReplicationPolicy

3. OPTION：如果要同步消费者组偏移，请添加配置来启用和管理同步：

   refresh.groups.interval.seconds=60
   sync.group.offsets.enabled=true 1
   sync.group.offsets.interval.seconds=60 2
   emit.checkpoints.interval.seconds=60 3

   1

   可选设置来同步消费者组偏移，这对于主动/被动配置中的恢复非常有用。默认情况下不启用同步。

   2

   如果启用了消费者组偏移的同步，您可以调整同步的频率。

   3

   调整检查偏移跟踪的频率。如果更改了偏移同步的频率，您可能还需要调整这些检查的频率。

4. 在目标集群中启动 ZooKeeper 和 Kafka：

   su - kafka
   /opt/kafka/bin/zookeeper-server-start.sh -daemon /opt/kafka/config/zookeeper.properties

   /opt/kafka/bin/kafka-server-start.sh -daemon /opt/kafka/config/server.properties

5. 使用您在属性文件中定义的集群连接配置和复制策略启动 MirrorMaker：

   /opt/kafka/bin/connect-mirror-maker.sh /config/connect-mirror-maker.properties

   MirrorMaker 设置集群间的连接。

6. 对于每个目标集群，验证是否正在复制这些主题：

   /bin/kafka-topics.sh --bootstrap-server <BrokerAddress> --list