跨集群复制数据

背景信息

自Confluent Platform 7.5起，ZooKeeper已不再用于新部署。Confluent建议新部署采用KRaft模式。要了解有关在KRaft模式下运行Kafka的更多信息，请参见KRaft Overview和Platform Quick Start。

本文提供了KRaft模式和ZooKeeper模式的示例。

对于KRaft，本示例展示了组合模式配置，其中每个集群的服务端和控制器都运行在同一台服务器上。目前，组合模式并不打算在生产中使用，为了简化教程，在此进行了展示。如果您想在单独的服务器上运行控制器和服务端，请在隔离模式下使用KRaft。如需了解更多信息，请参见KRaft Overview和KRaft mode。

前提条件

• 安装了Confluent Platform 更多信息，请参见Confluent。

• 安装Java 8、11或17。更多信息，请参见安装JDK。

Kafka服务端和Confluent Platform组件的端口

启动目标集群

启动源集群

在新的终端窗口中配置并启动源集群。

创建Topic

打开一个新的命令窗口来运行Kafka命令。

1. 在原始集群中使用以下命令创建一个名为“test-topic”的主题。

   kafka-topics --create --topic test-topic --replication-factor 1 --partitions 1 --bootstrap-server localhost:9082

2. 验证Topic是否被创建。

   kafka-topics --list --bootstrap-server localhost:9082

   预期输出（_confluent开头的Topic是内部Topic）：

   __confluent.support.metrics
   _confluent-command
   test-topic

   当配置并运行Replicator时，“test-topic”将会被复制到目的地集群（在端口2181上）。

配置并运行Replicator

在$CONFLUENT_HOME/my-examples/中创建以下文件。

1. 在名为consumer.properties的文件中配置源集群。

   cp etc/kafka/consumer.properties my-examples/.

   编辑该文件，确保其中包含源集群的代理地址。默认的代理列表将与您之前启动的源集群相匹配。

   # Origin cluster connection configuration
   bootstrap.servers=localhost:9082

2. 在名为producer.properties的新文件中配置目标集群。

   cp etc/kafka/producer.properties my-examples/.

   编辑该文件，确保它包含目的地集群代理的地址。默认代理列表将与您之前启动的目的地集群相匹配。

   # Destination cluster connection configuration
   bootstrap.servers=localhost:9092

3. 在名为replication.properties的新文件中为Connect Worker定义Replicator配置。

   # Replication configuration
   topic.rename.format=${topic}.replica
   replication.factor=1
   config.storage.replication.factor=1
   offset.storage.replication.factor=1
   status.storage.replication.factor=1
   confluent.topic.replication.factor=1

   说明

   • 如果replication.properties中未定义端口，则此Worker将在默认端口8083上运行。

   • 使用复制因子属性（全部设置为1）是因为这些测试集群规模较小。在生产环境中，建议的最小集群大小为3。

启动replicator

执行下面的命令在自己的终端启动Replicator可执行程序。

./bin/replicator --cluster.id replicator --consumer.config my-examples/consumer.properties --producer.config my-examples/producer.properties --replication.config my-examples/replication.properties --whitelist 'test-topic'

• --cluster.id：用于确定这个可执行文件应该加入哪个Replicator集群的标识符。拥有相同cluster.id的多个Replicator可执行实例将一起工作。

• --consumer.config：源集群配置的路径。

• --producer.config：目标集群配置的路径。

• --replication.config：包含任何非连接特定配置的文件路径。命令行参数将覆盖这些配置。

• --whitelist：从源集群复制到目标集群的Topic列表。

验证跨集群的Topic复制

Replicator完成初始化后，会检查源集群是否有需要复制的主题。

执行以下命令，检查test-topic.replica是否存在。

./bin/kafka-topics --describe --topic test-topic.replica --bootstrap-server localhost:9092

预期输出：

./bin/kafka-topics --describe --topic test-topic.replica --bootstrap-server localhost:9092
Topic: test-topic.replica    PartitionCount: 1       ReplicationFactor: 1    Configs: message.timestamp.type=CreateTime,segment.bytes=1073741824
      Topic: test-topic.replica      Partition: 0    Leader: 0       Replicas: 0     Isr: 0  Offline: 0

您还可以列出并描述目标集群上的主题。例如 test-topic.replica。

./bin/kafka-topics --list --bootstrap-server localhost:9092

在您在源集群中创建主题之后，您可以开始使用Kafka生产者向原始集群中的test-topic发送数据。然后，您可以通过从目标集群中的test-topic.replica消费来确认数据已被复制。例如，要使用Kafka的控制台生产者发送一系列数字，请在新的终端窗口中运行以下命令：

seq 10000 | ./bin/kafka-console-producer --topic test-topic --broker-list localhost:9082

你可以使用控制台消费者在它自己的终端窗口中确认目的地集群的传送情况：

./bin/kafka-console-consumer --from-beginning --topic test-topic.replica --bootstrap-server localhost:9092

如果消费者输出中出现了从1到10,000的数字，这表明您已经成功创建了多集群复制。

使用Control Center监控replicator

1. 停止源集群和目标集群上的Replicator和代理服务器，然后停止ZooKeeper实例。

2. 激活Replicator的监控扩展，更多信息，请参见Replicator Monitoring Extension。

   • 将replicator-rest-extension-.jar的完整路径添加到你的CLASSPATH中。

   • 将rest.extension.classes=io.confluent.connect.replicator.monitoring.ReplicatorMonitoringExtension添加到my-examples/replication.properties中。

