# Command Reference for Cluster Linking on Confluent Platform Confluent Enterprise now includes a feature called Cluster Linking that syncs data and metadata from Confluent Platform, Confluent Cloud, or Kafka clusters to Confluent Platform and Confluent Cloud clusters. Cluster Linking geo-replicates data using two key abstractions: - a cluster link, which connects a “source cluster” and a “destination cluster” - mirror topics, which are topics on a cluster link’s destination cluster that are identical copies of topics on the source cluster. Cluster links and mirror topics can be created and managed using either a REST API in the [Kafka REST v3 Proxy](https://docs.confluent.io/platform/current/kafka-rest/api.html#clusterlinking), or using CLI commands in a terminal. This section describes the commands for creating and managing cluster links and mirror topics. ## CLI commands Configuration options for the cluster links are available as values for flags on the `kafka-cluster-links` command. Some of these options are shown below in the context of command examples. A full list is provided in reference form in [Cluster link configuration properties](configs.md#cluster-link-specific-configs). Use `kafka-cluster-links` to create and manage links across clusters. ### About bootstrap-server As with other Kafka commands in Confluent Platform, `--bootstrap-server` is a required flag for the `kafka-cluster-links` command. `--bootstrap-server` : (Required) The connection string for the broker(s) in a cluster in the form `host:port` (which can be a comma-separated list for multiple brokers). You must specify the destination cluster where you plan to create mirror topics. The destination cluster must be running Confluent Platform 7.0.0 or later, which is required to support Cluster Linking.
* Type: string * Default: empty string Use `--bootstrap-server` in all of the following implementations of `kafka-cluster-links`. For Cluster Linking, you should typically specify the `--bootstrap-server` on the destination cluster (with one exception, noted below). Here is a quick summary: - For `kafka-mirrors`, `--bootstrap-server` is always the destination cluster. - For `kafka-cluster-links`, `--bootstrap-server` is usually the destination cluster. The only exception for a source-initiated link (recommended for Cluster Linking from a Confluent Platform source to a Confluent Cloud destination). In that case, you would create a cluster link object twice and the second time requires executing `kafka-cluster-links` with the source cluster as the bootstrap server. For common use cases, see: - [Tutorial: Share Data Across Topics Using Cluster Linking for Confluent Platform](topic-data-sharing.md#tutorial-topic-data-sharing) - [Tutorial: Migrate Data with Cluster Linking on Confluent Platform](migrate-cp.md#cluster-linking-migrate-cp) For examples of use cases with source-initiated links, see: - [Tutorial: Link Confluent Platform and Confluent Cloud Clusters](hybrid-cp.md#cluster-link-hybrid-cp) ### Creating a cluster link #### Examples **Example Command** ```bash kafka-cluster-links --bootstrap-server localhost:9093 \ --create \ --link example-link \ --config-file example-link.config ``` **Example Output** ```bash Cluster link 'example-link' creation successfully completed. ``` #### Commands and flags To create a cluster link, use `kafka-cluster-links` along with [bootstrap-server](#bootstrap-cluster-links) and the following flags. `--link` : (Required) The name of the cluster link to create. Must be a unique cluster link name within the cluster.
* Type: string `--cluster-id` : (Required) The ID of the source cluster to link to. You can find a cluster’s ID with the CLI command `kafka-cluster cluster-id`.
* Type: string (Required) One of the following parameters must be provided (not both) to specify how the destination cluster should communicate with the source. The available configurations are those that would be used to configure a client, including the required `bootstrap.servers` and other necessary security and authorization properties. `--config` : Comma-separated configurations to be applied to the cluster link on creation of the form “key=value”. When you use this flag, the configurations are specified directly on the command line (as opposed to in a file, as described for the next flag). You can use square brackets to group values that contain commas. For a full list of available configurations, see [Cluster link configuration properties](configs.md#cluster-link-specific-configs).
* Type: string `--config-file` : Property file containing [configurations](configs.md#cluster-link-specific-configs) for the cluster link. This is the recommended way to specify cluster link configurations.
* Type: string For example, if you specify the following configuration for a secure cluster link in a file named `link-config.properties`: ```bash bootstrap.servers=example-1:9092,example-2:9092,example-3:9092 sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="example-user" password="example-password" security.protocol=SASL_SSL ssl.endpoint.identification.algorithm=https ``` Then, you can create the cluster link `example-link` with the following command: ```bash kafka-cluster-links --bootstrap-server localhost:9093 --create --link example-link --config-file link-config.properties --cluster-id pz-s7W72Sdm7A11wzku9gA ``` Optional configurations: `--command-config` : Property file containing configurations to be passed to the [AdminClient](../../installation/configuration/admin-configs.md#cp-config-admin). For example, with security credentials for authorization and authentication. `--consumer-group-filters-json` : JSON string to use for configuration of `consumer.offset.group.filters`. To learn more, see [Migrating consumer groups from source to destination cluster](#cluster-link-migrate-consumer-groups).
* Type: string `--consumer-group-filters-json-file` : Path to JSON file to use for configuration of `consumer.offset.group.filters`. To learn more, see [Migrating consumer groups from source to destination cluster](#cluster-link-migrate-consumer-groups).
* Type: string `--acl-filters-json-file` : Path to the ACL filters JSON file to use for configuration of `acl.filters`. To learn more, see [Use ACL filters to migrate or sync ACLs from Source to Destination cluster](security.md#cluster-link-acls-migrate).
* Type: string `--validate-only` : If provided, validates that the cluster link can be created as specified, but does not create it. `--exclude-validate-link` : If provided, creates the link without validating that the source cluster can be reached. This is helpful only if the source cluster is not yet running or reachable. If the source cluster is running and available, using this option is not recommended, as it skips helpful validations. `--topic-filters-json` : JSON string to use for configuration of `auto.create.mirror.topics.filters`. To learn more, see [Mirror Topics](mirror-topics-cp.md#mirror-topics-concepts). `--topic-filters-json-file` : Path to JSON file to use for configuration of `auto.create.mirror.topics.filters`. To learn more, see [Mirror Topics](mirror-topics-cp.md#mirror-topics-concepts). #### Properties on source and destination You must supply separate configurations for source and destination clusters. You can use property files for both as shown on the basic tutorial on [Share data across topics](topic-data-sharing.md#tutorial-topic-data-sharing) , or use a combination of command line and configuration files. Also, you have the option to specify another property file with `--command-config` for additional configurations, such as security credentials for one of the clusters to be passed to the [AdminClient](../../installation/configuration/admin-configs.md#cp-config-admin). For example, if you create a destination-initiated link, you can specify source cluster configurations in `--config-file link-config.properties` and security configurations for the destination cluster in another file with `--command-config`. To learn more about how to configure cluster links for different use cases, see [basic tutorial](topic-data-sharing.md#tutorial-topic-data-sharing), the [Hybrid tutorial](hybrid-cp.md#cluster-link-hybrid-cp) linking Confluent Platform and Confluent Cloud clusters, the rest of this page which details the command line options, and the [Cluster Linking configuration guide](configs.md#cluster-link-configs). ### Listing cluster links **Example Command** ```bash kafka-cluster-links --list --bootstrap-server localhost:9093 ``` **Example Output** ```bash Link name: 'example-link', link ID: '123-some-link-id', remote cluster ID: '123-some-cluster-id', local cluster ID: ', local cluster ID: '456-some-other-cluster-id'', remote cluster available: 'true' ``` You can list existing cluster links. The command returns the link name, link ID (an internally allocated unique ID), the cluster ID of the linked cluster, and whether the linked cluster is available or not. `--link` : If provided, only lists the specified cluster link.
* Type: string `--command-config` : Property file containing configurations to be passed to the [AdminClient](../../installation/configuration/admin-configs.md#cp-config-admin). For example, with security credentials for authorization and authentication.
* Type: string `--include-topics` : If provided, includes a list of all mirror topics on this cluster link.
* Type: string You must have `DESCRIBE CLUSTER` authorization to list cluster links. ### Describing a cluster link **Example Command** ```bash kafka-configs --bootstrap-server localhost:9093 \ --describe \ --cluster-link example-link ``` **Example Output** ```bash Dynamic configs for cluster-link example-link are: metadata.max.age.ms=300000 sensitive=false synonyms={} ... ``` To describe a cluster link, use `kafka-configs` along with [bootstrap-server](#bootstrap-cluster-links) and these flags. `--cluster-link` : (Required) The name of the cluster link to describe.
* Type: string `--command-config` : Property file containing configurations to be passed to the [AdminClient](../../installation/configuration/admin-configs.md#cp-config-admin). For example, with security credentials for authorization and authentication.
* Type: string You must have `DESCRIBE CLUSTER` authorization to describe a cluster link. ### Altering a cluster link **Example Command** ```bash kafka-configs --bootstrap-server localhost:9093 \ --alter \ --cluster-link example-link \ --add-config cluster.link.retry.timeout.ms=10000 \ --delete-config request.timeout.ms ``` **Example Output** ```bash Completed updating config for cluster-link example-link. ``` To alter an existing link, use `kafka-configs` along with [bootstrap-server](#bootstrap-cluster-links) and these flags. `--alter` : (Required) Alter a link. `--cluster-link` : (Required) The name of the cluster link to alter.
* Type: string `--command-config` : Property file containing configurations to be passed to the [AdminClient](../../installation/configuration/admin-configs.md#cp-config-admin). For example, with security credentials for authorization and authentication. At least one of the following must be provided: `--add-config` : Configurations to add in the form of “key=value” directly on the command line. You can use square brackets to group values which contain commas. For a full list of available configurations, see [Cluster link configuration properties](configs.md#cluster-link-specific-configs).
* Type: string `--add-config-file` : Path to properties file containing [configurations](configs.md#cluster-link-specific-configs) to add.
* Type: string `--delete-config` : Comma-separated list of configuration keys to delete.
* Type: string You must have `ALTER CLUSTER` authorization to modify the cluster associated with a link, as described in [Authorization (ACLs)](security.md#cluster-link-acls). #### Pause and resume the link to verify the new configuration ​For dynamically updatable configurations (such as SSL) it is possible that some connections are using an older configuration while others are using a newer one. The system avoids recreating connections that that are not required for an action. The API and CLI output will show the latest persisted configurations, but these may not represent the true state of the configurations after a `kafka-configs --alter` operation. To confirm that your new configurations have been applied, use `cluster.link.paused` to pause and unpause the link to restart it. To learn more, see `cluster.link.paused` in [Configure Cluster Linking on Confluent Platform](configs.md#cluster-link-configs). ### Viewing a cluster link task status You can view the status of the following configurable tasks: - [Consumer offset sync](mirror-topics-cp.md#mirror-topics-consumer-offsets) - [ACL sync (migrate ACLs)](security.md#cluster-link-acls-migrate) - [Topic configurations sync](mirror-topics-cp.md#sync-topic-configs) - [Auto-create mirror topics](mirror-topics-cp.md#auto-create-mirror-topics-concepts) To view the status of any given task on Confluent Platform, use the following command: ```bash confluent kafka link task list ``` Or: ```bash ./bin/kafka-cluster-links.sh ... --list-tasks --link ``` #### Examples - Example 1: Using the Confluent CLI Get the a list of tasks. ```bash confluent kafka link task list link-1 ``` Your output should resemble: ```bash Task Name | State | Errors ---------------------+----------------+--------------------------------- AclSync | NOT_CONFIGURED | AutoCreateMirror | NOT_CONFIGURED | ConsumerOffsetSync | IN_ERROR | REMOTE_LINK_NOT_FOUND_ERROR: | | "Failed to get remote link | | config due to link not being | | found on the remote cluster." TopicConfigsSync | ACTIVE | ``` - Example 2: Using the command line Kafka scripts Get a list of tasks. ```bash kafka-cluster-links --list-tasks --bootstrap-server ``` For example: ```bash kafka-cluster-links --list-tasks --bootstrap-server localhost:9093 ``` Your output should resemble: ```bash Link: demo-link Task name: ConsumerOffsetSync State: NOT_CONFIGURED Task name: AclSync State: NOT_CONFIGURED Task name: TopicConfigsSync State: ACTIVE Task name: AutoCreateMirror State: NOT_CONFIGURED ``` #### Task state descriptions Tasks can be in any one of the following states: | Task Status | Description | |-------------------|--------------------------------------------------------------------------------------------------------------------------------------------| | `ACTIVE` | Task is configured, running and healthy. | | `IN_ERROR` | Task is configured, but is encountering errors. Use the error code and error message to understand why the task is in error. | | `NOT_CONFIGURED` | Task is not configured to run. | | `LINK_FAILED` | Link is in a failed state, so the task is not running. | | `LINK_UNAVAILBLE` | Link is in an unavailable state, so the task is not running. | | `LINK_PAUSED` | Link is in a paused state, so the task is not running. | | `UNKNOWN` | Link state is unknown. A task is not expected to be in this state for prolonged periods of time. If it is, reach out to Confluent support. | ### Deleting a cluster link **Example Command** ```bash kafka-cluster-links --bootstrap-server localhost:9093 \ --delete \ --link example-link ``` **Example Output** ```bash Cluster link 'example-link' deletion successfully completed. ``` To delete an existing link, use `kafka-cluster-links` along with [bootstrap-server](#bootstrap-cluster-links) and these flags. `--link` : (Required) The name of the cluster link to describe.
* Type: string `--command-config` : Property file containing configurations to be passed to the [AdminClient](../../installation/configuration/admin-configs.md#cp-config-admin). For example, with security credentials for authorization and authentication.
* Type: string `--validate-only` : If provided, validates the cluster link deletion but doesn’t apply the delete. `--force` : Force deletion of a link even if there are mirror topics are currently linked with it. You must have `ALTER CLUSTER` authorization to delete a cluster link, as described in [Authorization (ACLs)](security.md#cluster-link-acls). #### IMPORTANT - When deleting a cluster link, first check that all mirror topics are in the `STOPPED` state. If any are in the `PENDING_STOPPED` state, deleting a cluster link can cause irrecoverable errors on those mirror topics due to a temporary limitation. - When a cluster link is deleted, so is the history of any `STOPPED` topics. If you need the `Last Source Fetch Offset` or the `Status Time` of your promoted or failed-over mirror topics, make sure to save those before you delete the cluster link. - You cannot delete a cluster link that still has mirror topics on it (the delete operation will fail). - If you are using Confluent for Kubernetes (CFK), and you delete your cluster link resource, any mirror topics still attached to that cluster link will be forcibly converted to regular topics by use of the `failover` API. To learn more, see [Modify a mirror topic](https://docs.confluent.io/operator/current/co-link-clusters.html#modify-a-mirror-topic) in [Cluster Linking using Confluent for Kubernetes](https://docs.confluent.io/operator/current/co-link-clusters.html#). ### Creating a mirror topic A mirror topic is a read-only topic that reflects all the data and metadata in another topic. Creating a mirror topic with the CLI uses the `kafka-mirrors` tool. Once a mirror topic is created, the mirror automatically begins fetching data from the source topic. For more information, see [Mirror Topics](https://docs.confluent.io/platform/current/multi-dc-deployments/cluster-linking/mirror-topics-cp.html). **Example Command** ```bash kafka-mirrors --create --mirror-topic example-topic \ --link demo-link \ --bootstrap-server localhost:9093 ``` **Example Output** ```bash Created topic example-topic. ``` To create a mirror topic, use `kafka-cluster-links` along with [bootstrap-server](#bootstrap-cluster-links) and the following flags. `--mirror-topic` : (Required) The name of the mirror topic to create. This must match exactly the name of the source topic to mirror over the cluster link.
* Type: string `--link` : (Required) The name of the cluster link used to pull data from the source topic.
* Type: string `--command-config` : Property file containing configurations to be passed to the [AdminClient](../../installation/configuration/admin-configs.md#cp-config-admin). For example, with security credentials for authorization and authentication. The following are optional configurations when creating a mirror topic: `--config` : A comma-separated list of configs to override when creating the mirror topic. Each config to override should be specified as `name=value`. For more information about which configurations can be set on a mirror topic, see [Configurations](https://docs.confluent.io/platform/current/multi-dc-deployments/cluster-linking/mirror-topics-cp.html#configurations) in Mirror Topics.
* Type: string `--replication-factor` : The replication factor of the mirror topic being created. If not supplied, *defaults to the destination cluster’s default*, not the source topic’s replication factor.
* Type: string `--source-topic` : The name of the source topic to mirror. Required if the cluster link has a prefix configured. To learn more, see [Prefixing Mirror Topics and Consumer Group Names](https://docs.confluent.io/platform/current/multi-dc-deployments/cluster-linking/mirror-topics-cp.html#prefixing-mirror-topics-and-consumer-group-names).
* Type: string You must have `ALTER CLUSTER` authorization to create a mirror topic. ### Listing mirror topics To list mirror topics on a cluster across all cluster links, use `kafka-mirrors --list`. This command can either list the mirror topics for a specific cluster link, or can list all mirror topics across all cluster links on the cluster. **Example Command** ```bash kafka-mirrors --list --bootstrap-server localhost:9093 ``` **Example Output** ```bash topic1 topic2 topic3 topic4 ``` These parameters can be added to the command: `--link` : (optional) The name of the cluster link to filter on. If provided, only mirror topics on this cluster link will be listed.
* Type: string `--include-stopped` : (optional) If this flag is added, the list will include any topics that were formerly mirror topics but were stopped by a `promote` or `failover` command. This flag does not take an argument. ### Describing mirror topics Use `kafka-mirrors` to describe and manage topics that are mirrored over a cluster link. ```bash kafka-mirrors --describe --bootstrap-server pkc-nwnyk.us-west-2.aws.confluent.cloud:9092 --command-config lkc-rn220.config --links onprem-to-cloud Topic: web.orders.modified LinkName: onprem-to-cloud LinkId: 2c3dee72-f54b-4bb3-9694-cf29caaf3637 MirrorTopic: web.orders.modified State: ACTIVE StateTime: 2021-11-10 15:33:29 Partition: 0 State: ACTIVE DestLogEndOffset: 114123 LastFetchSourceHighWatermark: 114123 Lag: 0 TimeSinceLastFetchMs: 8385101 Partition: 1 State: ACTIVE DestLogEndOffset: 115278 LastFetchSourceHighWatermark: 115278 Lag: 0 TimeSinceLastFetchMs: 8387954 Partition: 2 State: ACTIVE DestLogEndOffset: 112210 LastFetchSourceHighWatermark: 112210 Lag: 0 TimeSinceLastFetchMs: 8508856 Partition: 3 State: ACTIVE DestLogEndOffset: 120887 LastFetchSourceHighWatermark: 120887 Lag: 0 TimeSinceLastFetchMs: 8389749 Partition: 4 State: ACTIVE DestLogEndOffset: 109225 LastFetchSourceHighWatermark: 109225 Lag: 0 TimeSinceLastFetchMs: 8385101 Partition: 5 State: ACTIVE DestLogEndOffset: 111669 LastFetchSourceHighWatermark: 111669 Lag: 0 TimeSinceLastFetchMs: 8387954 Topic: web.orders LinkName: onprem-to-cloud LinkId: 2c3dee72-f54b-4bb3-9694-cf29caaf3637 MirrorTopic: web.orders State: ACTIVE StateTime: 2021-11-10 15:33:29 Partition: 0 State: ACTIVE DestLogEndOffset: 294760 LastFetchSourceHighWatermark: 294760 Lag: 0 TimeSinceLastFetchMs: 8387954 Partition: 1 State: ACTIVE DestLogEndOffset: 285862 LastFetchSourceHighWatermark: 285862 Lag: 0 TimeSinceLastFetchMs: 8508856 Partition: 2 State: ACTIVE DestLogEndOffset: 284891 LastFetchSourceHighWatermark: 284891 Lag: 0 TimeSinceLastFetchMs: 8389749 Partition: 3 State: ACTIVE DestLogEndOffset: 285982 LastFetchSourceHighWatermark: 285982 Lag: 0 TimeSinceLastFetchMs: 8385101 Partition: 4 State: ACTIVE DestLogEndOffset: 277379 LastFetchSourceHighWatermark: 277379 Lag: 0 TimeSinceLastFetchMs: 8387954 Partition: 5 State: ACTIVE DestLogEndOffset: 283731 LastFetchSourceHighWatermark: 283731 Lag: 0 TimeSinceLastFetchMs: 8508856 Topic: inventory.shipments LinkName: onprem-to-cloud LinkId: 2c3dee72-f54b-4bb3-9694-cf29caaf3637 MirrorTopic: inventory.shipments State: ACTIVE StateTime: 2021-11-10 15:51:21 ``` As with other Kafka commands in Confluent Platform, `--bootstrap-server` is a required flag for the `kafka-mirrors` command. `--bootstrap-server` : (Required) The connection string for the broker(s) in a cluster is in the form `host:port` (which can be a comma-separated list for multiple brokers). You must specify the destination cluster where you plan to create mirror topics. The destination cluster must be running Confluent Platform 6.0.0 or later, which is required to support Cluster Linking.
* Type: string * Default: empty string These parameters can be added to the command: `--links` : (optional) The name of the cluster link to filter on. If provided, only mirror topics on this cluster link will be listed.
* Type: string `--include-stopped` : (optional) If this flag is added, the list will include any topics that were formerly mirror topics but were stopped by a `promote` or `failover` command. This flag does not take an argument. ### Stopping mirroring on a topic You can use either the `kafka-mirrors --failover` or `kafka-mirrors --promote` command to stop mirroring on a topic. Both commands permanently convert the mirror topic into a regular, writable topic, but each is designed for a specific use case. - The `promote` command is intended for migrations and, therefore, performs some additional verification and operations before stopping the mirroring. - The `failover` command is intended for disaster recovery, and so takes effect immediately and always succeeds. No additional operations are performed. - The `--validate-only` option previews the results of a command before executing it. You can perform `promote` and `failover` commands on multiple topics at the same time by passing in a list of topic names using the `--topics` flag, and a comma-separated list of topic names. Even if you are only promoting 1 topic, you still use the `--topics` (plural) flag. For example: ```bash --topics topic1 --topics topic1,topic2,topic3 ``` More examples are shown below. #### Promote a topic Use `kafka-mirrors --promote` to stop mirroring and convert a mirror topic to a regular topic in a graceful process typically appropriate for migration scenarios. This command checks that there is 0 lag between the source topic and the mirror topic, and does one final sync of the metadata (consumer group offsets and topic configs) before converting the mirror topic into a regular topic. #### NOTE - The `promote` command does not stop data from being produced to the source topic. It is possible for producers to produce more data to the source topic after the check for 0 lag succeeded. In that case, the two topics will have diverged, and that data will not be mirrored to the regular topic (formerly mirrored). - If there is lag between the source topic and the mirror topic at the time that the promote command is run, the promote command will fail. **Examples** You can use `promote` to specify one topic or multiple topics on the same cluster link to stop mirroring. The command syntax is: ```bash kafka-mirrors --promote --topics .. --bootstrap-server ``` Here is an example of using `kafka-mirrors --promote` to stop mirroring on a topic called `example-topic` which uses a link named `example-link`. ```bash kafka-mirrors --promote --topics example-topic --bootstrap-server localhost:9093 ``` #### Failover a topic Use `kafka-mirrors failover` to stop mirroring and convert a mirror topic to a regular topic immediately. This is typically appropriate for disaster recovery where the source cluster fails unexpectedly. The `failover` command does not perform additional checks or syncs. It works like a “forced” version of `promote`. You can specify one topic or multiple topics to stop mirroring on. **Examples** You can use `failover` to specify one topic or multiple topics on the same cluster link to stop mirroring. The command syntax is: ```bash kafka-mirrors --failover --topics .. --bootstrap-server ``` Here is an example of using `kafka-mirrors --failover` to stop mirroring on a topic called `example-topic` which uses a link named `example-link`. The link will be inferred; it should not be included in the command, as shown below. ```bash kafka-mirrors --failover --topics example-topic --bootstrap-server localhost:9093 ``` #### Reverse a source and mirror topic on bidirectional links with reverse commands The source topic -> mirror topic relationship can be reversed using the `reverse-and-start` or `reverse-and-pause` commands. These cause the source topic to become the mirror topic, and the mirror topic to become the source topic. These commands and this workflow require that the cluster link is in [bidirectional mode](configs.md#bidirectional-cluster-linking) and on Confluent Platform both clusters must be Confluent Platform version 7.7 or later. To learn more, see [Reverse a source and mirror topic](mirror-topics-cp.md#reverse-source-and-mirror-topic). #### Restore mirroring after a failover with truncate-and-restore If you want to restore mirroring after a `promote` or a `failover`, you can use the `truncate-and-restore` command. After failing over or promoting a mirror topic, you can run `truncate-and-restore` on the original primary topic that will make it a mirror fetching from the newly-stopped mirror topic. This command will also truncate and delete any divergent records that were produced to the original primary cluster after the point of failover. This means that there could be some loss of data if your clients are not set up to reprocess data. To learn more, see [Convert a mirror topic to a normal topic](mirror-topics-cp.md#convert-mirror-topic-to-normal-topic). `truncate-and-restore` is available only on [“bidirectional” links](mirror-topics-cp.md#bidirectional-linking-cp), and only in KRaft mode. 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 [Platform Quick Start](../../get-started/platform-quickstart.md#cp-quickstart-step-1). Also, the [basic Cluster Linking tutorial](topic-data-sharing.md#tutorial-topic-data-sharing) includes a full walkthrough of how to run Cluster Linking in KRaft mode. #### IMPORTANT Using the `truncate-and-restore` command on Confluent Platform **7.9.0 - 7.9.2 or 8.0.0** and using [Tiered Storage](../../clusters/tiered-storage.md#tiered-storage) on the topics being truncated can result in a silent failure leading to data inconsistency or even limited data loss. If you are using these versions and Tiered Storage as part of a disaster recovery (DR) failback strategy with bidirectional links, do not run `truncate-and-restore`. To resolve the issue, [upgrade to a patched version](../../installation/installing_cp/zip-tar.md#prod-kafka-cli-install): Confluent Platform 7.9.3, 8.0.1, 8.1.0 and later have mitigated this issue, and the features can be safely used. #### Validate only (dry run) To test the results of either `promote` or `failover` before executing the commands, add the `--validate-only` flag. For example: ```bash kafka-mirrors --failover --topics example-topic --bootstrap-server localhost:9093 --validate-only ``` ### Pausing and resuming a mirror topic You can use the pause (`kafka-mirrors --pause`) and resume (`kafka-mirrors --unpause`) commands to temporarily pause and resume mirroring. To pause a mirror topic: ```bash kafka-mirrors --pause --topics example-topic --bootstrap-server localhost:9093 ``` To resume a mirror topic use `--unpause`: ```bash kafka-mirrors --unpause --topics example-topic --bootstrap-server localhost:9093 ``` ## Migrating consumer groups from source to destination cluster To migrate a consumer group across the link, set `consumer.offset.sync.enable=true` in your link configuration, specify a group filter in a JSON file, and pass the name of the file as the value for the `--consumer-group-filters-json-file` flag on the [CLI commands](#bootstrap-cluster-links). You can set this at the time you create the link, or as an update to an existing configuration. #### NOTE Consumer group filters should only include groups that are not being used on the destination. This will help ensure that the system does not override offsets committed by other consumers on the destination. The system attempts to work around filters containing groups that are also used on the destination, but in these cases there are no guarantees; offsets may be overwritten. For mirror topic “promotion” to work, the system must be able to roll back offsets, which cannot be done if the group is being used by destination consumers. This example assumes you are migrating group “someGroup” from cluster “broker-west” to cluster “broker-east” and the state before the migration is executed is that you are currently migration all offsets with the following filter set. ```json {"groupFilters": [ { "name": "*", "patternType": "LITERAL", "filterType": "INCLUDE" } ]} ``` To migrate a consumer group from a source cluster to a destination cluster, follow these steps. 1. Stop the consumer on the source cluster. 2. Wait for a period of 2x `consumer.offset.sync.ms`. 3. Make sure that Cluster Linking replication is beyond the latest committed offset. You can confirm this with the following commands. - Check the CURRENT-OFFSET on the source cluster. ```bash kafka-consumer-groups --bootstrap-server broker-west:19091 --describe --group someGroup ``` Your output should resemble the following: ```bash GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID someGroup west-offsets 0 100 100 0 - - - ``` - Check the LOG-END-OFFSET on the destination cluster and ensure it is equal or larger than the CURRENT-OFFSET recorded above. ```bash kafka-consumer-groups --bootstrap-server broker-east:19092 --describe --group someGroup ``` Your output should resemble the following: ```bash GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID someGroup west-offsets 0 100 100 0 ``` 4. Verify the current offset is consistent in source and destination using the following commands. - Check the CURRENT-OFFSET on the source cluster. ```bash kafka-consumer-groups --bootstrap-server broker-west:19091 --describe --group someGroup ``` Your output should resemble the following: ```bash GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID someGroup west-offsets 0 100 100 0 - - - ``` - Check the CURRENT-OFFSET on the destination cluster. ```bash kafka-consumer-groups --bootstrap-server broker-east:19092 --describe --group someGroup ``` Your output should resemble the following: ```bash GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-ID someGroup west-offsets 0 100 100 0 ``` 5. Update the offset migration filters to remove the group from the migration process. ```json echo "consumer.offset.group.filters={\"groupFilters\": [ \ { \ \"name\": \"*\", \ \"patternType\": \"LITERAL\", \ \"filterType\": \"INCLUDE\" \ }, \ { \ \"name\": \"someGroup\", \ \"patternType\": \"LITERAL\", \ \"filterType\": \"EXCLUDE\" \ } \ ]}" > newFilters.properties kafka-configs --bootstrap-server broker-east:19092 --alter --cluster-link offsets-cluster-link --add-config-file newFilters.properties ``` 6. Start the consumer on the destination cluster. ## REST API commands The Cluster Linking REST API is documented in the [REST API Proxy v3 Documentation](https://docs.confluent.io/platform/current/kafka-rest/api.html#clusterlinking). ## Related content - [Configure Cluster Linking on Confluent Platform](configs.md#cluster-link-configs) - [CLI Tools Shipped With Confluent Platform](../../tools/cli-reference.md#cp-all-cli) - [Dynamically change broker settings](../../kafka/dynamic-config.md#kafka-dyn-broker-config) (to learn more about using the `kafka-configs` command) - [Kafka consumer group tool](../../clients/consumer.md#kafka-consumer-describe-group) (to learn more about using the `kafka-consumer-groups` command) - [Tutorial: Share Data Across Topics Using Cluster Linking for Confluent Platform](topic-data-sharing.md#tutorial-topic-data-sharing) - [Tutorial: Link Confluent Platform and Confluent Cloud Clusters](hybrid-cp.md#cluster-link-hybrid-cp) - [Manage Mirror Topics for Cluster Linking on Confluent Platform](mirror-topics-cp.md#mirror-topics-concepts) - [Manage Security for Cluster Linking on Confluent Platform](security.md#cluster-link-security)