# Tutorial: Share Data Across Topics Using Cluster Linking for Confluent Platform This tutorial provides a practical first look at how to use Cluster Linking for a topic data sharing use case. ## What the tutorial covers By the end of this tutorial, you will have configured two clusters and successfully used Cluster Linking to create a mirror topic and share topic data across the clusters. You will also learn how to stop mirroring to make the topic writable, and verify that the two topics have diverged.  ## About prerequisites and command examples - These instructions assume you have a local installation of [Confluent Platform 7.0.0 or later](https://www.confluent.io/download/#confluent-platform), the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/installing.html), and Java 8, 11, or 17 (recommended). For details on Java requirements, see [Java](../../installation/system-requirements.md#sys-req-java) in System Requirements for Confluent Platform. If you are new to Confluent Platform, you may want to work through the [Quick Start for Confluent Platform](../../get-started/platform-quickstart.md#quickstart) first, and then return to this tutorial. - The examples assume that your properties files are in the default locations on your Confluent Platform installation, except as otherwise noted. This should make it easier to copy/paste example commands directly into your terminal in most cases. - With Confluent Platform is installed, Confluent CLI commands themselves can be run from any directory (`kafka-topics`, `kafka-console-producer`, `kafka-console-consumer`), but for commands that access properties files in `$CONFLUENT_HOME` (`kafka-server-start`), the examples show running these from within that directory. A reference for these open source utilities is provided in [Kafka Command-Line Interface (CLI) Tools](/kafka/operations-tools/kafka-tools.html). A reference for Confluent premium command line tools and utilities is provided in [CLI Tools for Confluent Platform](/platform/current/installation/cli-reference.html). - Confluent CLI commands can specify the bootstrap server at the beginning or end of the command: `kafka-topics --list --bootstrap-server localhost:9092` is the same as `kafka-topics --bootstrap-server localhost:9092 --list`. In these tutorials, the target bootstrap server is specified at the end of commands. The rest of the tutorial refers to `$CONFLUENT_HOME` to indicate your Confluent Platform install directory. Set this as an environment variable, for example: ```bash export CONFLUENT_HOME=$HOME/confluent-8.3.1 PATH=$CONFLUENT_HOME/bin:$PATH ``` ### KRaft and ZooKeeper #### IMPORTANT As of Confluent Platform 8.0, ZooKeeper is no longer available for new deployments. Confluent recommends migrating to KRaft mode for new deployments. To learn more about running Kafka in KRaft mode, see [KRaft Overview for Confluent Platform](../../kafka-metadata/kraft.md#kraft-overview), [KRaft Configuration for Confluent Platform](../../kafka-metadata/config-kraft.md#configure-kraft), and the KRaft steps in the [Platform Quick Start](../../get-started/platform-quickstart.md#cp-quickstart-step-1). To learn about migrating from older versions, see [Migrate from ZooKeeper to KRaft on Confluent Platform](../../installation/migrate-zk-kraft.md#migrate-zk-kraft). This tutorial provides examples for KRaft mode only. Earlier versions of this documentation provide examples for both KRaft and ZooKeeper. For KRaft, the examples show a *combined mode* configuration, where for each cluster the broker and controller run on the same server. Currently, combined mode is for local experimentation only and is not supported by Confluent. It is shown here to simplify the tutorial. If you want to run controllers and brokers on separate servers, use KRaft in isolated mode. To learn more, see [KRaft Overview for Confluent Platform](../../kafka-metadata/kraft.md#kraft-overview) the Kafka section under [Configure Confluent Platform for production](../../installation/installing_cp/zip-tar.md#config-cp-for-production). ## Ports and configuration mapping The example deployment in this tutorial uses the following port and feature configurations, and assumes that services will run on `localhost`. | | Source | Destination | |-------------------|------------------------------------------|--------------------------------------------| | Kafka brokers | 9092 (location of original `demo` topic) | 9093 (location of `demo` mirror topic) | | KRaft controllers | 9094 | 9095 | | HTTP listeners | 8090 | 8091 | | Cluster link | | Cluster link is enabled on the destination | ## Configure Kafka brokers and controllers Configure the following files to set up the source and destination clusters. You can copy and modify the existing properties files to use as a starting point. ### Configure the source cluster The source cluster configurations (ports, data directories, and so on) are largely based on the defaults for the template properties files, with only a few changes. The source cluster must be Confluent Platform 5.4 or later (Apache Kafka® 2.4). 1. Change directories to the location where Confluent Platform is installed: ```bash cd $CONFLUENT_HOME ``` 2. Create a directory to be used for all of your example files: ```bash mkdir my-examples ``` 3. Copy `etc/kafka/server.properties` into the examples directory and rename it to match its purpose: ```bash cp etc/kafka/server.properties my-examples/server-src.properties ``` 4. Update the port numbers. ```bash sed -i '' -e "s/9093/9094/g" my-examples/server-src.properties ``` 5. Append the following line at the end of `server-src.properties` to set the HTTP listener on the source cluster to `8090`: ```bash echo "confluent.http.server.listeners=http://localhost:8090" >> my-examples/server-src.properties ``` ### Configure the destination cluster While you configured the source cluster to run on default ports, you must set up the destination cluster on different ports to avoid collisions, as summarized in [ports mapping](../replicator/replicator-quickstart.md#replicator-tutorial-ports-map). Copy the configuration files to your examples directory and modify them as shown below. The destination cluster must be Confluent Platform 7.0.0 or later. 1. Change directories to the location where Confluent Platform is installed: ```bash cd $CONFLUENT_HOME ``` 2. Copy `my-examples/server-src.properties` (the file you just created) in the examples directory and rename the copy to match its purpose: ```bash cp my-examples/server-src.properties my-examples/server-dst.properties ``` 3. Update the port numbers. ```bash sed -i '' -e "s/9092/9093/g" my-examples/server-dst.properties ``` ```bash sed -i '' -e "s/9094/9095/g" my-examples/server-dst.properties ``` ```bash sed -i '' -e "s/8090/8091/g" my-examples/server-dst.properties ``` 4. Update data directories. ```bash sed -i '' -e "s/kraft-combined-logs/kraft-combined-logs-dst/g" my-examples/server-dst.properties ``` 5. Add the following line to `server-dst.properties` to set the replication factor to `1` for the cluster link topic metadata. (This configuration is set to simplify this demo. The recommended replication factor in production is `3`, which is the default.) ```bash echo "confluent.cluster.link.metadata.topic.replication.factor=1" >> my-examples/server-dst.properties ``` 6. Enable Cluster Linking on the destination. ```bash sed -i '' -e "s/confluent.cluster.link.enable=false/confluent.cluster.link.enable=true/g" my-examples/server-dst.properties ``` ## Start the source cluster Run the following commands and start the source cluster from `$CONFLUENT_HOME`. 1. Generate a `random-uuid` using the kafka-storage tool: ```bash KAFKA_CLUSTER_ID="$(bin/kafka-storage random-uuid)" ``` 2. Format the log directories for this server: ```bash ./bin/kafka-storage format -t $KAFKA_CLUSTER_ID -c my-examples/server-src.properties --standalone ``` 3. Start the server for the source cluster by running this command in its own terminal. ```bash ./bin/kafka-server-start my-examples/server-src.properties ``` ## Start the destination cluster Run the following commands and start the destination cluster from `$CONFLUENT_HOME`. 1. Generate a `random-uuid` using the kafka-storage tool: ```bash KAFKA_CLUSTER_ID="$(bin/kafka-storage random-uuid)" ``` 2. Format the log directories for this server: ```bash ./bin/kafka-storage format -t $KAFKA_CLUSTER_ID -c my-examples/server-dst.properties --standalone ``` 3. Start the server for the destination cluster by running this command in its own terminal. ```bash ./bin/kafka-server-start my-examples/server-dst.properties ``` ## Populate the source cluster 1. Create a topic on the source cluster. ```bash kafka-topics --create --topic demo --bootstrap-server localhost:9092 ``` You should get confirmation that the topic was successfully created. ```bash Created topic demo. ``` 2. Run the producer in a new command window to send some messages to the `demo` topic on the source cluster, and fill it with data. ```bash kafka-console-producer --topic demo --bootstrap-server localhost:9092 ``` When the producer starts, you will get a prompt `>`. Type some messages at the prompt. Press return after each message. The producer window will look like this: ```bash >first >second >third ``` 3. Consume from the topic on the source cluster. In a new terminal, run a consumer to consume messages from the `demo` topic. ```bash kafka-console-consumer --topic demo --from-beginning --bootstrap-server localhost:9092 ``` If the topic successfully consumes the messages, your output will be: ```bash first second third ``` ## Create the cluster link and the mirror topic 1. Create the cluster link on the destination cluster. ```bash kafka-cluster-links --bootstrap-server localhost:9093 \ --create --link demo-link --config bootstrap.servers=localhost:9092 ``` The confirmation message looks like this: ```bash Cluster link 'demo-link' creation successfully completed. ``` 2. List cluster links (you should see the one you just created). ```bash kafka-cluster-links --list --bootstrap-server localhost:9093 ``` Your output should resemble: ```bash Link name: 'demo-link', link ID: '', remote cluster ID: '', local cluster ID: '', remote cluster available: 'true', link state: 'ACTIVE ``` 3. Initialize the mirror topic. The following command establishes a mirror topic of the source `demo` topic, using the cluster link `demo-link`. ```bash kafka-mirrors --create --mirror-topic demo --link demo-link \ --bootstrap-server localhost:9093 ``` The confirmation message looks like this: ```bash Created topic demo. ``` 4. Consume from the mirror topic on the destination cluster to verify it. In a new terminal, run a consumer to consume messages from the mirror topic. ```bash kafka-console-consumer --topic demo --from-beginning --bootstrap-server localhost:9093 ``` Your output should be: ```bash first second third ``` ## Check the replica status on the destination 1. Run the following command to monitor the replicas on the destination cluster. ```bash kafka-replica-status --topics demo --include-linked --bootstrap-server localhost:9093 ``` Your output should resemble the following example. ```bash Topic Partition Replica ClusterLink IsLeader IsObserver IsIsrEligible IsInIsr IsCaughtUp LastCaughtUpLagMs LastFetchLagMs LogStartOffset LogEndOffset demo 0 0 - true false true true true 0 0 0 3 demo 0 0 demo-link true false true true true 2 2 0 3 ``` The output shows both the source and destination’s replicas as distinguished by the `ClusterLink` field, where `-` means local and the link name (`demo-link` in this case) indicates the replica(s) over the cluster link. 2. Run the following command to populate values for “MirrorState”, “MirrorLastFetchTimeMs”, “MirrorLastFetchHighWatermark”: ```bash kafka-replica-status --topics demo --include-mirror --bootstrap-server localhost:9093 ``` Your output should resemble the following: ```bash Topic Partition Replica IsLeader IsObserver IsIsrEligible IsInIsr IsCaughtUp LastCaughtUpLagMs LastFetchLagMs LogStartOffset LogEndOffset LeaderEpoch MirrorState MirrorLastFetchTimeMs MirrorLastFetchHighWatermark demo 0 1 true false true true true 0 0 0 3 0 ACTIVE -1 -1 ``` ## Verify that the mirror topic is read-only Attempt to produce messages to the mirror topic on the destination as a way of verifying that the mirror topic is read-only. 1. Start a producer, targeting the mirror topic. ```bash kafka-console-producer --topic demo --bootstrap-server localhost:9093 ``` 2. Type a message at the `>` prompt, and press return. You should get an error similar to the following example because the mirror topic is not writable. ```bash >hi 2020-08-13 16:55:53,571] ERROR Error when sending message to topic demo with key: null, value: 2 bytes with error: (org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) ``` ## Change the Source Topic Configuration 1. Describe the mirror topic to verify the destination and the mirror topic configs. ```bash kafka-configs --describe --topic demo --bootstrap-server localhost:9093 ``` The output will show parameters that are set for strictly mirrored configs. ```bash Dynamic configs for topic demo are: compression.type=producer sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:compression.type=producer, DEFAULT_CONFIG:compression.type=producer} cleanup.policy=delete sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:cleanup.policy=delete, DEFAULT_CONFIG:log.cleanup.policy=delete} segment.bytes=1073741824 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:segment.bytes=1073741824, STATIC_BROKER_CONFIG:log.segment.bytes=1073741824, DEFAULT_CONFIG:log.segment.bytes=1073741824} max.message.bytes=1048588 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:max.message.bytes=1048588, DEFAULT_CONFIG:message.max.bytes=1048588} ``` 2. Modify the source set `retention.ms`. This is one of the configurations that will mirror only if it is explicitly set on the source. (Note that `retention.ms` was not included in the output for the previous step.) ```bash kafka-configs --alter --topic demo --add-config retention.ms=123456890 --bootstrap-server localhost:9092 ``` The confirmation message looks like this: ```bash Completed updating config for topic demo. ``` 3. Rerun the `kafka-configs` command to verify that `retention.ms` is now on the mirror topic. ```bash kafka-configs --describe --topic demo --bootstrap-server localhost:9093 ``` The dynamic configurations now include `retention.ms`. ```bash Dynamic configs for topic demo are: compression.type=producer sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:compression.type=producer, DEFAULT_CONFIG:compression.type=producer} cleanup.policy=delete sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:cleanup.policy=delete, DEFAULT_CONFIG:log.cleanup.policy=delete} segment.bytes=1073741824 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:segment.bytes=1073741824, STATIC_BROKER_CONFIG:log.segment.bytes=1073741824, DEFAULT_CONFIG:log.segment.bytes=1073741824} retention.ms=123456890 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:retention.ms=123456890} max.message.bytes=1048588 sensitive=false synonyms={DYNAMIC_TOPIC_CONFIG:max.message.bytes=1048588, DEFAULT_CONFIG:message.max.bytes=1048588} ``` ## Change the source topic’s partitions 1. Alter the number of partitions on the source topic. ```bash kafka-topics --alter --topic demo --partitions 8 --bootstrap-server localhost:9092 ``` 2. Verify the change on the source topic. ```bash kafka-topics --describe --topic demo --bootstrap-server localhost:9092 ``` The output will be: ```bash Topic: demo PartitionCount: 8 ReplicationFactor: 1 Configs: segment.bytes=1073741824,retention.ms=123456890 Topic: demo Partition: 0 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: demo Partition: 1 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: demo Partition: 2 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: demo Partition: 3 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: demo Partition: 4 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: demo Partition: 5 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: demo Partition: 6 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: demo Partition: 7 Leader: 0 Replicas: 0 Isr: 0 Offline: ``` 3. Verify the change on the mirror topic. ```bash kafka-topics --describe --topic demo --bootstrap-server localhost:9093 ``` The output of this same command for the mirror topic should match the original topic exactly. ```bash Topic: demo PartitionCount: 8 ReplicationFactor: 1 Configs: segment.bytes=1073741824,retention.ms=123456890 Topic: demo Partition: 0 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: demo Partition: 1 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: demo Partition: 2 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: demo Partition: 3 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: demo Partition: 4 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: demo Partition: 5 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: demo Partition: 6 Leader: 0 Replicas: 0 Isr: 0 Offline: Topic: demo Partition: 7 Leader: 0 Replicas: 0 Isr: 0 Offline: ``` ## List mirror topics To list mirror topics run `kafka-cluster-links --list` with the `--include-topics` option. ```bash kafka-cluster-links --list --link demo-link --include-topics --bootstrap-server localhost:9093 ``` Your output should resemble the following. ```bash Link name: 'demo-link', link ID: '123-some-link-id', remote cluster ID: '123-some-cluster-id', local cluster ID: '456-some-other-cluster-id', remote cluster available: 'true', topics: [demo] ``` ## Cut over the mirror topic to make it writable 1. Stop the mirror topic for the `demo` topic on the destination. ```bash kafka-mirrors --promote --topics demo --bootstrap-server localhost:9093 ``` You will get the following confirmation. ```none Calculating max offset and ms lag for mirror topics: [demo] Finished calculating max offset lag and max lag ms for mirror topics: [demo] Request for stopping topic inventory.arrivals's mirror was successfully scheduled. Use the describe command with the --pending-stopped-only option to monitor progress. ``` 2. Verify that the mirror topic has stopped mirroring. ```bash kafka-mirrors --describe --topics demo --pending-stopped-only --bootstrap-server localhost:9093 ``` You will get the following confirmation that the topic has entered the `STOPPED` state. ```bash Topic: demo LinkName: demo-link LinkId: 123-some-link-id MirrorTopic: demo State: STOPPED StateTime: 2021-11-15 14:06:26 ``` ## Produce to both topics to verify divergence 1. Run a producer to send messages to the original topic on the source. (Or switch back to this command window if you still have this producer running.) ```bash kafka-console-producer --topic demo --bootstrap-server localhost:9092 ``` Type the message `old` at the prompt. ```bash > old ``` 2. Run a producer to send messages to the topic on the destination, which is no longer a mirror topic. ```bash kafka-console-producer --topic demo --bootstrap-server localhost:9093 ``` Type the message `new` at the prompt. ```bash > new ``` 3. Run consumers to read from both topics. Your output will show they have diverged. - Run a consumer to read messages from the original topic on the source. (Or simply view the output on this window, if you still have this consumer running) ```bash kafka-console-consumer --topic demo --from-beginning --bootstrap-server localhost:9092 ``` The output on the source topic now includes the `old` message. ```bash first second third old ``` - Run a consumer to read messages from the topic on the destination. (Or simply view the output on this window, if you still have this consumer running.) ```bash kafka-console-consumer --topic demo --from-beginning --bootstrap-server localhost:9093 ``` The output on the destination topic includes the `new` message. ```bash first second third new ``` ## Delete the link 1. List cluster links. ```bash kafka-cluster-links --list --bootstrap-server localhost:9093 ``` The output should resemble the following example. ```bash Link name: 'demo-link', link ID: '123-some-link-id', cluster ID: '123-some-cluster-id' ``` 2. Delete the cluster link. ```bash kafka-cluster-links --bootstrap-server localhost:9093 --delete --link demo-link ``` If the command is successful, the output is: ```bash Cluster link 'demo-link' deletion successfully completed. ``` ## Teardown Run shutdown and cleanup tasks. 1. Stop the consumers and producers if they are still running, in reverse order in which you started them. 2. Stop the clusters: use Ctl-C in each command window to first stop the destination cluster, then stop the source cluster. 3. Delete the log directories from `/tmp`. This will clear out the metadata from your system and enable you to configure and run new local deployments with no collisions with the legacy metadata.