创建 MirrorMaker 2.0 检查点连接器

MirrorMaker 2.0 检查点连接器会将某个 Kafka 集群的使用方偏移值复制到另一个 Kafka 集群。Kafka 使用方使用使用方偏移量来记录分区中最后一条成功使用的消息。通过复制偏移量,使用方可以切换到目标集群,并从同一点继续处理。

您可以使用此连接器执行以下操作:

  • 帮助确保在从源集群切换到目标集群期间尽可能缩短停机时间。

  • 通过在集群之间提供一致的使用方状态,实现无缝故障切换。

  • 在将数据移至目标集群时,保留使用方进度。

Checkpoint 连接器的工作原理

Checkpoint 连接器可与 MirrorMaker 2.0 Source 连接器搭配使用。Checkpoint 连接器会按如下方式复制偏移量:

  1. 对于每个复制的分区,MirrorMaker 2.0 源连接器都会将复制的记录写入目标集群。

  2. 写入目标集群的记录通常具有与相应源记录不同的偏移量。为了实现从源偏移量到目标偏移量的转换,MirrorMaker 2.0 源连接器会将偏移量对写入源集群上名为 mm2-offset-syncs.target.internal 的主题。这些偏移对会将来源偏移映射到相应的目标偏移。

  3. 使用方从来源集群读取记录。为了记录进度,消费者会定期提交每个分区中最新处理的消息。Kafka 将提交的偏移量(也称为检查点)存储在名为 __consumer_offsets 的主题中。检查点记录包含消费者的消费者群组 ID。

  4. MirrorMaker 2.0 检查点连接器会将检查点记录复制到目标集群中名为 source.checkpoints.internal 的主题。

  5. 如果为 sync.group.offsets.enabled=true,则检查点连接器会定期将检查点从 source.checkpoints.internal 复制到目标集群上的 __consumer_offsets。这样,消费者便可以在故障切换后从最新提交的偏移量重新开始。如需恢复处理,消费者在故障切换后使用相同的消费者群组 ID。

    sync.group.offsets.interval.seconds 配置设置用于控制连接器将检查点复制到 __consumer_offsets 的频率。

创建 MirrorMaker 2.0 检查点连接器

控制台

  1. 在 Google Cloud 控制台中,前往连接集群页面。

    前往“连接集群”

  2. 点击要在其中创建连接器的 Connect 集群。

    系统会显示连接集群详情页面。

  3. 点击创建连接器。

  4. 在连接器名称字段中,输入连接器的名称。

  5. 在连接器插件列表中,选择“MirrorMaker 2.0 Checkpoint”。

  6. 对于源集群,选择一个选项:

    • Managed Service for Apache Kafka 集群:在 Kafka 集群列表中,选择 Managed Service for Apache Kafka 集群。
    • 自行管理的 Kafka 集群或外部 Kafka 集群:在集群引导服务器字段中,输入 Kafka 集群的引导地址,格式为 HOSTNAME:PORT_NUMBER。
  7. 可选:在配置框中,添加配置属性或修改默认属性。如需了解详情,请参阅配置连接器。

  8. 在群组名称或群组正则表达式字段中,输入要复制的使用方群组的英文逗号分隔列表。您可以对群组名称使用正则表达式。

  9. 选择任务重启政策。如需了解详情,请参阅任务重启政策。

  10. 点击创建。

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 时适用。

后续步骤

Apache Kafka® 是 Apache Software Foundation 或其关联公司在美国和/或其他国家/地区的注册商标。