ApsaraMQ for Kafka:Replicate data between Apache Kafka clusters with Replicator

Consumer properties (origin cluster)

Copy the default consumer configuration and update it to point to the origin cluster:

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

Verify that the file contains:

# Origin cluster connection
bootstrap.servers=localhost:9082

Producer properties (destination cluster)

Copy the default producer configuration and update it to point to the destination cluster:

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

Verify that the file contains:

# Destination cluster connection
bootstrap.servers=localhost:9092

Replication properties

Create my-examples/replication.properties with the following content:

# 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

Start Replicator

Start the Replicator executable in its own terminal:

./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'

Log messages about starting the source task and creating the replicated topic confirm that Replicator is running. For the full list of command-line options, see Command line parameters of Replicator executable.

Set up monitoring

1. Stop Replicator and brokers on both clusters, then stop any ZooKeeper instances.

2. Activate the Replicator monitoring extension (see Replicator monitoring extension):

   • Add the full path of replicator-rest-extension-<version>.jar to your CLASSPATH.

   • Add the following line to my-examples/replication.properties:

     rest.extension.classes=io.confluent.connect.replicator.monitoring.ReplicatorMonitoringExtension

3. Add or uncomment the following lines in both my-examples/server_destination.properties and my-examples/server_origin.properties. Set confluent.metrics.reporter.bootstrap.servers to localhost:9092 in both files:

      # Development mode -- replication factor of 1
      confluent.metrics.reporter.topic.replicas=1
      # Enable metrics reporting for Control Center
      metric.reporters=io.confluent.metrics.reporter.ConfluentMetricsReporter
      confluent.metrics.reporter.bootstrap.servers=localhost:9092

4. Add the monitoring interceptor to my-examples/producer.properties:

      interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringProducerInterceptor

5. Add the monitoring interceptor to my-examples/consumer.properties:

      interceptor.classes=io.confluent.monitoring.clients.interceptor.MonitoringConsumerInterceptor

6. Add the following lines to etc/confluent-control-center/control-center-dev.properties to specify the bootstrap servers for both clusters:

   Note

   • Control Center requires the Connect REST endpoint to find Replicator monitoring metrics. In control-center-dev.properties, the default is configured as:

     confluent.controlcenter.connect.cluster=http://localhost:8083

     Separate multiple host names with commas.

   • If you use control-center-production.properties, the default value is commented out. Uncomment it or specify your Connect endpoints. For details, see General settings.

   • If you run both Replicator and Connect clusters, specify them separately:

     • 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

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

Restart services and launch Control Center

1. Restart ZooKeeper instances (ZooKeeper mode only):

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

2. Restart the brokers on both clusters:

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

3. Restart 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'

4. Launch Control Center:

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

5. Open Control Center in your browser. The clusters appear with auto-generated names based on your configuration.

   

6. (Optional) Edit the cluster names to match your use case.

7. Select the destination cluster, then click Replicators in the left navigation pane to monitor replication throughput and drill down into source and replicated topics.

   

Generate test data for monitoring

To view message flow on Control Center, generate test data with kafka-producer-perf-test:

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

Expected output:

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.
...

Confirm that the replicated topic receives the messages:

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

For more about monitoring, see Manage Replicator Using Control Center for Confluent Platform.