# Manage Mirror Topics for Cluster Linking on Confluent Platform A mirror topic is a read-only topic on a destination cluster. A cluster link creates and owns it, and it receives a byte-for-byte, offset-preserving copy of a source topic. This page explains how to create, configure, and manage mirror topics on Confluent Platform. ## Mirror topics overview Mirror topics are read-only topics that a cluster link creates and owns. A cluster link connects two Kafka clusters and manages data transfer between them. To learn more, see [Cluster Linking for Confluent Platform](index.md#cluster-linking). Mirror topics are the building blocks for moving data with Cluster Linking. The following sections describe what mirror topics are, how they are created and configured, and how they work in operation. This information applies to both Confluent Platform and Confluent Cloud. ## How mirror topics relate to cluster links A cluster link connects a mirror topic to its source topic. Any messages produced to the source topic are mirrored over the cluster link to the mirror topic. A mirror topic syncs many of its configurations from its source topic. It can also sync access control lists (ACLs) and consumer group offsets from its source topic, if you enable those features on the cluster link. For a detailed summary and explanation of how mirror topics get their configurations, see [Mirror topic configuration syncing](#sync-topic-configs). You can [convert a mirror topic to a regular topic](#convert-mirror-topic-to-normal-topic) and stop the mirroring relationship using the Cluster Linking `promote` and `failover` commands. To restore mirroring after a `failover` or a `promote`, you can run `truncate-and-restore` on the original source topic to make it mirror from the newly stopped mirror topic. You can [reverse the mirroring relationship](#reverse-source-and-mirror-topic) with the `reverse-and-start` and `reverse-and-pause` commands, which cause the mirror topic to become the source topic and the source topic to become the mirror topic. The following diagram shows how mirror topics work, including the relationship between the mirror topic and its source topic, and the syncing of ACLs and consumer offsets. ![Diagram of a mirror topic receiving data, ACLs, and consumer offsets from its source topic over a cluster link](multi-dc-deployments/cluster-linking/images/cluster-link-mirror-topics-example.png) ## Mirror topic properties Mirror topics have the following properties: - A cluster link creates and owns its mirror topics. - Mirror topics get their messages from their source topic. They are byte-for-byte, offset-preserving asynchronous copies of their source topics. - Mirror topics are read-only; you can consume them in the same way as any other topic, but you cannot produce into them. If a producer tries to produce a message into a mirror topic, the action fails. The only way to get a message into a mirror topic is to produce the message to the mirror topic’s source topic. - Many of the mirror topic’s configurations are copied and synced from the source topic. For the full list, see [Mirror topic configuration syncing](#sync-topic-configs). ## Mirror topic creation You can create a mirror topic using the [Confluent Cloud Console](/cloud/current/get-started/cloud-basics.html), the Confluent Cloud REST API, the Confluent CLI, the [Confluent Platform AdminClient API](#create-mirror-topic-adminclient-api), or Confluent for Kubernetes. Alternatively, you can configure your cluster link to automatically create mirror topics that match certain [prefixes](#cluster-link-prefix-concepts). ### Requirements You can create mirror topics only with the [mirror topic command](#mirror-topic-create-command-syntax) or by enabling [auto-create mirror topics](#auto-create-mirror-topics-concepts) on the cluster link. You cannot pre-create a mirror topic with a non-mirror-topic command. 1. Creating a mirror topic requires an existing cluster link. The cluster link creates the mirror topic on its destination cluster. You must have access to the destination cluster. 2. A mirror topic is always created with the same name as its source topic. A topic of that name must exist on the source cluster. The only exception is when a cluster link has `link.prefix` configured, which adds a prefix to the name of the mirror topic. 3. The destination cluster must be able to reach the source cluster and verify that a suitable topic exists. A mirror topic cannot be created if the source cluster is unreachable. 4. Creating a mirror topic on Confluent Cloud requires you to have the `CloudClusterAdmin`, `EnvironmentAdmin`, or `OrgAdmin` role over the destination cluster, that is, the cluster where the mirror topic is created. Alternatively, you can have the [appropriate ACLs](/cloud/current/multi-cloud/cluster-linking/security-cloud.html#rbac-roles-and-kafka-acls-summary). 5. The cluster link’s principal must have both `DeveloperRead` and `DeveloperManage` on the relevant source topic on the source cluster. Alternatively, it could have `ResourceOwner` or the [appropriate ACLs](/cloud/current/multi-cloud/cluster-linking/security-cloud.html#rbac-roles-and-kafka-acls-summary) on that topic. You do not need any permissions on the source cluster. ### Create a mirror topic on Confluent Cloud Console If the destination cluster is a Confluent Cloud cluster, you can view and create mirror topics on the [Confluent Cloud Console](https://confluent.cloud/): 1. On the top-level page showing your environments, select **Cluster links**. 2. Click an existing cluster link, or create a new one. If you create a new cluster link, follow the prompts. 3. To add a mirror topic to an existing cluster link, click **Add mirror topic**. - If the source cluster is a Confluent Cloud cluster that you have access to, the dialog shows a drop-down list of the source topics. - If the source cluster is external to Confluent Cloud or you do not have access to it, a text box appears instead, where you add the name of the source topic. Enter the source topic name and click **Add** to create the mirror topic. ![Confluent Cloud Console showing the Cluster links tab on the environment page](multi-dc-deployments/cluster-linking/images/cluster-link-mirror-topics-add-1.png)![Confluent Cloud Console showing the details of a cluster link with the Add mirror topic option](multi-dc-deployments/cluster-linking/images/cluster-link-mirror-topics-add-2.png)![Confluent Cloud Console Add mirror topic dialog with a field for the source topic name](multi-dc-deployments/cluster-linking/images/cluster-link-mirror-topics-add-3.png) ### Create a mirror topic with the Confluent CLI To create a mirror topic with the Confluent CLI, the general syntax is: ```bash confluent kafka mirror create --link ``` The command must be run against the destination cluster. If needed, you can specify the destination cluster with `--cluster `. To learn more, see [confluent kafka mirror create](https://docs.confluent.io/confluent-cli/current/command-reference/kafka/mirror/confluent_kafka_mirror_create.html) in the command reference. If the cluster link is configured with `link.prefix`, then `--source-topic source-topic-name` must be passed, too. For example: ```bash confluent kafka mirror create west.clicks --link from-west --source-topic clicks ``` On Confluent Platform clusters, you can use either the Confluent CLI or the `bin/kafka-mirrors` script. The general syntax to create a mirror topic is: ```bash kafka-mirrors --create --mirror-topic \ --link \ --bootstrap-server ``` To learn more, see [Cluster Linking on Confluent Platform](/platform/current/multi-dc-deployments/cluster-linking/index.html). ### Create a mirror topic with the REST API On Confluent Cloud: - To create a mirror topic, send a `POST` request to the destination cluster’s REST API endpoint at: `/kafka/v3/clusters/{cluster_id}/links/{link_name}/mirrors`. - Include the following in the payload: ```bash { "source_topic_name": "", "mirror_topic_name": "", // (only required if link.prefix is configured) } ``` The preceding payload shows the only required parameters. More options are available to override topic configurations. To learn more, see [Create a mirror topic](https://docs.confluent.io/cloud/current/ccloud/create-kafka-mirror-topic/) in the Confluent Cloud API reference. ### Mirror topic creation examples For examples of how to create mirror topics on Confluent Platform, see [Create the cluster link and mirror topic](/platform/current/multi-dc-deployments/cluster-linking/topic-data-sharing.html#create-the-cluster-link-and-the-mirror-topic) (step 2, “Initialize the mirror topic”) in the basic tutorial and [Creating a mirror topic](/platform/current/multi-dc-deployments/cluster-linking/commands.html#creating-a-mirror-topic) in the commands documentation. For examples of how to create mirror topics on Confluent Cloud, see the following sections: - [Create source and mirror topics](/cloud/current/multi-cloud/cluster-linking/quickstart.html#create-source-and-mirror-topics) in the Quick Start for Cluster Linking on Confluent Cloud - [Mirror a topic](/cloud/current/multi-cloud/cluster-linking/topic-data-sharing.html#mirror-a-topic) on the Confluent Cloud Cluster Linking tutorial - [examples](https://docs.confluent.io/confluent-cli/current/command-reference/kafka/mirror/confluent_kafka_mirror_create.html#examples) on the Confluent CLI command reference ### Create a mirror topic with the AdminClient API On Confluent Platform, you can use the AdminClient API to create mirror topics. To learn more, see [ConfluentAdmin API reference](/platform/current/clients/javadocs/javadoc/org/apache/kafka/clients/admin/ConfluentAdmin.html). ### Additional requirements when prefixing is enabled When a cluster link has a prefix set, the cluster link adds the specified prefix to the beginning of mirror topic names. For example, if you set the prefix to `west`, the source topic `orders` is mirrored as `west.orders`. If the cluster link is configured for prefixing mirror topic names, then to create a mirror topic you must pass both the mirror topic name and the source topic name (instead of only the source topic name). To learn more about prefixing, see [Prefix mirror topics and consumer group names](#cluster-link-prefix-concepts). ### Bidirectional cluster linking To establish bidirectional linking between two clusters, you must use two cluster links. You cannot establish bidirectional linking with a single cluster link. For an example of bidirectional linking, see the Hybrid tutorial (on either [Confluent Cloud](/cloud/current/multi-cloud/cluster-linking/hybrid-cc.html) or [Confluent Platform](/platform/current/multi-dc-deployments/cluster-linking/hybrid-cp.html)), which sets up bidirectional linking between on-premises and cloud clusters. Bidirectional linking is supported for different topics. For a specific topic, only unidirectional linking is supported. ### Cherry pick which topics to mirror To cherry pick topics to be mirrored, you can use any of the following methods: - [Set filters](#filters-for-auto-create-mirror-topics). - Manually mirror individual topics with the [confluent kafka mirror create](https://docs.confluent.io/confluent-cli/current/command-reference/kafka/mirror/confluent_kafka_mirror_create.html#examples) command or `kafka-mirrors` command as shown in the tutorials: [quick start for Confluent Cloud](/cloud/current/multi-cloud/cluster-linking/quickstart.html#create-source-and-mirror-topics) and [tutorial for Confluent Platform](/platform/current/multi-dc-deployments/cluster-linking/topic-data-sharing.html#create-the-cluster-link-and-the-mirror-topic). - Use the [Confluent Cloud Cluster Linking (v3) REST API](/cloud/current/api.html#tag/Cluster-Linking-(v3)) (see “Create a mirror topic”). ### Support for compacted topics Cluster Linking supports compacted topics. A compacted topic is mirrored as such from source to destination. To learn more, see the FAQs for [Confluent Cloud](/cloud/current/multi-cloud/cluster-linking/faqs-cc.html) and [Confluent Platform](/platform/current/multi-dc-deployments/cluster-linking/faqs-cp.html). ### Cluster link and mirror topic configuration example This diagram shows an example of a cluster link and a mirror topic configured with some of the preceding properties. ![A cluster link carrying topic data, ACLs, and consumer offsets from a source topic on one cluster to a mirror topic on another cluster](multi-dc-deployments/cluster-linking/images/cluster-link-example.png) ## Auto-create mirror topics A cluster link is able to automatically create mirror topics on the destination cluster for any topics that exist on the source cluster. This is called “auto-creating” mirror topics. This saves time and effort because you do not have to create mirror topics manually. You can scope this capability down to a specific set of topics by matching on the topics’ names. ### Enable auto-create mirror topics To enable auto-create mirror topics, you must set two properties on the cluster link. You can set these properties when a cluster link is created, or update an existing cluster link with these properties. These properties are: `auto.create.mirror.topics.enable` : Whether to auto-create mirror topics based on topics on the source cluster. When set to `true`, the cluster link auto-creates mirror topics. Setting this option to `false` disables mirror topic creation and clears any existing filters.
* Type: boolean * Default: false `auto.create.mirror.topics.filters` : - A JSON object with one property, `topicFilters`, that contains an array of filters that specify which topics to mirror. The following sections describe the filters. - This list must have at least one filter. - Ordering of the filters in this array does not matter.
* Type: array * Default: empty
**Syntax**
```properties { "topicFilters": [ ] } ``` ### Schedule and frequency of mirror topic auto-create task The auto-topic creation schedule and frequency is driven by `metadata.max.age.ms`, which controls the delay between subsequent auto mirror tasks. `metadata.max.age.ms` : Maximum amount of time in milliseconds that the client can use a cached metadata value before it is refreshed from the brokers.
* Type: integer * Default: five minutes (300,000 milliseconds) Using the default, clients cache metadata about the cluster for up to five minutes before they refresh it. Reducing the value specified for `metadata.max.age.ms` increases the frequency with which mirror topics are auto-created. If the `metadata.max.age.ms` value is too low, it can cause a refresh of the metadata too frequently, which can impact the performance of the auto-create mirror topic feature. In this case, you can see errors or delays in the creation of mirror topics. However, if the `metadata.max.age.ms` value is too high, it can result in stale metadata, which can lead to inconsistencies between the source and mirror topics. ### Filters for auto-create mirror topics In both Confluent Cloud and Confluent Platform, auto-creating mirror topics automatically excludes Confluent internal topics and topics prefixed with `_confluent` even if these are user-created. For full detail on this, see [Topics not mirrored](#topics-not-mirrored). All other filtering options described in this section are also available in both Confluent Cloud and current releases of Confluent Platform. Confluent Replicator uses the internal `__consumer_timestamps` topic for [consumer offset translation](/platform/current/multi-dc-deployments/replicator/replicator-failover.html#understanding-consumer-offset-translation); this topic should not be mirrored. You must filter this topic out using the auto-create mirror topics `EXCLUDE` filters, as described in the following sections. Other topics can be excluded using filters. For example, if a different topic name is used for Schema Registry storage, instead of `_schemas`, it can be excluded by using filters. The following sections describe how to filter topics for auto-create mirror topics. You can select exactly which source topics to automatically mirror through a list of filters. A cluster link accepts an unlimited number of filters. Each filter is a JSON object with the following fields: `name` : Text that is matched against the name of the topic. Set `name` to the wildcard, `*`, to apply to all topics. `patternType` : Either `LITERAL` or `PREFIXED`.
- If `name` is set to `foo`, then setting `patternType` to `LITERAL` matches only a topic named `foo`. - Setting `patternType` to `PREFIXED` matches any topic names that begin with “foo”, for example, “foo”, “football”, and “foo.fighters”. `filterType` : Either `INCLUDE` or `EXCLUDE`.
- If `filterType` is set to `INCLUDE`, any topic names on the source cluster that match this filter are created as mirror topics. - If `filterType` is set to `EXCLUDE`, any matching topic names are not created as mirror topics. In other words, prevents auto mirror topic creation for the specified topic names. `EXCLUDE` filters override any overlapping `INCLUDE` filters. For example, if you have an `INCLUDE` filter for the prefix “foo” but have an `EXCLUDE` filter for the prefix “foo.bar,” then a topic on the source cluster named “foo.fighters” is mirrored automatically, but a topic named “foo.bar.fighters” is not mirrored automatically. ### Example filters #### Mirror all topics This filter creates mirror topics for all current and future source cluster topics: ```properties { "topicFilters": [ {"name": "*", "patternType": "LITERAL", "filterType": "INCLUDE"} ] } ``` #### Mirror all topics that begin with a given string This filter mirrors all topics that begin with “foo”: ```properties { "topicFilters": [ {"name": "foo", "patternType": "PREFIXED", "filterType": "INCLUDE"} ] } ``` #### Mirror all topics except those that begin with “secret” This filter mirrors all topics except those that begin with “secret”: ```properties { "topicFilters": [ {"name": "*", "patternType": "LITERAL", "filterType": "INCLUDE"}, \ {"name": "secret", "patternType": "PREFIXED", "filterType": "EXCLUDE"} ] } ``` #### Mirror named topics if they exist on the source cluster This filter mirrors three topics, “liz”, “jack”, and “kenneth”, if they exist on the source cluster: ```properties { "topicFilters": [ {"name": "liz", "patternType": "LITERAL", "filterType": "INCLUDE"}, \ {"name": "jack", "patternType": "LITERAL", "filterType": "INCLUDE"}, \ {"name": "kenneth", "patternType": "LITERAL", "filterType": "INCLUDE"} ] } ``` ### How a mirror topic is auto-created For a given topic on a cluster link’s source cluster (the “source topic”), the cluster link auto-creates a new mirror topic if all these conditions are true: - `auto.create.mirror.topics.enable` is set to `true`. - `auto.create.mirror.topics.filters` has filters which `INCLUDE` the source topic name. - The cluster link’s security credential is authorized (through source cluster ACLs) to read the source topic. - No topic by that name already exists on the destination cluster. - If prefixing is enabled on the cluster link, then the source topic cannot be a mirror topic. You cannot “chain” mirror topics when both `auto.create.mirror.topics.enable` and prefixing are enabled. If any of the preceding conditions are false, the cluster link does not auto-create a mirror topic for the given source topic. ### Override topic configurations when using auto-create mirror topics To override a topic configuration when using auto-create mirror topics, you have two options: * Change the topic configuration after the mirror topic is automatically created. * Use the CLI or API to manually create the mirror topic, and override the configuration. Even if a topic matches the auto-create mirror topic filters, it can still be manually created as a mirror topic before the cluster link creates it automatically. Auto-create mirror topics runs once every five minutes, so the mirror topic can be manually created soon after the cluster link is created or soon after the source topic is created. ### Delete topics that were auto-created You cannot delete a mirror topic that matches the auto-create mirror topics filters. If you deleted such a topic, and there was a topic of the same name on the source cluster, the mirror topic would be automatically re-created and sync all its history (if `mirror.start.offset.spec` is set to the default). The mirror topic is recreated automatically, so the delete does not persist. To delete a mirror topic while auto-create mirror topics is enabled, you have three options: delete the source topic first, exclude the topic’s name from the auto-create mirror topics filters, or disable auto-create mirror topics. - **Option 1. Delete the source topic first** - Given a source topic named `cool-topic`, if you delete the source topic and then want to subsequently delete the associated mirror topic (`cool-topic` on the destination), wait until the mirror topic becomes a `FAILED` mirror topic (which can take up to five minutes), after which point you can delete it. You can also call `failover` or `promote` on the mirror topic to transition it to the `STOPPED` state. Both `FAILED` and `STOPPED` mirror topics can be deleted. - **Option 2. Exclude the topic name from the auto-create mirror topics filters** - This strategy prevents the mirror topic from overlapping with the auto-create filters. An easy way to remove a given topic from the filters is to add an `EXCLUDE` filter for that topic name. You can add `cool-topic` to the `EXCLUDE` filters, even if no such source topic exists. After editing the auto-create mirror topic filters, you can delete the mirror topic. - **Option 3. Disable auto-create mirror topics on the cluster link** - After the setting has been disabled, the mirror topic can be deleted. If needed, auto-create mirror topics can be immediately re-enabled on the cluster link. To learn more, see [Disable auto-create mirror topics](#disable-auto-create-mirror-topics) and [Mirror topic deletion](#mirror-topic-delete). ### Disable auto-create mirror topics To disable auto-create mirror topics entirely, set this property on the cluster link: ```properties auto.create.mirror.topics.enable=false ``` Here’s an example of how to set that property with the CLI: ```bash echo "auto.create.mirror.topics.enable=false" > tmp.txt confluent kafka link update --config-file tmp.txt rm tmp.txt ``` ## Prefix mirror topics and consumer group names Cluster links can be configured with a prefix (`cluster.link.prefix`) that is applied to the names of the mirror topics and, optionally, the names of the consumer groups that are managed by the cluster link at the destination cluster. This enables topics and consumer groups from different source clusters that have the same name to be synced to the destination without name clashes. It also enables all mirror topics from a cluster link to be categorized and managed under one prefix on the destination. For example, consider two links, link-1 and link-2. link-1 is linking data from cluster s1 to destination and link-2 is linking data from s2 to destination, and furthermore s1 and s2 both contain a topic “clicks”. Without prefixing, it would be impossible for both links to sync data for their own “clicks” topic as they would have the same name on the destination cluster. With prefixing, each link can have its own unique prefix that is applied to the topic name as its mirrored. link-1 could have prefix `usa_` and link-2 could have prefix `eu_`. Finally, at the destination cluster there would be two topics, `usa_clicks` and `eu_clicks`. If the link is configured with a prefix, when a mirror topic is created (for example, with `confluent kafka mirror create`) then the mirror topic name must begin with the prefix (otherwise, the operation fails). If [auto-create mirror topics](#auto-create-mirror-topics-concepts) is used, the cluster link automatically names the topics created on the destination with the prefix. The prefix can optionally be applied to the consumer groups that are created on the destination cluster because of consumer group offset syncing. When offsets are synced, consumer groups are created on the destination. With this feature, you can prefix the consumer group name on the destination. This enables consumer group offsets to be synced even when two (or more) consumer groups from two (or more) different source clusters have the same name. For example, if link-1 had consumer group g1 and link-2 had consumer group g1, then prefixing would result in two consumer groups at the destination: `usa_g1` and `eu_g1`. By default, consumer group names are not prefixed. To enable prefixing, set `consumer.group.prefix.enable` to `true` in the cluster link configuration. Here’s an example configuration file for Confluent Enterprise that contains only the elements relevant to prefixing: ```properties bootstrap.servers=localhost:9092 cluster.link.prefix=usa_ consumer.offset.sync.enable=true auto.create.mirror.topics.enable=true auto.create.mirror.topics.filters={"topicFilters":[{"name": "*","patternType": "LITERAL","filterType": "INCLUDE"}]} consumer.group.prefix.enable=false acl.sync.enable=false ``` Here, a prefix of `usa_` has been configured and `consumer.group.prefix.enable` has been set to `false` (which is the default, but shown here for context). All mirror topic names on the destination start with the prefix, and consumer group names remain the same as they are on the source. `acl.sync.enable` is set to `false`, which is required because `auto.create.mirror.topics.enable` is set to `true` and prefixing is enabled. For details, see [Limitations on prefixing](#limitations-on-prefixing-in-mirror-topics). On Confluent Cloud, these configurations are specified on the command line or the Confluent Cloud Console. ### Limitations on prefixing - The prefix cannot be changed after the cluster link is created. - A prefix can be a maximum of 12 characters. - Valid characters are `[a-zA-Z0-9._-]`. Note this is a regular expression pattern. The square brackets `[ ]` are not included in the valid characters set. The prefix can consist of alphanumeric characters, a period, an underscore, and a hyphen. - ACL syncing and prefixing cannot be enabled together on a single cluster link. Note that you can always sync ACLs on a separate link. Create a new link and configure it to sync ACLs. That said, even if a separate link is configured to sync ACLs, the ACLs are synced only for topics with the exact same name, and no mapping or sync exists for prefixed topics. - Consumer group prefixing cannot be enabled for bidirectional links. Setting `consumer.group.prefix.enable` to `true` on a bidirectional cluster link results in an “invalid configuration” error stating that the cluster link cannot be validated due to this limitation. - Prefixing cannot be combined with chaining and auto-create mirror topics at the same time. When auto-mirroring and prefixing is configured, a link cannot mirror a topic that is itself a mirror topic at the source cluster. For example, consider the preceding links, link-1 and link-2. If a new link-3 was created, auto-mirroring would not be able to mirror data from `usa_clicks` or `eu_clicks` or any mirror topic on the destination (even if it didn’t have a prefix) because they are mirror topics. This is done as a safeguard to prevent auto-mirroring from creating an infinite number of topics due to cyclical cluster link connections. - The `reverse-and-start` and `reverse-and-pause` commands are not supported on cluster links configured with a topic prefix (`cluster.link.prefix`). Failover and failback workflows that require reversing the link direction must use cluster links with standard (non-prefixed) topic names. ## Aggregate multiple source cluster topics into a single topic Cluster Linking can be used for aggregating data from multiple identical source clusters into one destination cluster. For example, each source cluster can be running in a different region, collecting local data, and Cluster Linking can stream data from each local cluster to a central, aggregate cluster. Every topic, on every source cluster, that you want to aggregate needs its own uniquely named mirror topic on the aggregate cluster. An easy way to do this is by setting a unique [prefix](#cluster-link-prefix-concepts) on each cluster link. If a consumer group needs to read the data from all source clusters (for example, from all regions), it can easily consume multiple mirror topics at the same time by consuming from a regular expression topic pattern that matches all the mirror topic names you want to source from (rather than consuming from a single topic name). Most open source Kafka clients support consuming from a regular expression topic pattern. ![Several regional source clusters each mirroring their topics, under distinct prefixes, into a single central aggregate cluster](multi-dc-deployments/cluster-linking/images/cluster-link-aggregate-topics.png) ## Topics not mirrored By design, the following topics are not mirrored (synced). Cluster Linking does not create mirrors for these topics, even if they match the topic name filter. - Internal or “system” topics (for example, any topic prefixed with `_confluent`) - Any topic prefixed with `_confluent`, even if these are user-created and not system topics. - `confluent-audit-log-events` - The topic that holds schemas (default name `_schemas`) For more background detail, some of the internal topics not mirrored include the following: - `__consumer_offsets`: This topic stores consumer group offsets. Cluster Linking uses a [built-in offset sync mechanism](#mirror-topics-consumer-offsets), rather than “mirror” this topic in the standard sense. - `__transaction_state`: This topic stores transaction states and helps maintain transactional integrity. This topic is not replicated because Cluster Linking does not support transactions. - `__cluster_metadata`: In KRaft mode, this topic is used for internal cluster coordination. You do not need to mirror this topic. ## Mirroring lag Mirroring lag is the delay between when a message is produced to the source topic and when it appears on the mirror topic. The mirror process runs asynchronously. Therefore, there is often some mirroring lag between the source topic and the mirror topic. The most recent messages on the source topic might not yet have been mirrored to the mirror topic, so the mirror topic can often be slightly behind the source topic. The same is true for syncing the topic configuration, the consumer group offsets, and the ACLs. All of these processes are asynchronous, so the changes happen first on the source topic, and then on the mirror topic shortly after. ## Sync consumer group offsets Cluster Linking can automatically synchronize consumer group offsets from a source cluster to mirror topics on a destination cluster. This is critical for disaster recovery (DR) scenarios, so that consumer applications can fail over to a backup cluster and resume processing at or near the point where they left off without reprocessing old messages or losing data. ### Enabling consumer group offset sync and specifying filters To set this up, you configure the following properties (as described in [Migrating consumer groups from source to destination cluster](commands.md#cluster-link-migrate-consumer-groups)): - `consumer.offset.sync.enable` - Set this to `true` to sync consumer group offsets. (The default is `false`.) - `consumer.offset.group.filters` - Pass in a JSON file with a pattern that is matched against consumer group names to identify which groups to mirror. If these two properties are set, the cluster link syncs the consumer group offsets of any matching consumer groups for all mirror topics that the link mirrors. #### NOTE Consumer group filters should not include groups that are being used on the destination. This helps ensure that the system does not override offsets committed by other consumers on the destination, or overwrite the consumer offsets while consumer groups are consuming from the mirror topic. If you are unsure about which consumer groups are being used on the destination, disable consumer offset sync on the cluster link until you verify this. ### Why consumer offsets are clamped after failover Consumer offsets are clamped to prevent a consumer group from resuming past the last offset the cluster link actually mirrored, which would cause it to skip unmirrored messages. Clamping happens when either `failover` or `promote` is called on a mirror topic, or when a consumer group moves to the destination cluster and consumes its first message from a mirror topic, and consumer offset sync is enabled on the cluster link. The cluster link does not allow the consumer offsets it synced for that topic to be larger than the last offsets on the mirror topic (the log end offset). If any of these consumer offsets are larger or further than the log end offset, those consumer offsets are reset to the log end offset. (The use of `failover` and `promote` is described in [Convert a mirror topic to a normal topic](#convert-mirror-topic-to-normal-topic).) To illustrate with an example: suppose a source topic with one partition had messages up to offset 100 and is mirrored over a cluster link. However, there was mirroring lag on the cluster link, and only the messages up to offset 90 were mirrored when a disaster hit the source topic. At this point, you call failover on the mirror topic. Consumer group A was at offset 80 on the source cluster, so it remains at offset 80 on the destination cluster, since that offset was mirrored to the mirror topic. But consumer group B was at offset 95, which was not mirrored to the mirror topic. If consumer group B started consuming at offset 95 on the mirror topic, then it would miss any messages at offsets 90-94 that were produced to the topic. To avoid that problem, the cluster link “clamps” consumer group B’s offsets down to 90, which is the highest offset on the mirror topic. ### Failover considerations for active-passive vs. active-active Key considerations related to offsets are the following: - In an active-passive deployment, all producers and consumers are interacting with the active cluster only. On failover, the producers and consumers stop interacting with the failed cluster, and start interacting with the active cluster only. The best practice for this scenario is to specify `LOCAL_MIRROR` only in the offset sync configuration. Given this configuration, upon failover the system always syncs offsets in one direction only. - In an active-active deployment, where consumers are on both sides, producers could be on both sides, or only one side. In this scenario, you must have unique consumer groups on both sides, and in the offset sync filtering you must specify the exact consumer groups you want to sync to prevent cycles. You can configure offsets to use `LOCAL_MIRROR` and `REMOTE_MIRROR`. Importantly, on failover, you should update the offset sync filter on both sides to include or exclude the consumer groups. To learn more, see [Bidirectional Cluster Linking](configs.md#bidirectional-cluster-linking). ### Limitations and troubleshooting An issue exists where consumer group offsets that are deleted on the destination cluster persist, instead of being removed as expected. To prevent this from happening, extend retention on the destination to make sure data is deleted on the source before it is deleted on the destination. To do this, increase `offsets.retention.minutes` on destination cluster by at least double `offsets.retention.check.interval.ms`. To learn more, see [Known limitations and best practices](index.md#cluster-linking-limitations) in the Cluster Linking overview. ### Related consumer offset sync documentation To learn more, see [Migrating consumer groups from source to destination cluster](commands.md#cluster-link-migrate-consumer-groups) and the command descriptions under [Configuration options specific to cluster links](configs.md#cp-cluster-link-config-options) in [Configure Cluster Linking on Confluent Platform](configs.md#cluster-link-configs). ## Define the direction of offset syncing When configuring consumer offset syncing for Cluster Linking, specify whether to sync offsets for local mirror topics (`LOCAL_MIRROR`), remote mirror topics (`REMOTE_MIRROR`), or both. Understanding when to use each option is critical for achieving reliable failover and avoiding offset sync conflicts. Use the `topicTypes` field in the group filter JSON to specify which topics to sync. - By default, offsets are synced for both local and remote mirrors (`["LOCAL_MIRROR", "REMOTE_MIRROR"]`). - `REMOTE_MIRROR` only takes effect when the link is in bidirectional mode. For example, the following filter syncs offsets for all consumer groups on both local and remote mirror topics in bidirectional mode: ```json { "groupFilters": [ { "name": "*", "patternType": "LITERAL", "filterType": "INCLUDE", "topicTypes": ["LOCAL_MIRROR", "REMOTE_MIRROR"] } ] } ``` For more examples of filter configurations, see [Guidance for active-passive deployments](#cluster-link-guidance-for-active-passive) and [Guidance for active-active deployments](#cluster-link-guidance-for-active-active). ### LOCAL_MIRROR `LOCAL_MIRROR` syncs consumer group offsets for topics that are mirrored on the local cluster; that is, the cluster where the cluster link object is configured. When `LOCAL_MIRROR` is specified: - The cluster link fetches consumer offsets from the remote cluster, which is the source of the mirror topic. - These offsets are applied to the local cluster, where the mirror topic resides. - This enables consumers to resume from the correct position when they fail over from the source cluster to the destination cluster where the mirror topics reside. `LOCAL_MIRROR` determines the exact offset a consumer should resume from when failing over from a remote source topic to a local mirror topic. ### REMOTE_MIRROR `REMOTE_MIRROR` syncs consumer group offsets for topics that are mirrored on the remote cluster. When `REMOTE_MIRROR` is specified: - The cluster link pulls consumer offsets from the mirror topic on the remote cluster and writes them to the source topic on the local cluster. - This is used in bidirectional configurations where the remote cluster also has mirror topics that need offset sync. - This enables consumers to resume from the correct position when they fail back from the destination cluster to the original source cluster. `REMOTE_MIRROR` determines the exact offset a consumer should resume from when failing back from a remote mirror topic to a local source topic. ### Offset syncing behavior summary | Mode | Offsets fetched from | Offsets written to | Use case | |-----------------|-------------------------|------------------------|--------------------------------------| | `LOCAL_MIRROR` | Remote cluster (source) | Local cluster (mirror) | Failover from source to DR cluster | | `REMOTE_MIRROR` | Remote cluster (mirror) | Local cluster (source) | Failback from DR to original cluster | ### Guidance for active-passive deployments In an active-passive deployment: - All producers and consumers interact with the active (primary) cluster only. - The passive (DR) cluster receives mirrored data but does not have active consumers until a failover occurs. - On failover, producers and consumers stop interacting with the failed cluster and start using the DR cluster. #### Recommended configuration for active-passive deployments Specify `LOCAL_MIRROR` only in the offset sync configuration on the disaster recovery (DR) cluster’s cluster link. This ensures offsets sync in one direction only: from the active cluster to the DR cluster. #### Example configuration for active-passive Create a configuration properties file, `dr-link.properties`, for the cluster link on the DR cluster. #### NOTE When defining complex JSON filters inside a Confluent Platform properties file, you must escape the double quotes (`\"`) and write the entire JSON block on a single line. ```properties link.mode=BIDIRECTIONAL consumer.offset.sync.enable=true consumer.offset.group.filters={\"groupFilters\": [{\"name\": \"*\", \"patternType\": \"LITERAL\", \"filterType\": \"INCLUDE\", \"topicTypes\": [\"LOCAL_MIRROR\"]}]} consumer.offset.sync.ms=5000 ``` To apply this configuration file to a new or existing cluster link, use the `kafka-cluster-links` command-line tool. For example, to update an existing link: ```bash kafka-cluster-links --bootstrap-server \ --alter \ --link my-cluster-link \ --add-config-file dr-link.properties ``` With this configuration: - Consumer offsets from the active cluster are synced to the DR cluster for all consumer groups. - When failover occurs, consumers can resume from their last committed offset on the DR cluster. - No offsets are synced back to the active cluster, which is appropriate because the active cluster is the source of truth until failover. #### Failback after recovery After the original cluster recovers and you want to fail back: 1. Use the `truncate-and-restore` command on the original cluster’s topics to make them mirror the DR cluster’s topics. 2. Update the cluster link configuration on the original cluster to enable `LOCAL_MIRROR` offset sync. 3. Use `reverse-and-start` to restore the original mirroring direction. ### Guidance for active-active deployments In an active-active deployment: - Producers may write to either, or both, clusters. - Consumers are active on both clusters simultaneously. - Each cluster has writable topics and mirror topics, typically with prefixes to avoid name clashes. #### Key requirements for active-active deployments - **Unique consumer group names:** Each consumer group must exist on only one cluster at a time. If you need consumers in both regions, use different consumer group names for each region. - **Explicit group filtering:** Do not use wildcard filters (`*`) for both `LOCAL_MIRROR` and `REMOTE_MIRROR`. Instead, to prevent cycles, explicitly specify which consumer groups should sync and in which direction. - **Prefixed topics:** Use `cluster.link.prefix` to distinguish mirror topics from source topics. #### Example configuration for active-active Consider two clusters, cluster A (us-east) and cluster B (us-west), with the following setup: - Cluster A has topic `orders` (source) and `west.orders` (mirror from cluster B) - Cluster B has topic `orders` (source) and `east.orders` (mirror from cluster A) - Consumer group `cg-east` consumes on cluster A - Consumer group `cg-west` consumes on cluster B **Cluster A’s cluster link configuration:** ```properties link.mode=BIDIRECTIONAL cluster.link.prefix=west. consumer.offset.sync.enable=true consumer.offset.group.filters={\"groupFilters\": [{\"name\": \"cg-east\", \"patternType\": \"LITERAL\", \"filterType\": \"INCLUDE\", \"topicTypes\": [\"REMOTE_MIRROR\"]}, {\"name\": \"cg-west\", \"patternType\": \"LITERAL\", \"filterType\": \"INCLUDE\", \"topicTypes\": [\"LOCAL_MIRROR\"]}]} ``` **Cluster B’s cluster link configuration:** ```properties link.mode=BIDIRECTIONAL cluster.link.prefix=east. consumer.offset.sync.enable=true consumer.offset.group.filters={\"groupFilters\": [{\"name\": \"cg-west\", \"patternType\": \"LITERAL\", \"filterType\": \"INCLUDE\", \"topicTypes\": [\"REMOTE_MIRROR\"]}, {\"name\": \"cg-east\", \"patternType\": \"LITERAL\", \"filterType\": \"INCLUDE\", \"topicTypes\": [\"LOCAL_MIRROR\"]}]} ``` With this configuration: - `cg-east` offsets sync from cluster A to cluster B for the `east.orders` mirror topic while active on cluster A. - If `cg-east` fails over to cluster B, its forward-progress offsets sync back from cluster B to cluster A based on the remote mirror mapping. - Because a consumer group can only actively consume on one cluster at a time, its offsets flow in only one direction at any given moment, preventing cyclical overwrites. #### Failover in active-active When a region fails in an active-active setup: 1. Move affected consumers to the surviving region. 2. Update the offset sync filters on the surviving cluster to exclude the moved consumer groups from sync to prevent the surviving cluster from overwriting their offsets. 3. After recovery, update filters again before moving consumers back. ### Best practices for offset sync configuration - **Start with LOCAL_MIRROR for DR scenarios:** For standard disaster recovery, `LOCAL_MIRROR` on the DR cluster is sufficient. Only add `REMOTE_MIRROR` if you need failback capabilities with offset preservation. - **Use explicit filters in active-active:** Never use `"name": "*"` with both `LOCAL_MIRROR` and `REMOTE_MIRROR` `topicTypes`, as this can create offset sync cycles. - **Match consumer group lifecycles:** Ensure that consumer groups are explicitly stopped on one cluster before being started on another to avoid offset conflicts. - **Configure appropriate sync intervals:** Adjust `consumer.offset.sync.ms` based on your recovery point objectives (RPO) requirements. Lower values, for example: 1000 ms, provide more frequent syncs but increase overhead. - **Monitor offset sync state:** Use the `confluent kafka link list` command to verify that consumer offset sync status is active and healthy across your links. ### Limitations on bidirectional consumer offset sync Consumer group offsets that are deleted on the destination cluster, especially those that are auto-deleted, persist instead of being removed as expected. Internally, the offsets are re-replicated to the destination before retention settings delete the offsets from source. This results in extended retention of inactive consumer group offsets. To prevent this, extend retention on the destination to ensure data is deleted on the source before it is deleted on the destination. To do this, increase `offsets.retention.minutes` on the destination cluster by at least double `offsets.retention.check.interval.ms` (converted to minutes). #### IMPORTANT If you use bidirectional linking to sync the same consumer group between both sides, the above argument infinitely extends retention due to the cyclical nature of the argument. To avoid this, ensure that each consumer group exists on only one cluster at a time, as recommended in [Guidance for active-active deployments](#cluster-link-guidance-for-active-active). #### Related bidirectional offset sync documentation - [Bidirectional Mode Cluster Linking](configs.md#bidirectional-cluster-linking) - [Considerations for failover scenarios](#considerations-for-failover-scenarios-offset-configs) ## Reverse a source and mirror topic 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. The following diagram shows how a `reverse-and-pause` command works. The `reverse-and-start` command behaves in a similar fashion, except that instead of the extra manual step to resume the mirror topic to become an active mirror topic, the command automatically converts the topic to an active mirror topic. ![Diagram of a reverse-and-pause operation, showing the source and mirror topics exchanging roles and the new mirror topic left in a paused state](multi-dc-deployments/cluster-linking/images/cluster-link-reverse-mirror.png) ### How reverse-and-start and reverse-and-pause work The [reverse-and-start](https://docs.confluent.io/cloud/current/ccloud/update-kafka-mirror-topics-reverse-and-start-mirror/) command leaves the new mirror topic in an active mirroring state, whereas [reverse-and-pause](https://docs.confluent.io/cloud/current/ccloud/update-kafka-mirror-topics-reverse-and-start-mirror/) leaves the new mirror topic in a paused state until the [resume](https://docs.confluent.io/cloud/current/ccloud/update-kafka-mirror-topics-resume/) command is called. These commands are available in the Confluent CLI at [confluent kafka mirror](https://docs.confluent.io/confluent-cli/current/command-reference/kafka/mirror/index.html#confluent-kafka-mirror), the REST API at [reverse](https://docs.confluent.io/cloud/current/ccloud/update-kafka-mirror-topics-reverse-and-start-mirror/), and `kafka-mirrors` in Confluent Platform 7.7+. Cluster Linking ensures that both topics have the same data and metadata at the point of change, so no data is left behind. After the command is called, the source topic does not accept new writes, which allows the mirror topic to catch up and perform the reversal. After the reversal is complete, data written to the (new) source topic then flows to the (new) mirror topic. This provides a fast and efficient fail back for a planned failover mechanism, allowing you to fail over to the mirror site, produce new data, and then fail back to the original site. ### Requirements for using “reverse” commands - The cluster link must be in [Bidirectional mode](/cloud/current/multi-cloud/cluster-linking/cluster-links-cc.html#bidirectional-mode). - On Confluent Platform, both clusters must be Confluent Platform version 7.7 or later. - Both clusters must be healthy and able to communicate over the network. - You must have the `CLUSTER:ALTER` ACL or the Admin role on the cluster where this command is run. Additionally, you need `ALTER` permissions on all relevant topics, such as `ALTER: Topic (Mirror)`. - Make sure you have monitoring in place to check all the different states the topics go through (as described in [Process flow for “reverse” commands](#reverse-mirror-process-flow)). If at any point in time there are any issues in the process, you can always use the `failover` command to get a mirror topic to a writable state. - You should run this command with only one topic at a time for transactional producers, and you must monitor each topic to the end state before running the command for the next topic. If you run the command in batch, you must make sure that all topics are transitioned to a writable state before restarting the application for production. Otherwise, if the applications are restarted before the topics are transitioned into the end state, this can result in the new records not being persisted in Kafka because it starts writing to an immutable topic. ### Process flow for “reverse” commands Calling the `reverse` commands follows this chain of events: 1. Call `reverse-and-start` or `reverse-and-pause` on the mirror topic. - Make sure this command is used against the cluster that hosts the mirror topic; for example, the disaster recovery (DR) cluster. - Multiple topics can be reversed at once using the REST API. - You must have the `CLUSTER:ALTER` ACL or the Admin role on the cluster where this command is run. 2. The mirror topic enters `PENDING_SYNCHRONIZE` state, and the source topic enters `PENDING_MIRROR` state. During this time: - The source topic does not accept any new writes (produce requests). - The mirror topic fetches all data from the source topic until it is up to date. - **Tip:** The larger the mirroring lag on the mirror topic, the longer this step takes. To minimize the amount of time when both topics are in a read-only state, call the command at a time when mirroring lag is at zero, or very low. 3. Once the data has been synchronized, the (old) mirror topic enters `PENDING_STOPPED` state. - During this time, the old mirror topic fetches any last metadata (such as, consumer offsets). - Monitor for errors that could cause this step to hang using the state transition error API or Metrics. 4. The (new) source topic enters the `STOPPED` mirror state, and accepts writes as a regular topic. - The (new) mirror topic enters the `ACTIVE` or `PAUSED` state, depending on which command was called. The mirroring relationship is reversed. ### Limitations on reverse commands - The “reverse” commands only work with hybrid links if the on-premises cluster is on Confluent Platform 7.7 or later. These commands do not work on earlier versions, pre Confluent Platform 7.7. - The “reverse” commands cannot be used in Confluent Platform when unclean leader election is enabled. - Terraform does not support the failback APIs. - The “reverse” commands do not support [prefixed cluster links](#cluster-link-prefix-concepts) (`cluster.link.prefix`). Failover and failback workflows that require reversing the link direction must use cluster links with standard (non-prefixed) topic names. ## Convert a mirror topic to a normal topic If you want to convert a mirror topic into a normal topic that you can produce into, you can call the `failover` or the `promote` command on the mirror topic, using the commands [confluent kafka mirror promote](https://docs.confluent.io/confluent-cli/current/command-reference/kafka/mirror/confluent_kafka_mirror_promote.html) and [confluent kafka mirror failover](https://docs.confluent.io/confluent-cli/current/command-reference/kafka/mirror/confluent_kafka_mirror_failover.html), respectively. Specifically, you call `confluent kafka mirror promote --link ` or `confluent kafka mirror failover --link `. Both the `failover` and `promote` commands occur on the destination cluster (the mirror topic’s cluster) and require you to pass in the cluster link’s name. promote : The `promote` option is often used for migrations. It checks that there is no mirroring lag, config sync lag, or consumer offset lag between the source topic and the mirror topic. Then, it converts the mirror topic into a full topic, with the assurance that this topic was exactly the same as its source topic. The destination cluster’s brokers must be able to reach the source cluster’s brokers to make this check, so your source cluster must be online. failover : The `failover` option is often used when a disaster has hit the source cluster (for example, a cloud region outage) and you want to shift operations from the source topic to the mirror topic. This command succeeds regardless of the mirroring lag or the source cluster’s reachability. 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 makes it a mirror fetching from the newly stopped mirror topic. This command also truncates and deletes 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. (This command is available only on “bidirectional” links, and only in KRaft mode. See the following important note.) reverse-and-start : After running `truncate-and-restore`, you can restore the original primary and secondary regions by running the `reverse-and-start` command on the new mirror topic. (This command is available only on “bidirectional” links.) #### IMPORTANT - You can revert a promoted or failed-over topic back into a mirror topic by using `truncate-and-restore`. This command is available only on [“bidirectional” links](#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. - You can run `mirror describe` (`confluent kafka mirror describe --link `) on a promoted or failed over mirror topic, if you do not delete the cluster link. If you delete the cluster link, you lose the history and, therefore, `mirror describe` does not find data on promoted or failed over topics. - You cannot change a mirror topic to use a different cluster link or make changes to the link itself, other than to recreate the mirror topic on a different link. - You cannot delete a cluster link that still has mirror topics on it (the delete operation fails). - If you are using Confluent for Kubernetes (CFK), and you delete your cluster link resource, CFK forcibly converts any mirror topics still attached to that cluster link to regular topics with 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#). ### Example of topic migration ![Diagram of a topic migration, showing a mirror topic promoted to a regular topic on the destination cluster](multi-dc-deployments/cluster-linking/images/cluster-link-migrate.png) ### Example of failing over a topic ![Diagram of a failover, showing a mirror topic converted to a writable topic after the source cluster becomes unavailable](multi-dc-deployments/cluster-linking/images/cluster-link-failover.png) ### Example of truncate and restore ![Diagram of a truncate-and-restore operation, showing divergent records truncated from the original primary topic before it resumes mirroring](multi-dc-deployments/cluster-linking/images/cluster-link-truncate-and-restore.png) ### Example of reverse and start ![Diagram of a reverse-and-start operation, showing the mirror topic becoming the source topic and the source topic becoming the mirror topic](multi-dc-deployments/cluster-linking/images/cluster-link-reverse-and-start.png) ## Mirror topic states and statuses When you describe a mirror topic, it returns one of these states: `ACTIVE` : The mirror is running normally, and messages are being mirrored from the source topic to the destination topic. `PAUSED` : - You paused mirroring for this mirror topic. - To reach this state, you must either pause this specific topic or pause its cluster link. `PENDING_SYNCHRONIZE` : - This topic is in the process of becoming a regular topic that is mirrored to the remote cluster. (Prior to this, the topic was a mirror topic.) - This topic is currently read-only. (It does not accept produce requests, but you can consume from it.) - The `PENDING_SYNCHRONIZE` state occurs when a `reverse` command is called on a topic. - Allowed operations to a topic in this state are as follows: - `pause`: to pause the reversal process. - `failover`: to permanently abort the reversal process and convert this to a writable, non-mirror topic. - The topic automatically transitions to the `STOPPED` state when ready. `PENDING_MIRROR` : - This topic is in the process of becoming a mirror topic (it was formerly a writable topic). - It does not accept produce requests, but you can consume from it (it is read-only). - The `PENDING_MIRROR` state occurs when a `reverse` command is called on the remote mirror topic for this source topic. - Allowed operations to a topic in this state are as follows: - `failover`: to permanently abort the reversal process and convert this to a writable, non-mirror topic. - When its conversion to a mirror topic has completed, the topic automatically transitions to the `ACTIVE` or `PAUSED` state, depending on which `reverse` command was used. `PENDING_SETUP_FOR_RESTORE` : - This topic is in the initial reconciliation phase of a restore operation. - In this state, Cluster Linking compares epoch histories and determines the exact offset boundaries required to identify divergent records produced while the topic was writable. - It does not accept produce requests. While the topic remains technically readable, clients should avoid consuming from it during this state, as reading data can cause offset clamping issues during the restoration process. - The `PENDING_SETUP_FOR_RESTORE` state occurs when you issue a `restore` command on a topic. - When offset reconciliation completes, the topic automatically transitions to the `PENDING_RESTORE_MIRROR` state. - Allowed operations to a topic in this state are as follows: - `failover`: to permanently stop the restoration process and convert this to a writable, non-mirror topic. `PENDING_RESTORE_MIRROR` : - This topic is in the active truncation and re-linking phase of a restore operation. - In this state, Cluster Linking truncates the divergent records identified during setup and re-establishes partition fetchers to the remote source cluster. - It does not accept produce requests. While the topic remains technically readable, clients should avoid consuming from it during this state, as reading data can cause offset clamping issues during the restoration process. - The `PENDING_RESTORE_MIRROR` state occurs automatically after the `PENDING_SETUP_FOR_RESTORE` state completes. - When truncation and fetcher initialization complete, the topic automatically transitions to the `ACTIVE` state, or to the `PAUSED` state if restored as paused. - Allowed operations to a topic in this state are as follows: - `failover`: to permanently abort the restoration process and convert this to a writable, non-mirror topic. `PENDING_STOPPED` : - You stopped this mirror topic with the `promote` command, and this topic moves to the `STOPPED` state shortly. - To force the mirror topic to immediately go from the `PENDING_STOPPED` state to the `STOPPED` state, call the `failover` command on it. Doing this cancels any synchronization that was happening between the source cluster and the destination cluster, and eliminates any guarantees that the `promote` command gives. - As a workaround, you can fail over a topic stuck in `PENDING_STOPPED` to force it into a writable state. However, some messages are reprocessed because, in the case of a “stuck” `reverse-and-start`, not all consumer group offsets are guaranteed to be copied over. To learn more about this troubleshooting scenario, see [Process flow for “reverse” commands](#reverse-mirror-process-flow). `STOPPED` : - Mirroring has permanently stopped for this topic. It no longer receives messages from its source topic. The topic is now writable and can receive messages produced directly to it. - To get into this state, you must call either `promote` or `failover` on this mirror topic. - Even though a `STOPPED` topic is no longer a mirror topic, it is still listed in the output for the commands `confluent kafka mirror list` and `confluent kafka mirror describe --link ` for as long as the cluster link exists. This is useful because the topic returns the last offset it fetched from its source topic (last source fetch offset) for each partition, and the time at which it was stopped (status time). `SOURCE_UNAVAILABLE` : - The mirror topic is unable to reach the source topic, and is not mirroring messages from the source topic. This could happen if the source cluster is experiencing an outage or if the network between the destination cluster and the source cluster is unstable. - Mirroring resumes after the issue is resolved and the destination cluster can reach the source cluster. #### NOTE Using a Confluent Platform 7.0.x source cluster with a source-initiated link to a KRaft destination cluster will generate a `SOURCE_UNAVAILABLE` error. Cluster Linking between a source cluster running Confluent Platform 7.0.x or earlier (non-KRaft) and a destination cluster running in KRaft mode is not supported. To solve for this, upgrade the source cluster to Confluent Platform 7.1.0 or later. `LINK_FAILED` : - An error has broken the mirror topic’s cluster link, and no data is being mirrored. You must manually reconfigure the link. `FAILED` : - The mirror topic has permanently failed. It no longer mirrors data. This can happen if the cluster link ACLs are removed from the source cluster, or if the source topic is deleted. In both cases, the failed status takes effect only after [cluster.link.retry.timeout.ms](/platform/current/multi-dc-deployments/cluster-linking/configs.html#configuration-options) is reached (by default, the system retries the link for five minutes). - You can stop this mirror with the `failover` command, and it becomes a regular topic. - If you want to restore mirroring for this topic, you must delete the legacy mirror topic and create a new mirror topic with the same name. ## View mirror topic state transition errors You can use the following commands to view mirror topic state transition errors. For example, when a mirror topic is promoted, it transitions from the `PENDING_STOPPED` state to `STOPPED` state. During that process, various actions are performed to implement the transition and errors can occur during that implementation. The following APIs allow you to view these errors and unblock the mirror topic state transitions. For example, if you see an authentication issue, you can reconfigure the link’s credentials to allow the mirror topic to be fully promoted. For a full list of possible task states and error codes, see [Troubleshooting Cluster Linking on Confluent Cloud](/cloud/current/multi-cloud/cluster-linking/trouble-cc.html). To view errors associated with a state transition: ### Confluent Cloud ```bash confluent kafka mirror state-transition-error list --link ``` ### Confluent Platform ```bash ./bin/kafka-mirrors.sh ... --list-state-transition-errors --topics ``` ### REST API See [Describe the mirror topic](https://docs.confluent.io/cloud/current/ccloud/read-kafka-mirror-topic/) in the Confluent Cloud REST API documentation. To view a mirror topic status, send a GET request to `/clusters//links//mirrors/?include_state_transition_errors=true`. ### Examples ### Confluent Cloud ```bash confluent kafka mirror state-transition-error list topic-1 --link link-1 Mirror State Transition Error | Mirror State Transition Error Code | Message ---------------------------------+--------------------------------- AUTHENTICATION_ERROR | Failed to describe topic | configs due to authentication | issues. ---------------------------------+--------------------------------- ``` ### Confluent Platform ```bash ./bin/kafka-mirrors.sh --bootstrap-server pkc-j581r8.us-west2.gcp.confluent.cloud:9092 --command-config command-config.properties --list-state-transition-errors --topics topic-1 Topic: topic-1 State: PENDING_STOPPED Error Code: AUTHENTICATION_ERROR Error Message: "Failed to describe topic configs due to authentication issues." ``` ### REST API ```bash curl -H "Authorization: Basic XXX" --request GET \ --url 'https://pkc-j581r8.us-west2.gcp.confluent.cloud:443/kafka/v3/clusters/lkc-ok51xj/links/link-1/mirrors/topic-2?include_state_transition_errors=true' | jq % Total % Received % Xferd Average Speed Time Time Time Current Dload Upload Total Spent Left Speed 100 844 0 844 0 0 2217 0 --:--:-- --:--:-- --:--:-- 2215 { "kind": "KafkaMirrorData", "metadata": { "self": "https://pkc-j581r8.us-west2.gcp.confluent.cloud/kafka/v3/clusters/lkc-ok51xj/links/link-1/mirrors/topic-2" }, "link_name": "link-1", "mirror_topic_name": "topic-2", "source_topic_name": "topic-2", "num_partitions": 6, "mirror_lags": [ { "partition": 5, "lag": 0, "last_source_fetch_offset": -1 }, { "partition": 4, "lag": 0, "last_source_fetch_offset": -1 }, { "partition": 3, "lag": 0, "last_source_fetch_offset": -1 }, { "partition": 2, "lag": 0, "last_source_fetch_offset": -1 }, { "partition": 1, "lag": 0, "last_source_fetch_offset": -1 }, { "partition": 0, "lag": 0, "last_source_fetch_offset": -1 } ], "mirror_status": "PENDING_STOPPED", "mirror_topic_error": "NO_ERROR", "state_time_ms": 1706820222493, "mirror_state_transition_errors": [ { "error_code": "AUTHENTICATION_ERROR", "error_message": "Failed to describe topic configs due to authentication issues." } ] } ``` ## Mirror topic deletion You can safely delete a mirror topic. Deleting a mirror topic permanently stops data mirroring to that topic. If you create a new normal topic of the same name on the same cluster, data is not mirrored to it. To delete a mirror topic, use the same command you would use to delete a normal topic: ```bash confluent kafka topic delete ``` To learn more, see [Delete topics that were auto-created](#deleting-auto-created-mirror-topics). #### 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. - You cannot delete a cluster link that is attached to any mirror topics. You must first delete, failover, or promote all of the mirror topics, and then you can delete the cluster link. ## Source topic deletion While it may be possible to delete a source topic that is being mirrored by a mirror topic and a cluster link, doing so is not recommended. In particular, unpredictable behavior can occur if a source topic is deleted, and a topic by the same name is then created within a few minutes. This scenario can cause permanent data loss on any mirror topics that are still mirroring from that source topic, and can also cause performance issues on the source cluster or destination cluster. Before deleting a source topic, you should stop any mirroring to associated mirror topics. You can stop mirroring on a mirror topic in one of these ways: - Delete the mirror topic. - Call `promote` or `failover` so the mirror topic enters the `STOPPED` state. - Revoke the security permissions for the cluster link to read the source topic. You can do this in one of three ways: (1) delete the cluster link’s `ALLOW` ACL for the source topic, (2) create a `DENY` ACL for the source topic, or (3) delete the cluster link’s API key. For further discussion about Kafka limitations with topic deletion and how topic IDs help, see [KIP-516](https://cwiki.apache.org/confluence/display/KAFKA/KIP-516%3A+Topic+Identifiers#KIP516:TopicIdentifiers-Motivation). ## How schemas work with mirror topics Cluster Linking preserves the schema ID stored in each message. Therefore, to consume from a mirror topic that is using schemas, the consumer clients must use a Schema Registry context with the same schema IDs as on the Schema Registry context used by the producers to the source topic. Consequently, to consume from a mirror topic that uses schemas, do one of the following: * Option 1: Use the same Schema Registry that the producers used. * Option 2: Use a Schema Registry context that was synced through [Schema Linking](/cloud/current/sr/schema-linking.html) from the Schema Registry that the producers used. ![Producers writing schema IDs to a source topic, and consumers reading the mirror topic through a Schema Registry context that holds the same schema IDs](multi-dc-deployments/cluster-linking/images/mirror-topics-and-schemas.png) To learn more about how Schema Registry supports disaster recovery scenarios, see [Manage Schema Linking in Disaster Recovery Failover Scenarios](../../schema-registry/schema-linking-cp.md#schema-linking-dr). ## Advanced mirror topic architectures The following patterns combine cluster links and mirror topics to build multi-hop and multi-destination topologies. ### Mirror topic chaining A mirror topic can be a source topic itself. A mirror topic can be mirrored by different cluster links, allowing you to “chain” cluster links and mirror topics together. For example, Topic A (source topic) on Cluster 1 is mirrored by a cluster link to Topic A on Cluster 2, where it acts as both a mirror topic and a source topic. A second cluster link mirrors it again to Topic A (mirror topic) on Cluster 3. ![A source topic mirrored to a second cluster, where the mirror topic acts as the source for a third cluster](multi-dc-deployments/cluster-linking/images/cluster-link-chain.png) ### Mirror topic fan-out A source topic can be mirrored to multiple mirror topics. These mirror topics must exist on multiple different clusters. For example, Topic A on Cluster 1 is mirrored by one cluster link to Topic A on Cluster 4, and by a second cluster link to Topic A on Cluster 5. ![One source topic on a single cluster mirrored by two cluster links to mirror topics on two separate destination clusters](multi-dc-deployments/cluster-linking/images/cluster-link-fan-out.png) ## Mirror topic configuration syncing The following sections provide a quick reference of which Cluster Linking configurations are synced from the source to the mirror topic, overrides, and concepts related to syncing. For a full configuration reference, see [Configure Cluster Linking on Confluent Platform](configs.md#cluster-link-configs). A reference for all available topic configurations is available at [Kafka Topic Configuration Reference for Confluent Platform](../../installation/configuration/topic-configs.md#cp-config-topics). ### Synced mirror topic configurations for Confluent Platform These configurations are always synced from the source topic to the mirror topic. Mirror topics always have the same value as their source topic, to ensure the properties of mirror topics are met. - number of partitions - `max.message.bytes` - `cleanup.policy` - `message.timestamp.type` By default, the following configurations also are synced from the source topic to the mirror topic unless they are explicitly removed from `topic.config.sync.include`, as described in the following section. - `retention.bytes` - `retention.ms` - `delete.retention.ms` - `min.compaction.lag.ms` - `max.compaction.lag.ms` Setting retention configurations to always sync keeps the source and destination data identical. With this default configuration, the starting offset is also synced from source to mirror topics. By maintaining consistent log start offsets, Cluster Linking guarantees that records deleted from the source cluster are also deleted from the destination cluster. This may be a regulatory requirement for some customers. ### Override default syncing to specify independent mirror topic behavior Some use cases require independent retention for source and destination topics. For example, when mirroring data from small edge clusters to large centralized clusters, low-footprint edge clusters may use very small retention, but rely on the data being available for a long time on the destination cluster. To satisfy these cases, you can override the defaults by explicitly setting the following property to specify only those topic configurations you want synced from source to destination: - `topic.config.sync.include` - The list of topic configurations to sync from the source topic. For example, the topic configurations could be set to the following (which does not include the retention properties): ```properties topic.config.sync.include=max.message.bytes,cleanup.policy,message.timestamp.type,message.timestamp.difference.max.ms,min.compaction.lag.ms,max.compaction.lag.ms ``` `topic.config.sync.include` is a cluster-link-level configuration and must be set when creating or updating the cluster link itself, not when creating individual mirror topics. If you encounter the error `Unknown topic config name: message.timestamp.difference.max.ms` when creating a link or during consumer offset syncing, remove `message.timestamp.difference.max.ms` from the link configuration `topic.config.sync.include`. To learn more, see [Known limitations and best practices](index.md#cluster-linking-limitations). Use `kafka-configs` to dynamically override other topic-level configurations on the existing cluster link, as described in [Creating a Mirror Topic](/platform/current/multi-dc-deployments/cluster-linking/commands.html#creating-a-mirror-topic) and [confluent kafka mirror create](https://docs.confluent.io/confluent-cli/current/command-reference/kafka/mirror/confluent_kafka_mirror_create.html). With these overrides in place, mirror topics have independent retention periods and starting offsets instead of syncing with their source topics. #### IMPORTANT Configuration overrides (like `topic.config.sync.include`) are specified at the cluster link level and apply to all mirror topics on the cluster. If independent retention is specified (by omission in `topic.config.sync.include`), you must either specify the retention value or use the Kafka defaults. ### Overridable mirror topic configurations You can override these configurations in Confluent Platform, which means you can set values on the mirror topic that differ from the source topic. - `segment.jitter.ms` - `segment.index.bytes` - `flush.messages` - `flush.ms` - `index.interval.bytes` - `min.cleanable.dirty.ratio` - `file.delete.delay.ms` - `preallocatemessage.format.version` - `confluent.segment.speculative.prefetch.enable` - `compression.type` ### Mirror topic configurations not synced for Confluent Platform Any configuration that is not in the preceding lists is not synced to a mirror topic in Confluent Platform. Therefore, the mirror topic configuration could be different from the source topic configuration. If you don’t override the mirror topic configuration, it inherits its cluster’s default. A few important examples of configurations that are not synced to mirror topics in Confluent Platform: - `min.insync.replicas` - `confluent.placement.constraints` - `confluent.tier.enable` - `confluent.key.schema.validation` - `confluent.value.schema.validation` - `replication.factor` ### Hybrid cloud configuration syncs Confluent Platform and Confluent Cloud have different policies for which mirror topic configurations are synced. If you cluster link between Confluent Platform and Confluent Cloud, the destination cluster’s policy applies. For example, if you cluster link from a Confluent Platform source cluster to a Confluent Cloud destination cluster, the value of `compression.type` is not synced. But if you cluster link from a Confluent Cloud source cluster to a Confluent Platform destination cluster, `compression.type` is synced. ## Related content - [Troubleshooting mirror topics](/platform/current/multi-dc-deployments/cluster-linking/trouble-cp.html#troubleshooting-mirror-topics) - [Cluster Linking for Confluent Platform](index.md#cluster-linking) - [Tutorial: Share Data Across Topics Using Cluster Linking for Confluent Platform](topic-data-sharing.md#tutorial-topic-data-sharing) - [Configure Cluster Linking on Confluent Platform](configs.md#cluster-link-configs) - [Kafka Topic Configuration Reference for Confluent Platform](../../installation/configuration/topic-configs.md#cp-config-topics)