MirrorMaker 2.0 检查点连接器会将某个 Kafka 集群的使用方偏移值复制到另一个 Kafka 集群。Kafka 使用方使用使用方偏移量来记录分区中最后一条成功使用的消息。通过复制偏移量，使用方可以切换到目标集群，并从同一点继续处理。
您可以使用此连接器执行以下操作：
帮助确保在从源集群切换到目标集群期间尽可能缩短停机时间。
通过在集群之间提供一致的使用方状态，实现无缝故障切换。
在将数据移至目标集群时，保留使用方进度。
Checkpoint 连接器的工作原理
Checkpoint 连接器可与 MirrorMaker 2.0 Source 连接器搭配使用。Checkpoint 连接器会按如下方式复制偏移量：
对于每个复制的分区，MirrorMaker 2.0 源连接器都会将复制的记录写入目标集群。
写入目标集群的记录通常具有与相应源记录不同的偏移量。为了实现从源偏移量到目标偏移量的转换，MirrorMaker 2.0 源连接器会将偏移量对写入源集群上名为
mm2-offset-syncs.target.internal的主题。这些偏移对会将来源偏移映射到相应的目标偏移。使用方从来源集群读取记录。为了记录进度，消费者会定期提交每个分区中最新处理的消息。Kafka 将提交的偏移量（也称为检查点）存储在名为
__consumer_offsets的主题中。检查点记录包含消费者的消费者群组 ID。MirrorMaker 2.0 检查点连接器会将检查点记录复制到目标集群中名为
source.checkpoints.internal的主题。如果为
sync.group.offsets.enabled=true，则检查点连接器会定期将检查点从source.checkpoints.internal复制到目标集群上的__consumer_offsets。这样，消费者便可以在故障切换后从最新提交的偏移量重新开始。如需恢复处理，消费者在故障切换后使用相同的消费者群组 ID。sync.group.offsets.interval.seconds配置设置用于控制连接器将检查点复制到__consumer_offsets的频率。
创建 MirrorMaker 2.0 检查点连接器
控制台
在 Google Cloud 控制台中，前往连接集群页面。
点击要在其中创建连接器的 Connect 集群。
系统会显示连接集群详情页面。
点击创建连接器。
在连接器名称字段中，输入连接器的名称。
在连接器插件列表中，选择“MirrorMaker 2.0 Checkpoint”。
对于源集群，选择一个选项：
- Managed Service for Apache Kafka 集群：在 Kafka 集群列表中，选择 Managed Service for Apache Kafka 集群。
- 自行管理的 Kafka 集群或外部 Kafka 集群：在集群引导服务器字段中，输入 Kafka 集群的引导地址，格式为
HOSTNAME:PORT_NUMBER。
可选：在配置框中，添加配置属性或修改默认属性。如需了解详情，请参阅配置连接器。
在群组名称或群组正则表达式字段中，输入要复制的使用方群组的英文逗号分隔列表。您可以对群组名称使用正则表达式。
选择任务重启政策。如需了解详情，请参阅任务重启政策。
点击创建。
gcloud
运行 gcloud
managed-kafka connectors create 命令：
gcloud managed-kafka connectors create CONNECTOR_ID \
--connect-cluster=CONNECT_CLUSTER_ID \
--location=LOCATION \
--config-file=CONFIG_FILE
替换以下内容：
CONNECTOR_ID：连接器的 ID 或名称。CONNECT_CLUSTER_ID：在其中创建连接器的 Connect 集群的 ID。LOCATION：Connect 集群的位置。CONFIG_FILE：本地 YAML 或 JSON 配置文件的路径。
以下是 MirrorMaker 2.0 Checkpoint 连接器的配置文件示例：
connector.class: "org.apache.kafka.connect.mirror.MirrorCheckpointConnector"
emit.checkpoints.interval.seconds: "CHECKPOINT_INTERVAL"
groups: "CONSUMER_GROUP_LIST"
groups.exclude: "console-consumer-.*,connect-.*,__.*"
source.cluster.alias: "source"
source.cluster.bootstrap.servers: "SOURCE_CLUSTER_BOOTSTRAP_ADDRESS"
sync.group.offsets.enabled: "true"
sync.group.offsets.interval.seconds: "OFFSETS_SYNC_INTERVAL"
target.cluster.alias: "target"
target.cluster.bootstrap.servers: "TARGET_CLUSTER_BOOTSTRAP_ADDRESS"
tasks.max: "3"
替换以下内容：
CHECKPOINT_INTERVAL：写入检查点的频率（以秒为单位）。CONSUMER_GROUP_LIST。偏移量被复制的使用方群组的逗号分隔列表。该列表可以包含正则表达式。SOURCE_CLUSTER_BOOTSTRAP_ADDRESS：源 Kafka 集群的引导地址。OFFSETS_SYNC_INTERVAL：将检查点复制到__consumer_offsets主题的频率（以秒为单位）。如需了解详情，请参阅检查点同步。TARGET_CLUSTER_BOOTSTRAP_ADDRESS：目标 Kafka 集群的引导地址。
配置连接器
本部分介绍了您可以在连接器上设置的一些配置属性。
如需查看此连接器特有的属性的完整列表，请参阅 Apache Kafka 文档中的 MirrorMaker 检查点配置。
复制的使用方群组
如需指定要复制的使用方群组，请设置 groups 配置属性。您可以指定单个名称或使用正则表达式。示例："group1,group2,customer-*"。默认值为 .*。
如需从复制中排除使用方群组，请设置 groups.exclude 属性。
检查点同步
以下设置控制连接器如何将检查点写入目标集群。
sync.group.offsets.enabled：是否将偏移量从source.checkpoints.internal复制到目标集群上的__consumer_offsets。默认值为false。如需使目标集群中的使用方在故障切换后无缝重启，请将此属性设置为true。emit.checkpoints.interval.seconds：将检查点写入内部检查点主题的频率。默认值为 60 秒。设置此属性时，请考虑在跟踪进度与最大限度减少写入检查点的开销之间取得平衡。10 秒是一个合理的值。sync.group.offsets.interval.seconds：将检查点从内部检查点主题复制到__consumer_offsets的频率。默认值为 60 秒。 此设置仅在sync.group.offsets.enabled为true时适用。