3. 在目标集群和源集群的Kafka配置文件中取消注释或添加以下行，分别是my-examples/server_destination.properties和my-examples/server_origin.properties。confluent.metrics.reporter.bootstrap.servers的配置必须在两个文件中都指向端口9092的localhost，您需要编辑这两个端口号中的一个或两个。

   # 表明你的部署处于开发模式，使用的复制因子为1。
   confluent.metrics.reporter.topic.replicas=1
   # 在Control Center上启用了指标报告，并提供了访问Confluent内部Topic的权限，该Topic收集并存储监控数据。
   metric.reporters=io.confluent.metrics.reporter.ConfluentMetricsReporter
   confluent.metrics.reporter.bootstrap.servers=localhost:9092

4. 编辑my-examples/producer.properties，为生产者添加监控拦截器。

   # Monitoring interceptor for producer
   interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor

5. 编辑my-examples/consumer.properties，为消费者添加监控拦截器。

   # Monitoring interceptor for consumer
   interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor

6. 编辑etc/confluent-control-center/control-center-dev.properties，添加以下两行，指定Control Center的源引导服务器和目标引导服务器。

   # multi-cluster monitoring
   confluent.controlcenter.kafka.origin.bootstrap.servers=localhost:9082
   confluent.controlcenter.kafka.destination.bootstrap.servers=localhost:9092

   说明

   • Control Center 需要 Connect REST 端点的主机和端口，才能知道在哪里查找 Replicator 监控指标。在本示例使用的配置文件（control-center-dev.properties）中，已为您配置了默认端口，因此开箱即用：

     # 以逗号分隔的连接主机名列表。
     confluent.controlcenter.connect.cluster=http://localhost:8083

   • 生产就绪配置文件（control-center-production.properties）的默认值已被注释掉。如果使用该文件、拥有多个连接程序或希望以不同方式配置连接集群，则必须指定连接端点，方法是取消注释默认值或为自己的连接集群指定主机。要了解更多信息，请参见Control Center Configuration Reference。

   • 如果在部署中同时运行Replicator和Connect集群。必须分别指定：

     • Connect

       cluster: confluent.controlcenter.connect.<connect-cluster-name>.cluster=http://connect-host-1:8083

     • Replicator

       confluent.controlcenter.connect.<replicator-name>.cluster=http://replicator-host:8083

7. 重新启动目标集群和原集群上的ZooKeeper实例。

   ./bin/zookeeper-server-start etc/kafka/zookeeper.properties
   ./bin/zookeeper-server-start my-examples/zookeeper_origin.properties

8. 重新启动目标集群和原集群上的Broker。

   ./bin/kafka-server-start my-examples/server_destination.properties
   ./bin/kafka-server-start my-examples/server_origin.properties

9. 重新启动Replicator和Connect worker。

   ./bin/replicator --cluster.id replicator --consumer.config my-examples/consumer.properties --producer.config my-examples/producer.properties --replication.config my-examples/replication.properties --whitelist 'test-topic'

10. 使用以下命令启动Control Center。

    ./bin/control-center-start etc/confluent-control-center/control-center-dev.properties

11. 在您的浏览器中打开Control Center，根据配置，集群会以自动生成的名称呈现在Control Center上。

12. （可选）在Control Center上，编辑集群名称以适应您的使用情况。

13. 在Control Center选择目标集群，单击左侧导航栏的Replicators，然后使用Control Center监控replication性能，并深入查看源Topic和复制Topic。

    要在Control Center查看向源Topic和复制Topic生成的消息，请在命令窗口中尝试使用kafka-consumer-perf-test，以自动向test-topic生成测试数据。

    kafka-producer-perf-test \
       --producer-props bootstrap.servers=localhost:9082 \
       --topic test-topic \
       --record-size 1000 \
       --throughput 1000 \
       --num-records 3600000

    预期输出：

    4999 records sent, 999.8 records/sec (0.95 MB/sec), 1.1 ms avg latency, 240.0 ms max latency.
    5003 records sent, 1000.2 records/sec (0.95 MB/sec), 0.5 ms avg latency, 4.0 ms max latency.
    5003 records sent, 1000.2 records/sec (0.95 MB/sec), 0.6 ms avg latency, 5.0 ms max latency.
    5001 records sent, 1000.2 records/sec (0.95 MB/sec), 0.3 ms avg latency, 3.0 ms max latency.
    5001 records sent, 1000.0 records/sec (0.95 MB/sec), 0.3 ms avg latency, 4.0 ms max latency.
    5000 records sent, 1000.0 records/sec (0.95 MB/sec), 0.8 ms avg latency, 24.0 ms max latency.
    5001 records sent, 1000.2 records/sec (0.95 MB/sec), 0.6 ms avg latency, 3.0 ms max latency.
    ...

    您可以使用kafka-console-consumer从命令行读取这些信息，以验证副本Topic是否接收到了这些信息。

    ./bin/kafka-console-consumer --from-beginning --topic test-topic.replica --bootstrap-server localhost:9092

    您也可以在Control Center上进行验证。

要了解在Control Center监控复制器的更多信息，请参见Replicators” in Control Center User Guide。

完成本教程的实验后，请务必按以下步骤进行清理：

• 在每个命令窗口中使用Ctrl-C停止任何生产者和消费者。

• 在每个命令窗口中使用Ctrl-C按启动服务的相反顺序停止每个服务（先停止Control Center，然后停止Replicator、Kafka Brokers，最后停止ZooKeepers）。