如何在数据迁移、复制和用例中使用 Kafka MirrorMaker 2.0

MirrorMaker 2.0 (MM2) 旨在更轻松地将主题从一个 Kafka 群集镜像或复制到另一个群集。 它使用 Kafka Connect 框架来简化配置和缩放。 它动态检测主题的更改，并确保源和目标主题属性同步，包括偏移和分区。

本文介绍如何在数据迁移/复制和用例中使用 Kafka MirrorMaker 2.0。

• 环境至少有两个 HDI Kafka 群集。
• Kafka 版本高于 2.4 (HDI 4.0)
• 源群集应具有数据点和主题来测试 MirrorMaker 2.0 复制过程的各种功能

模拟 MirrorMaker 2.0 以在 HDInsight 中的两个 Kafka 群集之间复制数据点/偏移。 这样同样可用于像在两个或更多 Kafka 群集之间进行必需的数据复制这样的场景，例如灾难恢复、云适应、异地复制、数据隔离和数据聚合。

MirrorMaker 2.0 工具由不同的连接器组成。 这些连接器是标准 Kafka Connect 连接器，可以在独立模式或分布式模式下直接与 Kafka Connect 配合使用。

代理设置过程的摘要如下：

MirrorSourceConnector：

1. 复制单个源群集的远程主题、主题 ACL 和配置。
2. 向内部主题发出偏移同步。

MirrorSinkConnector：

1. 使用主群集并将主题复制到单个目标群集。

MirrorCheckpointConnector：

1. 使用 offset-syncsr。
2. 发出检查点以启用故障转移点。

MirrorHeartBeatConnector：

1. 向远程群集发出检测信号，从而监视复制过程。

1. 与 Kafka 库捆绑的 connect-mirror-maker.sh 脚本实现分布式 MM2 群集，该群集根据配置文件在内部管理 Connect 辅助角色。 在内部，MirrorMaker 驱动程序创建和处理每个连接器的配对 - MirrorSourceConnector、MirrorSinkConnector、MirrorCheckpoint 连接器和 MirrorHeartbeatConnector。

2. 启动 MirrorMaker 2.0。

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

export KAFKA_OPTS="-Djava.security.auth.login.config=<path-to-jaas.conf>"    

# specify any number of cluster aliases
clusters = source, destination

# connection information for each cluster
# This is a comma separated host:port pairs for each cluster
# for example. "A_host1:9092, A_host2:9092, A_host3:9092"  and you can see the exact host name on Ambari > Hosts
source.bootstrap.servers = wn0-src-kafka.bx.internal.chinacloudapp.cn:9092,wn1-src-kafka.bx.internal.chinacloudapp.cn:9092,wn2-src-kafka.bx.internal.chinacloudapp.cn:9092
destination.bootstrap.servers = wn0-dest-kafka.bx.internal.chinacloudapp.cn:9092,wn1-dest-kafka.bx.internal.chinacloudapp.cn:9092,wn2-dest-kafka.bx.internal.chinacloudapp.cn:9092

# enable and configure individual replication flows
source->destination.enabled = true

# regex which defines which topics gets replicated. For eg "foo-.*"
source->destination.topics = toa.evehicles-latest-dev
groups=.*
topics.blacklist="*.internal,__.*"

# Setting replication factor of newly created remote topics
replication.factor=3

checkpoints.topic.replication.factor=1
heartbeats.topic.replication.factor=1
offset-syncs.topic.replication.factor=1

offset.storage.replication.factor=1
status.storage.replication.factor=1
config.storage.replication.factor=1    

如果设置需要 SSL 配置

destination.security.protocol=SASL_SSL
destination.ssl.truststore.password=<password>
destination.ssl.truststore.location=/path/to/kafka.server.truststore.jks
#keystore location in case client.auth is set to required
destination.ssl.keystore.password=<password> 
destination.ssl.keystore.location=/path/to/kafka.server.keystore.jks
destination.sasl.mechanism=GSSAPI

为什么在主题的源群集和目标群集帖子复制的最后偏移中会看到差异？

有可能是源主题的数据点已被清除，因此实际记录数将小于最后的偏移值。 这会导致源群集和目标群集帖子复制的最后偏移之间存在差异，因为复制始终从目标群集的偏移值 0 开始。

如果目标群集可能具有映射到数据点的其他偏移，则使用者在迁移时的行为如何？

MirrorMaker 2.0 MirrorCheckpointConnector 功能会自动在源群集上存储使用者组的使用者组偏移检查点。 每个检查点都包含源群集中每个组最后提交的偏移到目标群集中同等偏移的映射。 因此，迁移时，从目标群集上的同一主题开始使用的使用者能够从源群集上提交的最后一个偏移继续接收消息。

由于源别名以复制的所有主题为前缀，如何在目标群集中保留确切的主题名称？

这是 MirrorMaker 2.0 中的默认行为，以避免在复杂的镜像拓扑中重写数据。 需要在复制流设计和主题管理方面小心自定义此项，以避免数据丢失。 可以通过对“replication.policy.class”使用自定义复制策略类来完成此操作。

为什么我们会看到在源和目标 Kafka 中创建的新内部主题？

MirrorMaker 2.0 内部主题由连接器创建，以跟踪复制过程、监视、偏移映射和检查点。

为什么 MirrorMaker 在目标群集中只创建两个主题副本，而源具有更多副本？

MirrorMaker 2 不会将主题的复制因子复制到目标群集。 这可以在 MM2 配置中通过指定所需的“replication.factor”数目来控制。 该复制因子的默认值为 2。

如何在 MirrorMaker 2.0 中使用自定义复制策略？

自定义复制策略可通过实现以下接口来创建。

/** Defines which topics are "remote topics", e.g. "us-west.topic1". */
public interface ReplicationPolicy {

    /** How to rename remote topics; generally should be like us-west.topic1. */
    String formatRemoteTopic(String sourceClusterAlias, String topic);

    /** Source cluster alias of given remote topic, e.g. "us-west" for "us-west.topic1".
        Returns null if not a remote topic.
    */
    String topicSource(String topic);

    /** Name of topic on the source cluster, e.g. "topic1" for "us-west.topic1".
        Topics may be replicated multiple hops, so the immediately upstream topic
        may itself be a remote topic.
        Returns null if not a remote topic.
    */
    String upstreamTopic(String topic);

    /** The name of the original source-topic, which may have been replicated multiple hops.
        Returns the topic if it is not a remote topic.
    */
    String originalTopic(String topic);

    /** Internal topics are never replicated. */
    boolean isInternalTopic(String topic);
}

需要将实现添加到 Kafka 类路径，以便对 MM2 属性中的 replication.policy.class 使用类引用。

什么是 Apache Kafka on HDInsight？

MirrorMaker 2.0 更改 Apache Doc

HDI Kafka 的客户端证书设置

HDInsight Kafka

Apache Kafka 2.4 文档

将本地网络连接到 Azure