ApsaraMQ for Kafka:Replicate data across Kafka clusters with Cluster Linking

Source cluster configuration

Create a file named /tmp/source.config with the following content. This configuration enables automatic mirror topic creation, consumer offset synchronization, and ACL synchronization.

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 cluster configuration

Create a file named /tmp/destination.config with the following content:

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

Create a test topic

Create a single-partition topic named test-topic. A single partition makes it easier to observe message replication order.

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

To verify the topic was created, list and describe it:

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

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

Produce test messages

Send five numbered messages to test-topic:

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

Consume and verify messages

Consume messages from the beginning and assign the consumer to a group named test-group:

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

Expected output:

1
2
3
4
5

To inspect the consumer group offset details:

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

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

Add an ACL entry

Grant the READ permission on test-topic to user test-user:

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

To verify the ACL was created:

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

Topic filter

Create /tmp/topic_filter.json to select test-topic for replication:

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

Consumer group filter

Create /tmp/group.json to select test-group for offset synchronization:

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

ACL filter

Create /tmp/acl.json to synchronize all ACL entries:

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

Promote the mirror topic

After synchronization completes, promote the mirror topic to make it a regular, writable topic. Once promoted, the topic stops synchronizing with the source cluster.

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

Expected output:

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.

Check topics, consumer groups, and ACLs

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

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

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

Confirm that test-topic, test-group, and the ACL entry for test-user appear in the output.

Test produce and consume on the destination cluster

Verify that the promoted topic accepts new messages and that consumption works correctly:

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

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

List all cluster links

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

View cluster link details

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

Promote a mirror topic

Convert a mirror topic into a regular, writable topic. After promotion, the topic stops receiving updates from the source cluster.

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

Expected output:

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.

Delete a cluster link

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

Expected output:

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