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

流程

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

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

   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

   使用 bootstrap address 和端口 443 用于 cluster-1 的连接信息。两个集群都使用端口 443 连接到使用 OpenShift Routes 的 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. OPTION：如果需要，请添加一个策略来覆盖远程主题的自动重命名。该主题不会用源集群的名称来附加名称，而是保留其原始名称。

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

   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