# Configure and Manage Cluster Links on Confluent Cloud A cluster link is a connection between two Apache Kafka® clusters that enables data replication, consumer offset sync, and ACL synchronization across clusters. You can create, view, manage, and delete cluster links using the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html) and the [Confluent Cloud Cluster Linking (v3) REST API](https://docs.confluent.io/cloud/current/ccloud/cluster-linking-v-3/). ## Configure cluster link behavior Configure the following properties to control the behavior of a cluster link that goes to a Confluent Cloud destination cluster. These properties control features such as ACL synchronization, consumer offset migration, automatic mirror topic creation, and the starting position for topic mirroring. If you disable a feature that has filters after having it enabled initially, then any existing filters will be cleared or deleted from the cluster link. `acl.filters` : JSON string that lists the ACLs to migrate. Define the ACLs in a file, `acl.filters.json`, and pass the file name as an argument to `--acl-filters-json-file`. Populate `acl.filters` by passing a JSON file on the command line that specifies the ACLs. Do not include in the filter ACLs that are managed independently on the destination cluster. This is to prevent cluster link migration from deleting ACLs that were added specifically on the destination and should not be deleted. For more information, see [ACL syncing](index.md#cluster-link-acl-sync-limits-cc).
* Type: string * Default: `""` (Empty string) `acl.sync.enable` : Whether to migrate ACLs. See also, [Syncing ACLs from source to destination cluster](#cloud-cluster-link-acl-sync) and [ACL syncing](index.md#cluster-link-acl-sync-limits-cc).
* Type: boolean * Default: false `acl.sync.ms` : The ACL refresh interval in milliseconds when ACL migration is enabled. The default is 5000 milliseconds (5 seconds).
* Type: int * Default: 5000 `auto.create.mirror.topics.enable` : Whether to auto-create mirror topics based on topics on the source cluster. When set to “true”, mirror topics will be auto-created. Setting this option to “false” disables mirror topic creation and clears any existing filters. For details on this option, see [Auto-create mirror topics](mirror-topics-cc.md#auto-create-mirror-topics-concepts). `auto.create.mirror.topics.filters` : A JSON object with one property, `topicFilters`, that contains an array of filters to apply to indicate which topics should be mirrored. For details on this option, see [Auto-create mirror topics](mirror-topics-cc.md#auto-create-mirror-topics-concepts). `cluster.link.prefix` : A prefix that is applied to the names of the mirror topics. The same prefix is applied to consumer groups when [consumer.group.prefix](#consumer-group-prefix) is set to `true`. The prefix cannot be changed after the cluster link is created. To learn more, see “Prefixing Mirror Topics and Consumer Group Names” in [Mirror Topics](mirror-topics-cc.md#mirror-topics-concepts).
* Type: string * Default: null
#### NOTE - The prefix cannot be changed after the cluster link is created. - The [reverse](mirror-topics-cc.md#reverse-source-and-mirror-topic) commands do not support prefixed links for disaster recovery. If your architecture requires reversing mirroring relationships during a failover, do not configure a prefix on the link. `consumer.group.prefix.enable` : When set to `true`, the prefix specified for the [cluster link prefix](#cluster-link-prefix) is also applied to the names of consumer groups. The cluster link prefix must be specified in order for the consumer group prefix to be applied. To learn more, see “Prefixing Mirror Topics and Consumer Group Names” in [Mirror Topics](mirror-topics-cc.md#mirror-topics-concepts).
* Type: boolean * Default: false `consumer.offset.group.filters` : JSON array that specifies the consumer groups to migrate. To migrate consumer groups, you must configure both the `consumer.offset.group.filters` and `consumer.offset.sync.enable` properties. Specify the list of consumer groups in the `consumer.offset.group.filters` property, and set `consumer.offset.sync.enable` to `true` to enable it. To learn more, see [Migrate consumer groups from source to destination cluster](#cluster-link-migrate-consumer-groups) and [Sync consumer group offsets](mirror-topics-cc.md#mirror-topics-consumer-offsets).
* Type: string * Default: `""` (Empty string) `consumer.offset.sync.enable` : Whether to migrate consumer offsets from the source cluster. To migrate consumer groups, you must configure both `consumer.offset.group.filters` and `consumer.offset.sync.enable`. Specify the list of consumer groups in `consumer.offset.group.filters`, and set `consumer.offset.sync.enable` to `true` to enable it. To learn more, see [Migrate consumer groups from source to destination cluster](#cluster-link-migrate-consumer-groups) and [Sync consumer group offsets](mirror-topics-cc.md#mirror-topics-consumer-offsets).
* Type: boolean * Default: false `consumer.offset.sync.ms` : The consumer offset synchronization interval in milliseconds when consumer offset sync is enabled.
* Type: int * Default: 30000 `link.mode` : Indicates the direction of the cluster link, which can start from `DESTINATION` (default), `SOURCE` for a source-initiated link, or `BIDIRECTIONAL`. This setting cannot be changed once the link is created. This configuration is not available to set or create from the Confluent Cloud Console. The value for `link.mode` is not case-sensitive. For example, you can also set this as `destination`, `source`, or `bidirectional`.
* Type: string * Default: destination If overriding the default, the link must be used exactly as specified in [Bidirectional Mode Cluster Linking](#bidirectional-mode-cluster-linking), [Private to public Cluster Linking](private-networking.md#cluster-linking-private-to-public), or [Hybrid Cloud Cluster Linking](hybrid-cc.md#cluster-link-hybrid-cc). `mirror.start.offset.spec` : Whether to get the full history of a mirrored topic (`earliest`), only the `latest` messages, or the history starting at a given timestamp.
* Type: string * Default: earliest
- If set to a value of `earliest` (the default), new mirror topics will get the full history of their associated topics. - If set to a value of `latest`, new mirror topics will exclude the history and only replicate messages sent after the mirror topic is created. - If set to a timestamp in ISO 8601 format: `YYYY-MM-DDTHH:mm:SS.sss`, new mirror topics get the history of the topics starting from the timestamp.
When a mirror topic is created, it reads the value of this configuration and begins replication accordingly. If the setting is changed, it does not affect existing mirror topics; new mirror topics use the new value when they’re created.
If some mirror topics need to start from earliest and some need to start from latest, there are two options:
- Change the value of the cluster link’s `mirror.start.offset.spec` to the desired starting position before creating the mirror topic, or - Use two distinct cluster links, each with their own value for `mirror.start.offset.spec`, and create mirror topics on the appropriate cluster link as desired. `topic.config.sync.ms` : The topic configuration refresh interval in milliseconds.
* Type: int * Default: 5000 ## What you need to create a cluster link To create a cluster link, you need: - Access to the destination cluster. Cluster links are created on the destination cluster. For the specific security authorization needed, see [Manage Security for Cluster Linking on Confluent Cloud](security-cloud.md#cloud-cluster-link-security). - The name you wish to give the cluster link. - The source cluster’s bootstrap server and cluster ID. - If the source cluster is a Confluent Cloud cluster, you can get those using these commands: - `confluent kafka cluster list` to get the source cluster’s ID. For example, `lkc-12345` - `confluent kafka cluster describe ` to get the bootstrap server. It will be the value of the row called `Endpoint`. For example, `SASL_SSL://pkc-12345.us-west-2.aws.confluent.cloud:9092` - If the source cluster is a Confluent Platform or Apache Kafka® cluster, your system administrator can give you your bootstrap server. You can get your cluster ID by doing `confluent cluster describe` or `zookeeper-shell get /cluster/id 2> /dev/null`. For example, `0gXf5xg8Sp2G4FlxNriNaA`. - Authentication credentials for the cluster link to use with the source cluster. These can be updated, allowing you to rotate credentials used by your cluster link. See the [section on configuring security credentials](#cluster-link-configure-security-cc) for more details. - (Optional) Any additional configuration parameters you wish to add to the cluster link. See [the list of possible configurations](#configure-cluster-links-cc) below. You will then pass this information into the CLI or API command to create a cluster link. You cannot update the cluster link’s name, source cluster ID, or destination cluster ID. These values are only set at creation. ## Bidirectional mode Cluster Linking bidirectional mode (a bidirectional cluster link) enables better Disaster Recovery and active/active architectures, with data and metadata flowing bidirectionally between two or more clusters. ![image](multi-cloud/cluster-linking/images/cluster-link-bidirectional.png) ### Mental model A useful analogy is to consider a cluster link as a bridge between two clusters. - By default, a cluster link is a one-way bridge: topics go from a source cluster to a destination cluster, with data and metadata always flowing from source to destination. - In contrast, a bidirectional cluster link is a two-way bridge: topics on either side can go to the other cluster, with data and metadata flowing in both directions. In the case of a “bidirectional” cluster link, there is no “source” or “destination” cluster. Both clusters are equal, and can function as a source or destination for the other cluster. Each cluster sees itself as the “local” cluster and the other cluster as the “remote” cluster. ![image](multi-cloud/cluster-linking/images/cluster-link-bidirectional-both.png) ### Benefits Bidirectional cluster links are advantageous in disaster recovery (DR) architectures, and certain types of migrations, as described below. #### Disaster recovery Bidirectional cluster links are useful for [Disaster Recovery](/cloud/current/multi-cloud/cluster-linking/dr-failover.html#cloud-cluster-linking-dr), both active/passive and active/active. In a disaster recovery setup, two clusters in different regions are deployed so that at least one cluster is available at all times, even if a region experiences an outage. A bidirectional cluster link ensures both regions have the latest data and metadata from the other region, should one of them fail, or should applications need to be rotated from region to region for DR testing. It is easier to test and practice DR by moving producers and consumers to the DR cluster and reversing the direction of data and metadata, with fewer commands and moving pieces. For **active/passive** setups, a bidirectional link: - Gives access to the reverse and truncate-and-restore commands, which make for easier DR workflows with fast and efficient failover and failback. To learn more about these commands, see [Convert a mirror topic to a normal topic](mirror-topics-cc.md#convert-mirror-topic-to-normal-topic). - If you have any consumers on the DR site consuming from the mirror topics, their consumer offsets can be synced to the Primary site should you need to move them from DR to Primary. For **active/active** setups, a bidirectional link syncs all consumer offsets–from both regular topics and mirror topics–to both sides. Since consumers use a mix of regular and mirror topics, it is crucial to use a bidirectional link so that the consumer’s offsets are synced to the opposite side for failover. #### Consumer-last migrations Bidirectional cluster links are useful for certain types of migrations, where consumers are migrated after producers. - In most migrations from an old cluster to a new cluster, a default cluster link suffices because consumers are migrated before or at the same time as producers. - If there are straggling consumers on the old cluster, a bidirectional cluster link can help by ensuring their consumer offsets flow to the new cluster and are available when these consumers need to migrate. A default cluster link does not do this. ### Restrictions and limitations To use bidirectional mode for Cluster Linking, both clusters must be one of these types: - [Dedicated or Enterprise cluster on Confluent Cloud](/cloud/current/clusters/cluster-types.html) - [Confluent Platform 7.5 or later](/platform/current/installation/versions-interoperability.html) Bidirectional mode is not supported if either of the clusters is a Basic or Standard Confluent Cloud cluster, a version of Confluent Platform 7.4 or earlier, or open source Apache Kafka®. Consumer group prefixing cannot be enabled for bidirectional links. Setting `consumer.group.prefix.enable` to `true` on a bidirectional cluster link will result in an “invalid configuration” error stating that the cluster link cannot be validated due to this limitation. ### Security The cluster link will need one or more principal to represent it on each cluster, and those principals will be given cluster permissions via ACLs or RBAC, consistent with how authentication and authorization works for Cluster Linking. To learn more about authentication and authorization for Cluster Linking, see [Manage Security for Cluster Linking on Confluent Cloud](/cloud/current/multi-cloud/cluster-linking/security-cloud.html#cloud-cluster-link-security) and [Manage Security for Cluster Linking on Confluent Platform](/platform/current/multi-dc-deployments/cluster-linking/security.html) On Confluent Cloud, the same service account or identity pool can be used for both clusters, or two separate service accounts and identity pools can be used. #### Default security config for bidirectional connectivity By default, a cluster link in bidirectional mode is configured similar to the default configuration for two cluster links. ![image](multi-cloud/cluster-linking/images/cluster-link-bidirectional-security.png) Each cluster requires: - The ability to connect outbound to the other cluster. If this is not possible, see [Advanced options for bidirectional Cluster Linking](#cluster-linking-bidirectional-advanced). - A user to create a cluster link object on it with: - An authentication configuration, such as API key or OAuth, for a principal such as a user or service account on its remote cluster with ACLs or role-based access control (RBAC) role bindings giving permission to read topic data and metadata. - The Describe:Cluster ACL - The `DescribeConfigs:Cluster` ACL if consumer offset sync is enabled, which is recommended - The required ACLs or RBAC role bindings for a cluster link, as described in [Manage Security for Cluster Linking on Confluent Cloud](security-cloud.md#cloud-cluster-link-security) in the rows for a cluster link on a source cluster. - `link.mode=BIDIRECTIONAL` In some cases, only one cluster can reach the other. For example, if one of the clusters is in a private network or private datacenter, and the other is not. For details on how to configure a bidirectional link in this scenario, see [Advanced options for bidirectional Cluster Linking](#cluster-linking-bidirectional-advanced). The value for `link.mode` is not case-sensitive. For example, you can also set this to `bidirectional`. ### Create a cluster link in bidirectional mode The following sections explain how to create a cluster link in bidirectional mode using the Confluent CLI. The REST API can also be used with the supplied configurations. If both clusters can reach each other, follow these steps. If only one cluster can reach the other, see [Advanced options for bidirectional Cluster Linking](#cluster-linking-bidirectional-advanced). 1. Log in as a user with the necessary permissions to create a cluster link and service accounts. Typically, this requires an `OrganizationAdmin`, `EnvironmentAdmin`, or `CloudClusterAdmin`. 2. Decide which security and connection model to use. To learn about security and connection models for bidirectional Cluster Linking, see [Security](#bidirectional-cluster-link-security). 3. Locate the cluster IDs, optionally saving them for easy use. For example: ```bash confluent kafka cluster list ``` - Save cluster A ID in an environment variable. If you are using the unidirectional model, save that one as cluster A, corresponding with the descriptions in [Security](#bidirectional-cluster-link-security). Otherwise, either cluster can be cluster A. ```bash aID= ``` - Save cluster B ID in an environment variable. ```bash bID= ``` ![Terminal output showing cluster IDs from confluent kafka cluster list](multi-cloud/cluster-linking/images/cluster-link-bidirectional-examples-get-cluster-ids.png) 4. Locate your clusters’ bootstrap servers, optionally saving them for easy reuse: ```bash confluent kafka cluster describe $aID ``` ```bash aBootstrap= ``` ```bash confluent kafka cluster describe $bID ``` ```bash bBootstrap= ``` ![Terminal output showing bootstrap server endpoints from confluent kafka cluster describe](multi-cloud/cluster-linking/images/cluster-link-bidirectional-examples-describe-cluster-ids.png) 5. Create principals for the cluster link. If both clusters are Confluent Cloud clusters, create a service account as the principal for the cluster link. ```bash confluent iam service-account create --description "" ``` Take note of the service account ID. Optionally, save this ID for later: ```bash saID= ``` ![Terminal output showing the created service account ID](multi-cloud/cluster-linking/images/cluster-link-bidirectional-example-commands-save-service-account-ids.png) You can choose any name and description for the service account. If the clusters belong to different Confluent Cloud organizations, two service accounts will be needed. A service account cannot belong to or have permissions for more than one organization. When creating API keys and ACLs, use the service account from that organization. 6. Create an API key for the service account on both clusters. ```bash confluent api-key create --resource $aID --service-account $saID ``` ```bash confluent api-key create --resource $bID --service-account $saID ``` Take note of the API keys and secrets, optionally saving them for easy use: ```bash aApiKey= ``` ```bash aApiSecret= ``` ```bash bApiKey= ``` ```bash bApiSecret= ``` ![Terminal output showing the created API key and secret](multi-cloud/cluster-linking/images/cluster-link-bidirectional-example-commands-save-secret-keys.png) Alternatively, use OAuth for authentication. 7. Assign permissions to the service account on both clusters. In this example, create ACLs to allow the cluster link to read from topics beginning with the prefix “public.” on either cluster. To allow other topics, consumer offset sync, or ACL sync, add the ACLs listed in [Manage Security for Cluster Linking on Confluent Cloud](security-cloud.md#cloud-cluster-link-security). Or, use RBAC role bindings. ```bash confluent kafka acl create --allow --service-account $saID --operations read,describe-configs --topic "public." --prefix --cluster $aID ``` ```bash confluent kafka acl create --allow --service-account $saID --operations describe --cluster-scope --cluster $aID ``` ```bash confluent kafka acl create --allow --service-account $saID --operations read,describe-configs --topic "public." --prefix --cluster $bID ``` ```bash confluent kafka acl create --allow --service-account $saID --operations describe --cluster-scope --cluster $bID ``` ![Terminal output showing ACL creation for the service account](multi-cloud/cluster-linking/images/cluster-link-bidirectional-examples-assign-permissions-service-account.png)![Terminal output showing cluster-scope ACL creation](multi-cloud/cluster-linking/images/cluster-link-bidirectional-examples-assign-permissions-cluster-scope.png) 8. Create the cluster link object on cluster B, using the command [confluent kafka link create](https://docs.confluent.io/confluent-cli/current/command-reference/kafka/link/confluent_kafka_link_create.html). 1. Create a configuration file that includes the line `link.mode=BIDIRECTIONAL`. ```bash echo "link.mode=BIDIRECTIONAL">link-configs/bidirectional-link.config ``` Note that more [cluster link configurations](#configure-cluster-links-cc) can be added to this file if desired. These will be applied to data and metadata coming to cluster B only, and will not affect cluster A. 2. Issue the following CLI command to create a link named “bidirectional-link”. Name the link anything you want, but use identical link names on both clusters. ```bash confluent kafka link create bidirectional-link \ --remote-cluster $aID \ --cluster $bID \ --remote-api-key $aApiKey \ --remote-api-secret $aApiSecret \ --remote-bootstrap-server $aBootstrap \ --config link-configs/bidirectional-link.config ``` 9. Create the cluster link object on cluster A. 1. Use the same configuration file from cluster B, or create a new one with `link.mode=BIDIRECTIONAL`. Note that more [cluster link configurations](#configure-cluster-links-cc) can be added to this file if desired. These will be applied to data and metadata coming to cluster A only, and will not affect cluster B. 2. Issue the following CLI command to create a link named “bidirectional-link”. Name the link anything you want, but use identical link names on both clusters. ```bash confluent kafka link create bidirectional-link \ --remote-cluster $bID \ --cluster $aID \ --remote-api-key $bApiKey \ --remote-api-secret $bApiSecret \ --remote-bootstrap-server $bBootstrap \ --config link-configs/bidirectional-link.config ``` ![Terminal output showing the bidirectional cluster link creation command](multi-cloud/cluster-linking/images/cluster-link-bidirectional-example-create-link.png) 10. Create mirror topics on either cluster, using the same link name and standard create mirror topic process. ```bash confluent kafka topic create public.topic-on-a --cluster $aID ``` ```bash confluent kafka mirror create public.topic-on-a --link bidirectional-link --cluster $bID ``` ```bash confluent kafka topic create public.b-was-here --cluster $bID ``` ```bash confluent kafka mirror create public.b-was-here --link bidirectional-link --cluster $aID ``` ### Manage a cluster link in bidirectional mode Configurations for a cluster link in bidirectional mode can be changed using the [Confluent CLI](#cluster-link-manage-with-cli), the [REST API](#cluster-link-manage-with-rest-api), or the Confluent Cloud Console. Configurations are applied independently for clusters A and B. For example: - If consumer offset sync is only enabled on the cluster A cluster link object, then consumer offsets will only sync to A from B. No consumer offsets will be synced to B. - Conversely, to have consumer offsets sync to both clusters, it must be enabled on the cluster link objects on both clusters. - The same is true for other configuration options, including prefixing, auto-create mirror topics, and ACL sync. - Whether you use the CLI, REST API, or Confluent Cloud Console, editing configurations for one of the cluster link objects is done one at a time, with commands issued against that cluster. In the Confluent Cloud Console: - A bidirectional cluster link appears as two rows, one for each cluster. - To add mirror topics for a given cluster or to edit the configurations for that side of the cluster link, click the row that has destination cluster set as that cluster. #### Limitations on bidirectional cluster link management - The Confluent Cloud Console displays “Unknown” for the “Source Cluster” field in bidirectional cluster links, even when the remote cluster exists in Confluent Cloud. - Mirror topic names must be entered manually because the in-browser autocomplete does not work for bidirectional cluster links. ### Delete a cluster link in bidirectional mode To delete a cluster link that is in bidirectional mode: - Both cluster link objects on both clusters must be deleted. Delete them using the CLI, REST API, or Confluent Cloud Console. - Deleting the cluster link object on only one cluster will stop data from flowing either direction over the cluster link, but will not stop all billing charges, as some hourly charges may still occur on the remaining cluster link object. ### Advanced options for bidirectional Cluster Linking An advanced option for bidirectional Cluster Linking is a “unidirectional” security configuration for private-to-public or Confluent Platform to Confluent Cloud with a source-initiated link. In advanced situations, security requirements may require that only one cluster can reach the other and/or that security credentials be stored on only one cluster. This would be the case, for example, if one of the clusters has private networking or is located in a datacenter, and the other cluster is configured with Internet networking. In this scenario, configure the bidirectional cluster link so that only the more privileged private cluster connects to the remote cluster, and not the other way around. This is similar to a source-initiated cluster link. The following steps describe how to create a unidirectional, source-initiated cluster link. To learn more about the advantages of creating this type of link in bidirectional mode, see [What is the difference between unidirectional (source-initiated) and bidirectional (source-initiated) links?](../../clusters/cluster-faq.md#faq-unidirectional-vs-bidirectional-source-based-links) in the FAQ. ![Diagram of a bidirectional cluster link between a private cluster and a public cluster](multi-cloud/cluster-linking/images/cluster-link-bidirectional-advanced.png) #### Step 1. Configure the link on the public cluster The less privileged cluster, cluster B in the diagram above requires: - Create the inbound link on this less privileged cluster before its remote cluster, with `connection.mode=INBOUND` and `link.mode=BIDIRECTIONAL`. Otherwise, the links on both sides may end up with different IDs even though they may have the same names, and it is the link IDs that establish the connection between the two sides For example, create a `.config` file for cluster B as shown: ```bash cat link-configs/bidirectional-inbound.config link.mode=BIDIRECTIONAL connection.mode=INBOUND ``` Then, run the command to create an INBOUND only link on cluster B, including a call to your cluster B config file: ```bash confluent kafka link create bidirectional-link --remote-cluster $aID --environment $bEnvId --cluster $bId --remote-bootstrap-server $aBootstrap --config link-configs/bidirectional-inbound.config ``` #### Step 2. Configure the link on the private cluster The more privileged / private cluster, cluster A in the diagram, requires: - Connectivity to its remote cluster. One-way connectivity is acceptable, such as AWS PrivateLink. - A user to create a cluster link object on it second, after the remote cluster, with the following configuration: ```properties link.mode=BIDIRECTIONAL connection.mode=OUTBOUND security.protocol=SASL_SSL sasl.mechanism=PLAIN sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='' password=''; local.security.protocol=SASL_SSL local.sasl.mechanism=PLAIN local.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='' password=''; local.listener.name= ``` The `local.listener.name` is optional in single-listener setups, but recommended in multi-listener environments. Rather than allowing the cluster link to infer a default listener, specify the exact network listener name on the private cluster to handle the incoming connection. In multi-listener scenarios, specifying `local.listener.name` ensures predictable network routing and prevents traffic drift if cluster configurations are updated. - An authentication configuration, such as API key or OAuth, for a principal such as a user or service account on its remote cluster with ACLs or RBAC role bindings giving permission to read topic data and metadata. - Alter:Cluster ACL – this is used in the advanced mode and for the `reverse` and `truncate-and-restore` commands - Describe:Cluster ACL - The required ACLs or RBAC role bindings for a cluster link, as described in [Manage Security for Cluster Linking on Confluent Cloud](security-cloud.md#cloud-cluster-link-security) in the rows for a cluster link on a source cluster and for a source-initiated link on the destination cluster. - The required ACLs or RBAC role bindings as described in [Manage Security for Cluster Linking on Confluent Cloud](security-cloud.md#cloud-cluster-link-security) (the rows for a cluster link on a source cluster. This authentication configuration never leaves this cluster). For example, run the following command to create a bidirectional INBOUND/OUTBOUND link on the private cluster, cluster A in the diagram, including the call to your cluster A configuration file: ```bash confluent kafka link create bidirectional-link --remote-cluster $bId --environment $aEnvId --cluster $aID --remote-api-key $bApiKey --remote-api-secret $bApiSecret --remote-bootstrap-server $bBootstrap --config link-configs/bidirectional-outbound.config ``` ## Managing cluster links with the CLI To create or configure a cluster link with the [Confluent Cloud CLI](https://docs.confluent.io/confluent-cli/current/overview.html), create a configuration file with the configurations you wish to set. Each configuration must be on one line in the file, with the format `key=value`, where `key` is the name of the configuration, and `value` is the value you wish to set on that cluster link. In the following commands, the location of this file is referred to as ``. ### Create a cluster link To create a cluster link with the `confluent` CLI, use this command on the destination cluster: ```bash confluent kafka link create \ --source-cluster \ --source-bootstrap-server \ --config \ --cluster ``` `--source-cluster-id` was replaced with `--source-cluster` in version 3 of the Confluent CLI, as described in the command reference for [confluent kafka link create](https://docs.confluent.io/confluent-cli/current/command-reference/kafka/link/confluent_kafka_link_create.html). ### Create a source-initiated cluster link By default, cluster links are destination-initiated, meaning the destination cluster reaches out to the source cluster to establish the connection and pull data. You can also create a source-initiated cluster link, where the source cluster initiates the network connection to the destination cluster. This pattern is particularly useful when security configurations, firewall rules, or private networking constraints prevent the destination cluster from reaching into the source environment. 1. Create the inbound link on the destination cluster. ```bash confluent kafka link create \ --cluster \ --config connection.mode=INBOUND \ --source-cluster ``` When a link is in INBOUND mode, it sits in a disconnected state, waiting for the source cluster to reach out and match the configuration. 2. Prepare the configuration file for the source cluster. The source cluster needs to know how to connect and authenticate to the destination cluster. Create a local configuration file named `source-link-config.properties` containing the destination cluster’s details and API credentials: ```properties # Connection mode for the source side of a source-initiated link connection.mode=OUTBOUND # Destination cluster bootstrap server link.destination.bootstrap.servers=:9092 # Security configuration to authenticate against the destination cluster link.destination.security.protocol=SASL_SSL link.destination.sasl.mechanism=PLAIN link.destination.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="" password=""; ``` 3. Create the outbound link on the source cluster. Next, establish the actual connection by creating the matching link on the source cluster using OUTBOUND mode and passing the configuration file you created in Step 2. Run this command against your source cluster: ```bash confluent kafka link create \ --cluster \ --config source-link-config.properties ``` The `` must be exactly identical on both the source and destination clusters for the handshake to succeed. 4. Verify the link state. After both sides are created, the source cluster actively initiates the handshake. Verify that the link state has transitioned to `ACTIVE` by running: ```bash confluent kafka link describe ``` Confirm that the `State` column reads `ACTIVE`. You are now ready to mirror topics across the source-initiated link. ### Update a cluster link To update the configuration for an existing cluster link with the `cloud` CLI, use this command on the destination cluster: ```bash confluent kafka link configuration update --config my-update-configs.txt --cluster ``` Follow these guidelines: - When updating the configuration for an existing cluster link, include only the configurations that change. Be mindful when you use `--config` to specify a configuration file that it contains only the configurations you want to update. For example, `my-update-configs.txt` might include: ```properties consumer.offset.sync.ms=25384 topic.config.sync.ms=38254 ``` - You can change several aspects of a cluster link configuration, but you cannot change its source cluster, ID, prefix, or link name. - Examples of creating and configuring cluster links with the CLI are shown in the [Cluster Linking Quick Start on Confluent Cloud](quickstart.md#cloud-cluster-link-quickstart), and the tutorials for [Share Data Across Clusters, Regions, and Clouds using Confluent Cloud](topic-data-sharing.md#cloud-cluster-link-topic-data-sharing) and [Cluster Linking Disaster Recovery and Failover on Confluent Cloud](dr-failover.md#cloud-cluster-linking-dr). ### List all cluster links going to the destination cluster To see a list of all cluster links going to a destination cluster, use the command: ```bash confluent kafka link list --cluster ``` The output for this command will include the names of existing links; for example: ```bash Name | Source Cluster | Destination Cluster | State | Error | Error Message ----------+----------------+---------------------+--------+-------+---------------- my-link | lkc-1225nz | | ACTIVE | | ``` ### View the configuration for a given cluster link To view the configuration of a specific cluster link, use the command: ```bash confluent kafka --cluster link configuration list ``` ### View a cluster link task status You can view the status of the following configurable tasks: - [Sync consumer group offsets](mirror-topics-cc.md#mirror-topics-consumer-offsets) - [ACL sync](#cloud-cluster-link-acl-sync) - [Topic configurations sync](mirror-topics-cc.md#sync-topic-configs) - [Auto-create mirror topics](mirror-topics-cc.md#auto-create-mirror-topics-concepts) To view the status of any given task, use the following command: ```bash confluent kafka link task list ``` For example, to list tasks: ```bash confluent kafka link task list link-1 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 | ``` 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. | ### Delete a cluster link To delete a cluster link, use the command: ```bash confluent kafka link delete --cluster ``` Using the command `confluent kafka cluster use ` sets the destination cluster to the active cluster, so you won’t need to specify `--cluster ` in any commands. #### IMPORTANT - Before deleting a cluster link, verify 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, the history of any `STOPPED` topics is also deleted. If you need the `Last Source Fetch Offset` or the `Status Time` of your promoted or failed-over mirror topics, save those before you delete the cluster 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, 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#). - Cluster links are automatically deleted if the destination cluster is deleted. ## Managing cluster links with the REST API REST API calls and examples are provided in the reference documentation for the [Confluent Cloud Cluster Linking (v3) REST API](https://docs.confluent.io/cloud/current/ccloud/cluster-linking-v-3/). To learn how to authenticate with the REST API, see [the REST API documentation](https://docs.confluent.io/cloud/current/api.html/#authentication). To configure cluster links with the Confluent Cloud REST API, pass in configs in JSON format. Each config will take this shape: ```json { "name": "", "value": "" } ``` Since this is JSON, if your config value is a string, surround it in double quotes. Then, pass in the name/value pairs as an array to the `configs` argument of the JSON payload. For example, this JSON enables consumer offset sync for all consumer groups and ACL sync for all ACLs. ```json { "configs": [ { "name": "consumer.offset.sync.enable", "value": "true" }, { "name": "consumer.offset.group.filters", "value": "{\"groupFilters\": [{\"name\": \"*\",\"patternType\": \"LITERAL\",\"filterType\": \"INCLUDE\"}]}" }, { "name": "acl.sync.enable", "value": "true" }, { "name": "acl.filters", "value": "{ \"aclFilters\": [ { \"resourceFilter\": { \"resourceType\": \"any\", \"patternType\": \"any\" }, \"accessFilter\": { \"operation\": \"any\", \"permissionType\": \"any\" } } ] }" } ] } ``` ### Creating a cluster link through the REST API See [Create a cluster link](https://docs.confluent.io/cloud/current/ccloud/create-kafka-link/) in the Confluent Cloud REST API documentation. To create a cluster link using the REST API, the JSON payload needs: - An entry called `source_cluster_id` set to your source cluster’s ID - A configuration in the `configs` array called `bootstrap.servers` set to your source cluster’s bootstrap server - Your cluster link’s [security configuration](#cluster-link-configure-security-cc) listed as configs in the `configs` array The [security configuration](#cluster-link-configure-security-cc) must include the API key and secret resource-scoped for the cluster against which you are running the REST API commands. For more information on creating an API key, see [Add an API key](../../security/authenticate/workload-identities/service-accounts/api-keys/manage-api-keys.md#create-api-key). - (Optional) Additional configurations in the `configs` array Example JSON payload for a request to create a cluster link that syncs consumers offsets for all consumer groups, and syncs all ACLs: ```json { "source_cluster_id": "", "configs": [ { "name": "security.protocol", "value": "SASL_SSL" }, { "name": "bootstrap.servers", "value": "" }, { "name": "sasl.mechanism", "value": "PLAIN" }, { "name": "sasl.jaas.config", "value": "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"\" password=\"\";" } ] } ``` Include this JSON payload in a POST request to: `/kafka/v3/clusters//links/?link_name=` Where you replace the following: - `` is your cluster’s REST url. You can find this with `confluent kafka cluster describe `. It will look like: `https://pkc-XXXXX.us-west1.gcp.confluent.cloud:443` - `` is your destination cluster’s ID. You can find this with `confluent kafka cluster list`. It will look like `lkc-XXXXX`. - `` is the name of your cluster link. You can name it whatever you choose. ### Updating and viewing cluster link configs with the REST API #### Update an existing cluster link See [Alter the config under a cluster link](https://docs.confluent.io/cloud/current/ccloud/update-kafka-link-config/) in the Confluent Cloud REST API documentation. To update an existing link, you will send the JSON payload described above as a PUT request to `/clusters//links//configs:alter`. #### Get the value of a cluster link configuration See [Describe a cluster link](https://docs.confluent.io/cloud/current/ccloud/get-kafka-link/) in the Confluent Cloud REST API documentation. To get the value of your cluster link’s configurations using the REST API, use one of these options: - GET `/clusters//links//configs` to get the full list of configs. - GET `/clusters//links//` to get the value of a specific config. #### List all cluster links with the REST API See [List all cluster links in the destination cluster](https://docs.confluent.io/cloud/current/ccloud/list-kafka-links/) in the Confluent Cloud REST API documentation. To view a list of all cluster links going to a destination cluster, send a GET request to `/clusters//links`. #### View a cluster link task status with the REST API See [Describe a cluster link](https://docs.confluent.io/cloud/current/ccloud/get-kafka-link/) in the Confluent Cloud REST API documentation. To view a cluster link task status, send a GET request to `/clusters//links/?include_tasks=true`. For example, to list tasks: ```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?include_tasks=true' | jq % Total % Received % Xferd Average Speed Time Time Time Current Dload Upload Total Spent Left Speed 100 839 0 839 0 0 1727 0 --:--:-- --:--:-- --:--:-- 1729 { "kind": "KafkaLinkData", "metadata": { "self": "https://pkc-j581r8.us-west2.gcp.confluent.cloud/kafka/v3/clusters/lkc-ok51xj/links/link-1" }, "source_cluster_id": null, "destination_cluster": null, "remote_cluster_id": "lkc-038qdq", "link_name": "link-1", "link_id": "d27ab324-82e8-4806-8487-1c6dcb3377dc", "cluster_link_id": "0nqzJILoSAaEhxxtyzN33A", "topic_names": [], "link_error": "NO_ERROR", "link_error_message": "", "link_state": "ACTIVE", "tasks": [ { "task_name": "ConsumerOffsetSync", "state": "IN_ERROR", "errors": [ { "error_code": "REMOTE_LINK_NOT_FOUND_ERROR", "error_message": "Failed to get remote link config due to link not being found on the remote cluster." } ] }, { "task_name": "AclSync", "state": "NOT_CONFIGURED", "errors": [] }, { "task_name": "TopicConfigsSync", "state": "ACTIVE", "errors": [] }, { "task_name": "AutoCreateMirror", "state": "NOT_CONFIGURED", "errors": [] } ] } ``` To learn more about viewing task statuses, see [View a cluster link task status](#cluster-link-task-status-commands) in the CLI section. #### Delete a cluster link with the REST API See [Delete the cluster link](https://docs.confluent.io/cloud/current/ccloud/delete-kafka-link/) in the Confluent Cloud REST API documentation. To delete a cluster link, send a DELETE request to `/clusters//links/`. #### IMPORTANT - Before deleting a cluster link, verify 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, the history of any `STOPPED` topics is also deleted. If you need the `Last Source Fetch Offset` or the `Status Time` of your promoted or failed-over mirror topics, save those before you delete the cluster link. - You cannot directly call the “delete cluster link” command on a cluster link that still has mirror topics on it. The command 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#). - Cluster links are automatically deleted if the destination cluster is deleted. ## Managing cluster links with Terraform Create a cluster link by using the Confluent Terraform provider. For more information, see [confluent_cluster_link Resource](https://registry.terraform.io/providers/confluentinc/confluent/latest/docs/resources/confluent_cluster_link). ## Migrate 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 configuration when creating the link with [confluent kafka link create](https://docs.confluent.io/confluent-cli/current/command-reference/kafka/link/confluent_kafka_link_create.html) or use [confluent kafka link configuration update](https://docs.confluent.io/confluent-cli/current/command-reference/kafka/link/configuration/confluent_kafka_link_configuration_update.html) to update an existing link. 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 a source cluster to a destination cluster and the state before the migration is executed is that you are currently migrating 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. Verify that Cluster Linking replication is beyond the latest committed offset. Confirm this with the following commands. - Check the CURRENT-OFFSET on the source cluster. ```bash confluent kafka consumer group describe someGroup --cluster ``` 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 confluent kafka consumer group describe someGroup --cluster ``` 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 confluent kafka consumer group describe someGroup --cluster ``` 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 confluent kafka consumer group describe someGroup --cluster ``` 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. Create a configuration file with the updated filters: ```bash echo 'consumer.offset.group.filters={"groupFilters": [ { "name": "*", "patternType": "LITERAL", "filterType": "INCLUDE" }, { "name": "someGroup", "patternType": "LITERAL", "filterType": "EXCLUDE" } ]}' > newFilters.txt ``` Update the cluster link configuration: ```bash confluent kafka link configuration update --config newFilters.txt --cluster ``` 6. Start the consumer on the destination cluster. ### 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 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). #### What is 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 failover from the source cluster to the destination cluster where the mirror topics are located. `LOCAL_MIRROR` determines the exact offset a consumer should resume from when failing over from a remote source topic to a local mirror topic. #### What is 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 Specify `LOCAL_MIRROR` only in the offset sync configuration on the 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 file, `dr-link.config`, for the cluster link on the DR cluster: ```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 ``` 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 since 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 - **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-west", "patternType": "LITERAL", "filterType": "INCLUDE", "topicTypes": ["REMOTE_MIRROR", "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-east", "patternType": "LITERAL", "filterType": "INCLUDE", "topicTypes": ["REMOTE_MIRROR", "LOCAL_MIRROR"]}]} ``` With this configuration: - `cg-east` offsets for the `orders` and `west.orders` topics on cluster A are synced to cluster B. - `cg-west` offsets for the `orders` and `east.orders` topics on cluster B are synced to cluster A. - Each consumer group’s offsets flow in one direction only, preventing cycles. ##### 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 task list` command to verify that consumer offset sync is working correctly. Look for the `ConsumerOffsetSync` task in `ACTIVE` state. #### 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 destination cluster by at least double `offsets.retention.check.interval.ms`. #### IMPORTANT If you use bidirectional linking to sync the same consumer group between both sides, retention gets extended on both clusters in a loop, since each side’s retention delay defers to the other’s. 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). #### To learn more - [Migrate consumer groups from source to destination cluster](#cluster-link-migrate-consumer-groups) - [Bidirectional Mode Cluster Linking](#bidirectional-mode-cluster-linking) - [Considerations for failover scenarios](mirror-topics-cc.md#considerations-for-failover-scenarios-offset-configs) in the Mirror Topics documentation ## Syncing ACLs from source to destination cluster Cluster Linking can sync some or all of the ACLs on the source cluster to the destination cluster. This is critical for Disaster Recovery (DR) architectures, so that when your client applications failover to the DR cluster, they have the same access that they had on their original cluster. When you create the cluster link, specify which ACLs to sync, filtering down by the name of the resource, principal for a service account, operation, and/or permission. You can update this setting on the cluster link at any time. The cluster link will copy the initial set of matching ACLs, and then continuously sync any changes, additions, or deletions of matching ACLs from the source cluster to the destination cluster. By default, ACLs on the source that were themselves created by Cluster Linking are not synced. Override this behavior with another filter. To learn more, see [Sync ACLs created by a cluster link](#sync-acls-created-by-acl-sync) in the Examples section. In Confluent Cloud, ACL sync is only supported between two Confluent Cloud clusters that belong to the same Confluent Cloud organization. ACL sync is not supported between two Confluent Cloud clusters in different organizations, or between a Confluent Platform and a Confluent Cloud cluster. Do not include in the sync filter ACLs that are managed independently on the destination cluster. Before configuring and deploying with ACL sync, see [ACL syncing](index.md#cluster-link-acl-sync-limits-cc) limitations in the overview for full details. To learn more, see [Migrate consumer groups from source to destination cluster](#cluster-link-migrate-consumer-groups). ### How does ACL sync work? ACL sync creates an exact replica of the ACLs from the source. ACL sync is capable of syncing ACLs for multiple resource types. When enabled, it typically syncs topics, consumer groups, and transactional IDs. ### Prerequisites - **Authorizations**: You must have the appropriate authorizations: - DESCRIBE Cluster ACLs (`DescribeAcls` API) on the source cluster - ALTER Cluster ACLs (`CreateAcls`/ `DeleteAcls` APIs) on the destination cluster - **Cluster link prefixing cannot be used with ACL sync**: ACL syncing cannot be used in cluster links that are configured with a prefix, defined as `cluster.link.prefix`. For more details, see [Limitations on prefixing](mirror-topics-cc.md#cluster-link-prefix-limitations-overview). ### Configurations for ACL sync To enable ACL sync on a cluster link, specify these properties: `acl.sync.enable=true` : This turns on ACL sync. Updating this config to `false` will turn off ACL sync.
If you set this up and run Cluster Linking, then later disable it, the filters will be cleared (deleted) from the cluster link. `acl.filters` : A JSON object with one property, `aclFilters`, that contains an array of filters to include specific ACLs to sync. Examples are below. `acl.sync.ms (optional)` : How frequently to sync ACLs. The default is 5000 (5 seconds). Minimum is 1000. Configure ACLs at the time you create the cluster link, or as an update to an existing configuration. Here is an example of setting up a cluster link with ACL migration when you create the cluster link. Note that the link configurations, including ACL migration properties, are defined in a file, `link-configs.properties`, rather than specified directly on the command line. ```properties confluent kafka link create from-aws-basic \ --source-cluster lkc-12345 \ --source-bootstrap-server SASL_SSL://pkc-abcde.us-west1.gcp.confluent.cloud:9092 \ --source-api-key 1L1NK2RUL3TH3M4LL \ --source-api-secret ******* \ --config link-configs.properties ``` Here is an example of what you might have in `link-configs.properties`: ```properties acl.sync.enable=true acl.sync.ms=1000 acl.filters={ "aclFilters": [ { "resourceFilter": { "resourceType": "any", "patternType": "any" }, "accessFilter": { "operation": "any", "permissionType": "any" } } ] } ``` The following sections provide examples of how you might define the actual ACLs in `acl.filters`. ### Examples The following examples show how to configure the JSON for various types of ACL migration, with granularity on a topic, a resource, or a mixed set. #### Migrate all ACLs from source to destination cluster To migrate all ACLs from the source to the destination cluster, provide the following for `acl.filters`. ```properties acl.filters={ \ "aclFilters": [ \ { \ "resourceFilter": { \ "resourceType": "any", \ "patternType": "any" \ }, \ "accessFilter": { \ "operation": "any", \ "permissionType": "any" \ } \ } \ ] \ } ``` Each field in the JSON has the following options: - `resourceType`: Can be `ANY` (meaning any kind of resource), `topic`, `cluster`, `group`, or `transactionalID`, which are the resources specified in [ACL Overview](../../security/access-control/acls/overview.md#acl-manage). - `patternType`: Can be `ANY`, `LITERAL`, or `MATCH`. Note that while `PREFIXED` is a valid pattern type option, it is not supported in the context of ACL syncing in Cluster Linking. For more details, see [Limitations on filtering ACLs](#cluster-link-acl-sync-limitations-prefixing). - `name`: Name of resource. - If left empty, will default to the `*` wildcard, which will match all names of the specified `resourceType`. - If `patternType` is `"LITERAL"` then any resources with this exact name will be included. - If `patternType` is `"MATCH"`, then any resources that match a given literal or wildcard (`*`) pattern of the same name and type will be included. - `principal`: (optional) Name of principal specified on the ACL. If left empty will default to the `*` wildcard, which will match all principals with the specified `operation` and `permissionType`. - For example, a service account with ID 12345 has this principal: `User:sa-12345`. - `operation`: Can either be any or one of those specified in [Operations](/platform/current/kafka/authorization.html#operations) under the “Operation” column. - `permissionType`: Can be `any`, `allow`, or `deny`. ##### Limitations on filtering ACLs In Cluster Linking, ACL sync does not support prefix-pattern resource filtering. For example, using `"patternType": "PREFIXED"` to sync all ACLs for topics starting with `orders-` is unsupported. Instead, use the following alternatives to search for prefixed resources using ACL filters: - Configure ACL filters using the `LITERAL` pattern type, and define the exact name of each of the desired prefixed resources in each filter. - Configure ACL filters that each define a specific `principal` with access to the desired prefixed resources. - This alternative is best when ACLs for the desired prefixed resources are all assigned to the same principals. For example, if you want to sync read ACLs for topics `orders-us`, `orders-eu`, `orders-apac`, and `orders-latam` that are all assigned to the same principal, you would define a separate filter entry for each topic: ```properties acl.filters={ "aclFilters": [ { "resourceFilter": { "resourceType": "topic", "patternType": "LITERAL", "name": "orders-us" }, "accessFilter": { "principal": "User:sa-12345", "operation": "read", "permissionType": "allow" } }, { "resourceFilter": { "resourceType": "topic", "patternType": "LITERAL", "name": "orders-eu" }, "accessFilter": { "principal": "User:sa-12345", "operation": "read", "permissionType": "allow" } }, { "resourceFilter": { "resourceType": "topic", "patternType": "LITERAL", "name": "orders-apac" }, "accessFilter": { "principal": "User:sa-12345", "operation": "read", "permissionType": "allow" } }, { "resourceFilter": { "resourceType": "topic", "patternType": "LITERAL", "name": "orders-latam" }, "accessFilter": { "principal": "User:sa-12345", "operation": "read", "permissionType": "allow" } } ] } ``` #### Sync all ACLs specific to a topic To sync all ACLs for topic `pineapple`, provide the following configurations in `acl.filters`: ```properties acl.filters={ \ "aclFilters": [ \ { \ "resourceFilter": { \ "resourceType": "topic", \ "patternType": "literal", \ "name": "pineapple” \ }, \ "accessFilter": { \ "operation": "any", \ "permissionType": "any" \ } \ } \ ] \ } ``` #### Sync ACLs specific to a principal or permission To sync all ACLs specific to a service account with ID 12345 and also all ACLs that deny access to the cluster, provide the following configurations in `acl.filters`: ```properties acl.filters={ \ "aclFilters": [ \ { \ "resourceFilter": { \ "resourceType": "any", \ "patternType": "any" \ }, \ "accessFilter": { \ “principal”: “User:sa-12345”, \ "operation": "any", \ "permissionType": "any" \ } \ }, \ { \ "resourceFilter": { \ "resourceType": "any", \ "patternType": "any" \ }, \ "accessFilter": { \ "operation": "any", \ "permissionType": "deny" \ } \ } \ ] \ } ``` #### Sync ACLs created by a cluster link By default, ACLs on the source that were themselves created by Cluster Linking are not synced. For example, suppose you have three clusters: cluster-1, cluster-2, and cluster-3. Link-1 syncs ACLs from cluster-1 to cluster-2, and link-2 syncs ACLs from cluster-2 to cluster-3. If link-1 creates ACLs on cluster-2, by default link-2 does not migrate those ACLs to cluster-3 even if the filter is `any`. cluster-3 by link-2 even if the filter is `any`. You can override this by using two filters: one that supports standard ACL syncing, and another that syncs ACLs created by a cluster link, such as “link-1” in the following example. - The first is the regular type of filter that syncs all, except those created by any cluster link. - The second makes sure that the ACLs created by link-1 get synced. ```bash [ { "resourceFilter": { "resourceType": "any", "patternType": "any" }, "accessFilter": { "operation": "any", "permissionType": "any" } }, { "resourceFilter": { "resourceType": "any", "patternType": "any" }, "accessFilter": { "operation": "any", "permissionType": "any", "clusterLinkIds ": [ "link-1-ID" ] } } ] ``` ## Configure security credentials When you create a cluster link, give the cluster link credentials to authenticate to the source cluster. [Read the Cluster Linking security overview](security-cloud.md#cloud-cluster-link-security) for more information about Cluster Linking security requirements, including the required ACLs. ### Configure security credentials for Confluent Cloud source clusters If your source cluster is a Confluent Cloud cluster, your cluster link needs these properties configured: - `security.protocol` set to `SASL_SSL` - `sasl.mechanism` set to `PLAIN` - `sasl.jaas.config` set to `org.apache.kafka.common.security.plain.PlainLoginModule required username="" password="";` - Replace `` and `` with the API key to use on your source cluster. - Be careful not to forget the `"` characters and the ending `;` ### Configure security credentials for Confluent Platform and Kafka source clusters The cluster link’s authentication configuration is identical to a regular Kafka consumer’s, because cluster links consume from the source cluster using the same mechanism. For Confluent Cloud cluster links, if your source cluster uses SSL, also known as TLS, pass the keystore and truststore in-line, rather than with a keystore and truststore location. To learn more, see the security section on [mutual TLS (mTLS) authentication](security-cloud.md#cloud-cluster-linking-mtls). The following properties are supported by cluster links going to Confluent destination clusters: - `sasl.client.callback.handler.class` - `sasl.jaas.config` - `sasl.kerberos.kinit.cmd` - `sasl.kerberos.min.time.before.relogin` - `sasl.kerberos.service.name` - `sasl.kerberos.ticket.renew.jitter` - `sasl.kerberos.ticket.renew.window.factor` - `sasl.login.callback.handler.class` - `sasl.login.class` - `sasl.login.refresh.buffer.seconds` - `sasl.login.refresh.min.period.seconds` - `sasl.login.refresh.window.factor` - `sasl.login.refresh.window.jitter` - `sasl.mechanism` - `security.protocol` - `ssl.cipher.suites` - `ssl.enabled.protocols` - `ssl.endpoint.identification.algorithm` - `ssl.engine.factory.class` - `ssl.key.password` - `ssl.keymanager.algorithm` - `ssl.keystore.location` - `ssl.keystore.password` - `ssl.keystore.type` - `ssl.keystore.certificate.chain` - `ssl.protocol` - `ssl.provider` - `ssl.truststore.certificates` ## Related content - [Confluent CLI reference](https://docs.confluent.io/confluent-cli/current/command-reference/) - [Confluent Cloud Cluster Linking (v3) REST API](https://docs.confluent.io/cloud/current/ccloud/cluster-linking-v-3/) - [Manage Security for Cluster Linking on Confluent Cloud](security-cloud.md#cloud-cluster-link-security) - [Tutorials](index.md#cloud-cluster-link-tutorials-list)