使用 Cluster Linking 跨 Kafka 集群复制数据

Source 集群配置

创建连接 Source 集群的配置文件 /tmp/source.config，内容如下。该配置开启自动创建 Mirror Topic、同步 Consumer Group 位点和同步 ACL 条目。

security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="<username>" password="<password>";
bootstrap.servers=<source-cluster-address:port>
auto.create.mirror.topics.enable=true
consumer.offset.sync.enable=true
acl.sync.enable=true

Destination 集群配置

创建连接 Destination 集群的配置文件 /tmp/destination.config，内容如下：

security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="<username>" password="<password>";

创建测试 Topic

• 执行以下命令，在 Source 集群上创建名为 test-topic 的单分区 Topic，用于后续镜像复制。使用单分区便于观察复制消息的顺序。

kafka-topics --create --topic test-topic --partitions 1 \
--bootstrap-server <source-cluster-address:port> \
--command-config /tmp/source.config

使用 list topic 和 describe topic 命令查看 Topic 详情，验证 Topic 创建成功。

#list topic
kafka-topics --list --bootstrap-server <source-cluster-address:port> \
--command-config /tmp/source.config

#describe topic
kafka-topics --describe --topic test-topic \
--bootstrap-server <source-cluster-address:port> \
--command-config /tmp/source.config

生产测试消息

• 执行以下命令，向 Source 集群上的 test-topic 发送消息。

seq 1 5 | kafka-console-producer --topic test-topic \
--bootstrap-server <source-cluster-address:port> \
--producer.config /tmp/source.config

消费并验证消息

• 从开头消费 Source 集群中 test-topic 上的数据，并将消费者指定到名为 test-group 的 Consumer Group。

#consume
kafka-console-consumer --topic test-topic --from-beginning \
--bootstrap-server <source-cluster-address:port> --group test-group \
--consumer.config /tmp/source.config

#list consumer groups
kafka-consumer-groups --bootstrap-server <source-cluster-address:port> --list \
--command-config /tmp/source.config

#describe offsets of consumer groups
kafka-consumer-groups --bootstrap-server <source-cluster-address:port> \
--group test-group --describe --offsets \
--command-config /tmp/source.config

成功消费消息后，预期输出如下：

1

2

3

4

5

添加 ACL 条目

• 新增 ACL 条目，授予用户 test-user 对 test-topic 的读取（READ）权限。

#add READ ACL for test-user
kafka-acls --bootstrap-server <source-cluster-address:port> \
--command-config /tmp/source.config  --add --allow-principal User:test-user \
--operation READ --topic test-topic

#list
kafka-acls --list --bootstrap-server <source-cluster-address:port> \
--command-config /tmp/source.config

Topic 过滤器

• 创建用于筛选迁移 Topic 的配置文件 /tmp/topic_filter.json，选择复制 test-topic。

{
  "topicFilters": [
    {
      "name": "test-topic",
      "patternType": "LITERAL",
      "filterType": "INCLUDE"
    }
  ]
}

Consumer Group 过滤器

• 创建用于筛选迁移 Consumer Group 的配置文件 /tmp/group.json，选择同步 test-group 的位点。

{
  "groupFilters": [
    {
      "name": "test-group",
      "patternType": "LITERAL",
      "filterType": "INCLUDE"
    }
  ]
}

ACL 过滤器

• 创建用于筛选迁移 ACL 条目的配置文件 /tmp/acl.json，同步所有 ACL 条目。

{
  "aclFilters": [
    {
      "resourceFilter": {
        "resourceType": "any",
        "patternType": "any"
      },
      "accessFilter": {
        "operation": "any",
        "permissionType": "any"
      }
    }
  ]
}

创建集群链接并同步数据

• 执行以下命令，在 Destination 集群上创建名为 test-cluster-link 的集群链接，并复制 Topic、Consumer Group 和 ACL 条目。该命令应用上一步创建的 Topic、Consumer Group 和 ACL 过滤器。

kafka-cluster-links --create --link test-cluster-link \
  --config-file /tmp/source.config \
  --topic-filters-json-file /tmp/topic_filter.json \
  --consumer-group-filters-json-file /tmp/group.json \
  --acl-filters-json-file /tmp/acl.json \
  --bootstrap-server <destination-cluster-address:port> \
  --command-config /tmp/destination.config

集群链接创建完成后，Destination 集群开始从 Source 集群复制匹配的 Topic、Consumer Group 位点和 ACL 条目。

提升 Mirror Topic

• 数据同步完成后，执行以下命令提升 Mirror Topic，将其转换为可读写的普通 Topic。提升完成后，该 Topic 不再从源 Topic 同步消息。

kafka-mirrors --promote --topics test-topic \
--bootstrap-server <destination-cluster-address:port> \
--command-config /tmp/destination.config

预期输出：

Calculating max offset and ms lag for mirror topics: [test-topic]
Finished calculating max offset lag and max lag ms for mirror topics: [test-topic]
Request for stopping topic test-topic's mirror was successfully scheduled. Please use the describe command with the --pending-stopped-only option to monitor progress.

检查 Topic、Consumer Group 和 ACL

• 执行以下命令，查看 Destination 集群的 Topic、Consumer Group 和 ACL 条目是否已同步。

#list topic
kafka-topics --list --bootstrap-server <destination-cluster-address:port> \
--command-config /tmp/destination.config

#list consumer group
kafka-consumer-groups --bootstrap-server <destination-cluster-address:port> \
--list --command-config /tmp/destination.config

#list acl
kafka-acls --list --bootstrap-server <destination-cluster-address:port> \
--command-config /tmp/destination.config

确认输出中包含 test-topic、test-group 以及 test-user 的 ACL 条目。

验证生产与消费

• 验证提升后的 Topic 可接收新消息，且消费正常。

#produce
kafka-console-producer --topic test-topic \
--bootstrap-server <destination-cluster-address:port> \
--producer.config /tmp/destination.config

#consume
kafka-console-consumer --topic test-topic \
--bootstrap-server <destination-cluster-address:port> \
--consumer.config /tmp/destination.config

查看集群链接列表

• 执行以下命令，查看集群链接列表。

kafka-cluster-links --bootstrap-server <destination-cluster-address:port> \
 --list --command-config /tmp/destination.config

查看集群链接详情

• 执行以下命令，查看集群链接详情。

kafka-configs --describe --cluster-link test-cluster-link \
--bootstrap-server <destination-cluster-address:port> \
--command-config /tmp/destination.config

删除集群链接

• 执行以下命令，删除集群链接。

kafka-cluster-links --delete --link test-cluster-link \
  --bootstrap-server <destination-cluster-address:port> \
  --command-config /tmp/destination.config

预期输出：

Cluster link 'test-cluster-link' deletion successfully completed.