8.6. 在专用模式下运行 MirrorMaker 2.0

流程

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

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

   MirrorMaker 2.0 配置示例

   clusters=cluster-1,cluster-2 

   1

   
   
   cluster-1.bootstrap.servers=<cluster_name>-kafka-bootstrap-<project_name_one>: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-<project_name_two>: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-2->cluster-1.enabled=true 

   7

   
   cluster-1->cluster-2.topics=.* 

   8

   
   cluster-2->cluster-1.topics=topic-1, topic-2 

   9

   
   cluster-1->cluster-2.groups=.* 

   10

   
   cluster-2->cluster-1.groups=group-1, group-2 

   11

   
   
   replication.policy.separator=- 

   12

   
   sync.topic.acls.enabled=false 

   13

   
   refresh.topics.interval.seconds=60 

   14

   
   refresh.groups.interval.seconds=60 

   15

   1

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

   2

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

   3

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

   4

   cluster-2 的连接信息。

   5

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

   6

   从 cluster-1 启用了复制流到 cluster-2。

   7

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

   8

   将 cluster-1 到 cluster-2 的所有主题复制。源连接器复制指定的主题。检查点连接器可跟踪指定主题的偏移。

   9

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

   10

   将所有消费者组从 cluster-1 复制到 cluster-2。检查点连接器复制指定的消费者组。

   11

   将特定消费者组从 cluster-2 复制到 cluster-1。

   12

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

   13

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

   14

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

   15

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

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

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

   replication.policy.class=org.apache.kafka.connect.mirror.IdentityReplicationPolicy

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

   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 \
   /opt/kafka/config/connect-mirror-maker.properties

   MirrorMaker 设置集群之间的连接。

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

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