# Configure Cluster Linking on Confluent Platform Cluster Linking is a feature for replicating topics between Apache Kafka® clusters for disaster recovery, data sharing, and multi-region deployments. This page describes how to configure Cluster Linking with various Confluent tools, products, and security options. Before configuring Cluster Linking, refer to the general Confluent Platform configuration options that are not specific to Cluster Linking. ## Using Cluster Linking with Confluent for Kubernetes You can use Cluster Linking with Confluent Platform deployed with [Confluent for Kubernetes](https://docs.confluent.io/operator/current/overview.html). Confluent for Kubernetes 2.2 released built-in Cluster Linking support, as described in this section of the CFK documentation: [Cluster Linking using Confluent for Kubernetes](https://docs.confluent.io/operator/2.2.0/co-link-clusters.html#co-link-clusters). To configure Cluster Linking on earlier versions of CFK, use `configOverrides` in the Kafka custom resource. See [Configuration Overrides](https://docs.confluent.io/operator/current/co-configure-misc.html#configuration-overrides) in the CFK documentation for more information about using `configOverrides`. Also, pre Confluent Platform 7.0.0 releases required that you include a `configOverrides` section on the server to specify `confluent.cluster.link.enable: "true"`. For Confluent Platform 7.0.0 and later releases, Cluster Linking is enabled by default, so this element of the configuration is no longer needed, regardless of the Confluent for Kubernetes version. For example: ```bash apiVersion: platform.confluent.io/v1beta1 kind: Kafka metadata: name: kafka namespace: confluent spec: replicas: 3 image: application: confluentinc/cp-server:8.3.1 init: confluentinc/confluent-init-container:2.0.1 configOverrides: server: - confluent.cluster.link.enable=true # Enable Cluster Linking ``` ## Using Cluster Linking with Ansible You can use Cluster Linking with Confluent Platform deployed with [Ansible](https://docs.confluent.io/ansible/current/overview.html). Starting in Confluent Platform 7.0.0, Cluster Linking is enabled by default, so no changes are needed to the configuration file. However, in IPv6-based clusters, you must add the following JVM configuration to your `kafka_broker` for Cluster Linking to work: ```properties -Djava.net.preferIPv6Addresses=true ``` ## Cluster link configuration properties Several configurations are available for cluster links. The following sections describe how to set these using the [CLI commands](commands.md#bootstrap-cluster-links), and then list the available properties. For a full list of configuration options, see [Configuration options specific to cluster links](#cp-cluster-link-config-options) and [Common Apache Kafka configuration options](#common-config-options-cluster-linking). Cluster Linking configurations are isolated from Kafka broker configurations. As such, there is no property inheritance from Kafka broker to Cluster Linking. Only properties passed during cluster links create and update will override Cluster Linking behavior. ### Setting Properties on a Cluster Link You can set configurations on each individual cluster link. To do this, provide the configurations as “key=value” pairs in a properties file and pass the file as an argument to the [CLI commands](commands.md#bootstrap-cluster-links) using: - The `--config-file` flag, when you first create the link - The `--add-config-file` flag to update configurations on an existing link - The `--config-file` flag, when you first create the link - The `--add-config-file` flag to update configurations on an existing link Alternatively, you can specify or update properties for the cluster link by providing “key=value” pairs directly on the command line, using: - The `--config` flag when you first create the link - The `--add-config` flag to update configurations on an existing link - The `--config` flag when you first create the link - The `--add-config` flag to update configurations on an existing link Examples and command syntax for specifying link properties in a file and at the command line are shown in [Creating a cluster link](commands.md#cluster-link-create-command) and in [Altering a cluster link](commands.md#cluster-link-alter-command), and in the [Tutorial: Share Data Across Topics Using Cluster Linking for Confluent Platform](topic-data-sharing.md#tutorial-topic-data-sharing). ### Configuration options specific to cluster links These properties are available to specify for the cluster link. Disabling a feature that has filters (ACL sync, consumer offset sync, auto-create mirror topics) clears any existing filters 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`. See [Use ACL filters to migrate or sync ACLs from Source to Destination cluster](security.md#cluster-link-acls-migrate) for examples of how to define the ACLs in the JSON file.
* Type: string * Default: “”
#### NOTE Populate `acl.filters` by passing a JSON file on the command line that specifies the ACLs as described in [Use ACL filters to migrate or sync ACLs from Source to Destination cluster](security.md#cluster-link-acls-migrate). `acl.sync.enable` : Whether to migrate ACLs. To learn more, see [Use ACL filters to migrate or sync ACLs from Source to Destination cluster](security.md#cluster-link-acls-migrate).
* Type: boolean * Default: false `acl.sync.ms` : How often to refresh the ACLs if ACL migration is enabled, in milliseconds.
* 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 are 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](/platform/current/multi-dc-deployments/cluster-linking/mirror-topics-cp.html#auto-create-mirror-topics). `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](/platform/current/multi-dc-deployments/cluster-linking/mirror-topics-cp.html#auto-create-mirror-topics). `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.enable](#consumer-group-prefix) is set to `true`. To learn more, see [Prefix mirror topics and consumer group names](mirror-topics-cp.md#cluster-link-prefix-concepts).
* Type: string * Default: null
#### NOTE - The prefix cannot be changed after the cluster link is created. - Enabling a cluster link prefix disables the ability to use the [reverse](mirror-topics-cp.md#reverse-source-and-mirror-topic) commands for disaster recovery. If your architecture requires reversing mirroring relationships during a failover, do not configure a prefix on the link. `cluster.link.paused` : Whether or not the cluster link is running or paused. The default is false.
* Type: boolean * Default: false `cluster.link.retry.timeout.ms` : The number of milliseconds after which failures are no longer retried and partitions are marked as failed. If the source topic is deleted and re-created within this timeout, the link may contain records from the old as well as the new topic.
* Type: int * Default: 300000 (5 minutes) `availability.check.ms` : How often the cluster link checks to see if the source cluster is available. The frequency with which the cluster link checks is specified in milliseconds.
* Type: int * Default: 60000 (1 minute)
A cluster link regularly checks whether the source cluster is still available for mirroring data by performing a `DescribeCluster` operation (bounded by `default.api.timeout.ms`). If the source cluster becomes unavailable (for example, because of an outage or disaster), then the cluster link signals this by updating its status and the status of its mirror topics. `availability.check.ms` works in tandem with [availability.check.consecutive.failure.threshold](#cluster-link-availability-check-consecutive-failure-threshold). `availability.check.consecutive.failure.threshold` : The number of consecutive failed availability checks the source cluster is allowed before the cluster link status becomes `SOURCE_UNAVAILABLE`.
* Type: int * Default: 5
If, for example, the default (5) is used, the source cluster is determined to be unavailable after 5 failed checks in a row. If [availability.check.ms](#cluster-link-availability-check-ms) and `default.api.timeout.ms` are also set to their defaults of 1 minute and there are 5 failed checks, then the cluster link will show as `SOURCE_UNAVAILABLE` after 5 \* (1+1) mins = 10 minutes. Note that this reflects that source unavailability is detected after `availability.check.consecutive.failure.threshold` \* (`default.api.timeout.ms` + `availability.check.ms`), taking into account the `DescribeCluster` operation performed as a part of [availability.check.ms](#cluster-link-availability-check-ms). `connections.max.idle.ms` : Idle connections timeout. The server socket processor threads close any connections that idle longer than this.
* Type: int * Default: 600000 `connection.mode` : Used only for source-initiated links. Set this to INBOUND on the destination cluster’s link (which you create first). Set this to OUTBOUND on the source cluster’s link (which you create second). You must use this in combination with `link.mode`. This property should only be set for source-initiated cluster links.
* Type: string * Default: OUTBOUND `consumer.offset.group.filters` : JSON string that lists the consumer groups to migrate. To learn more, see [Migrating consumer groups from source to destination cluster](commands.md#cluster-link-migrate-consumer-groups).
* Type: string * Default: “”
#### 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. `consumer.offset.sync.enable` : Whether or not to migrate consumer offsets from the source cluster.
If you set this up and run Cluster Linking, then later disable it, the filters will be cleared (deleted) from the cluster link.
* Type: boolean * Default: false `consumer.offset.sync.ms` : How often to sync consumer offsets, in milliseconds, if enabled.
* Type: int * Default: 30000 `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 [Prefix mirror topics and consumer group names](mirror-topics-cp.md#cluster-link-prefix-concepts).
* Type: boolean * Default: false
#### NOTE Consumer group prefixing cannot be enabled for bidirectional links. `num.cluster.link.fetchers` : Number of fetcher threads used to replicate messages from source brokers in cluster links.
* Type: int * Default: 1 `topic.config.sync.ms` : How often to refresh the topic configurations, in milliseconds.
* Type: int * Default: 5000 `topic.config.sync.include` : The list of topic configurations to sync from the source topic. By default, certain topic configurations are synced from the source topic to the mirror topic to ensure consistency. Use this parameter to explicitly specify which topic configurations to sync, giving you control over which properties are copied from source to destination. For more information, see [Override default syncing to specify independent mirror topic behavior](mirror-topics-cp.md#override-default-mirror-topic-syncs-cp) in [Mirror Topics](mirror-topics-cp.md#mirror-topics-concepts).
* Type: string * Default: All default sync configurations are included `link.fetcher.flow.control` : Maximum lag between high watermark and log end offset after which Cluster Linking will stop fetching. This is to synchronize the Cluster Linking fetch rate and the in-sync replica (ISR) fetch rate to avoid being under the minimum ISR. Setting this value specifies the flow control approach.
* Type: int * Default: 0
The following values for this configuration option apply to the approach:
- `>=0`: Lag approach. - `-1`: Under min ISR approach. `-1` means the maximum lag is not enforced. Cluster Linking fetch will stop when the partition is under the minimum ISR. - `-2`: Under-replicated partition approach. `-2` specifies that Cluster Linking fetch will stop when the partition is under-replicated.
If a broker goes down on the destination cluster due to an outage or planned failover (for example, proactively shutting down a broker), mirror topics will lag source topics on under-replicated partitions at the destination. To minimize or resolve mirror topic lag in these scenarios, set `link.fetcher.flow.control=-1`. `local.listener.name` : For a source-initiated link, an alternative listener for the link on the source cluster. For more information, see [Understanding Listeners in Cluster Linking](#cluster-link-listeners) `link.mode` : Used only for source-initiated links. Set this to DESTINATION on the destination cluster’s link (which you create first). Set this to SOURCE on the source cluster’s link (which you create second). For [bidirectional mode](#bidirectional-cluster-linking), set this to BIDIRECTIONAL on both clusters. You must use this in combination with `connection.mode`. This property should only be set for source-initiated cluster links.
* Type: string * Default: DESTINATION `mirror.start.offset.spec` : Whether to get the full history of a mirrored topic (`earliest`), exclude the history and get only the `latest` version, or to get the history of the topic starting at a given timestamp.
* Type: string * Default: earliest
- If set to a value of `earliest` (the default), new mirror topics get the full history of their associated topics. - If set to a value of `latest`, new mirror topics 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 its own value for `mirror.start.offset.spec`, and create mirror topics on the appropriate cluster link as desired. ## Kafka broker configurations Several properties specific to Cluster Linking are available to set on the brokers. A full reference is provided in [Kafka Broker and Controller Configuration Reference for Confluent Platform](../../installation/configuration/broker-configs.md#cp-config-brokers). Broker properties related to Cluster Linking are prefixed with `confluent.cluster.link`. The following entries provide additional explanation for a few of these. `confluent.cluster.link.enable` : Enables or disables Cluster Linking. In Confluent Platform 7.0.0 and later versions, the default is `true` and Cluster Linking is enabled by default. To learn how to turn off Cluster Linking, see [Disabling Cluster Linking](#cluster-linking-disable). `confluent.cluster.link.allow.config.providers` : When set to `true`, allows cluster links to use configuration providers to resolve cluster link configurations. When set to `false`, existing cluster links that already use configuration providers continue to work, but subsequent [alter cluster link](commands.md#cluster-link-alter-command) commands throw an `InvalidConfig` exception. The default value is `true`. ## Common Apache Kafka configuration options The following common properties are not specific to Cluster Linking, but are particularly relevant to setting up and managing cluster links. These common Kafka configurations for clients, brokers, and security are described in their respective sections. ### KRaft and ZooKeeper Starting with Confluent Platform 8.0, ZooKeeper is no longer available and is replaced by KRaft mode. This affects some basic Confluent Platform configurations. For Cluster Linking, `password.encoder.secret` is not required in KRaft mode to encrypt the credentials stored in the cluster link. However, it is required when using ZooKeeper (supported on pre-8.0 Confluent Platform versions) and when [migrating from ZooKeeper to KRaft](../../installation/migrate-zk-kraft.md#migrate-zk-kraft). Use of this parameter for Cluster Linking, when needed for older versions on ZooKeeper, is shown in [Tutorial: Link Confluent Platform and Confluent Cloud Clusters](hybrid-cp.md#cluster-link-hybrid-cp). To learn more, see [KRaft and ZooKeeper](index.md#kraft-zoo-cl-overview) under “What’s supported”. ### Client Configurations For a full list of `AdminClient` configurations, see [Kafka AdminClient Configurations for Confluent Platform](../../installation/configuration/admin-configs.md#cp-config-admin). - `bootstrap.servers` - `client.dns.lookup` - `metadata.max.age.ms` - `retry.backoff.ms` - `request.timeout.ms` ### Cluster Link Replication Configurations These configuration options are fully described in [Kafka Broker and Controller Configuration Reference for Confluent Platform](../../installation/configuration/broker-configs.md#cp-config-brokers). - `replica.fetch.backoff.ms` - `replica.fetch.max.bytes` - `replica.fetch.min.bytes` - `replica.fetch.response.max.bytes` - `replica.fetch.wait.max.ms` - `replica.socket.receive.buffer.bytes` - `replica.socket.timeout.ms` ### Client SASL and TLS/SSL Configurations - `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.protocol` - `ssl.provider` - `ssl.secure.random.implementation` - `ssl.trustmanager.algorithm` - `ssl.truststore.location` - `ssl.truststore.password` - `ssl.truststore.type` ### Configuring Reconnection Speed and Behavior A cluster link has two sets of configuration options with exponential backoff timing that control connections. These are the same options that Apache Kafka® clients have. - `reconnect.backoff.ms` and `reconnect.backoff.max.ms` - These options determine how soon the cluster link retries after a connection failure. These are 50ms and 10s by default for cluster links. - `socket.connection.setup.timeout.ms` and `socket.connection.setup.timeout.max.ms` - These options determine how long the cluster link waits for a connection attempt to succeed before breaking and retrying after a “reconnect backoff”. These are 10s and 30s, respectively, by default. On Confluent Platform clusters, reducing the values for these options may give faster reconnection speeds, at the expense of CPU and network usage. These options cannot be updated by cluster links that have a Confluent Cloud destination cluster. ## Bidirectional Cluster Linking 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-dc-deployments/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-dc-deployments/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-cp.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-dc-deployments/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 on its remote cluster with ACLs or 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 [Authorization (ACLs)](security.md#cluster-link-acls) (the rows for a cluster link on a source cluster). - `link.mode=BIDIRECTIONAL` #### NOTE 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). ### Create a bidirectional cluster link for local development This tutorial demonstrates how to create two clusters with bidirectional Cluster Linking for local development and testing. You can deploy these two clusters with whatever method is best for your circumstances. The following steps show an example of how to deploy a cluster on a development laptop from the Confluent Platform distribution. The example commands assume that the environment variable $CONFLUENT_HOME is set to the location of your Confluent Platform install; for example, `CONFLUENT_HOME=/home/ubuntu/confluent-8.0.0-0`. The examples also assume that the environment variable $CONFLUENT_CONFIG is set to `CONFLUENT_CONFIG=$CONFLUENT_HOME/etc/kafka`. This tutorial provides examples for KRaft mode only. Earlier versions of this documentation provide examples for ZooKeeper. #### IMPORTANT - As of Confluent Platform 8.0, ZooKeeper is no longer available for new deployments. - You should use KRaft mode for all new deployments. To learn more about running Kafka in KRaft mode, see [KRaft Overview for Confluent Platform](../../kafka-metadata/kraft.md#kraft-overview) and the [Platform Quick Start](../../get-started/platform-quickstart.md#cp-quickstart-step-1). To learn about migrating from older versions, see [Migrate from ZooKeeper to KRaft on Confluent Platform](../../installation/migrate-zk-kraft.md#migrate-zk-kraft). - Using the [truncate-and-restore](commands.md#truncate-and-restore-command) command on Confluent Platform **7.9.0 - 7.9.2 or 8.0.0** and using [Tiered Storage](../../clusters/tiered-storage.md#tiered-storage) on the topics being truncated can result in a silent failure leading to data inconsistency or even limited data loss. If you are using these versions and Tiered Storage as part of a disaster recovery (DR) failback strategy with bidirectional links, do not run `truncate-and-restore`. To resolve the issue, [Upgrade to a patched version](../../installation/installing_cp/zip-tar.md#prod-kafka-cli-install): Confluent Platform 7.9.3, 8.0.1, 8.1.0 and later have mitigated this issue, and the features can be safely used. #### Download Confluent Platform [Download Confluent Platform 7.5 or later](../../installation/installing_cp/zip-tar.md#prod-kafka-cli-install). #### Ports and configuration mapping The example deployment in this tutorial uses the following port and feature configurations, and assumes that services will run on `localhost`. | | Cluster A | Cluster B | |-------------------|------------------------------------------------------|-----------------------------------------------------| | Kafka brokers | 9092 | 9992 | | KRaft controllers | 9094 | 9095 | | HTTP listeners | 8090 | 8091 | | Cluster link | 9092 ([links to cluster B](#create-link-to-b-on-a)) | 9992 ([links to cluster A](#create-link-to-a-on-b)) | #### Kafka command-line tools This tutorial uses the Apache Kafka® command-line interface (CLI) tools. For a command reference and more information, see [Kafka Command-Line Interface (CLI) Tools](/kafka/operations-tools/kafka-tools.html). The [Confluent CLI](https://docs.confluent.io/confluent-cli/current/command-reference/overview.html) provides analogous commands, but requires a login, which on Confluent Platform requires considerable security setup, not appropriate for this brief, lightweight tutorial. #### Create the examples directories 1. Change directories to the location where Confluent Platform is installed: ```bash cd $CONFLUENT_HOME ``` 2. Create a directory to be used for all of your example files: ```bash mkdir my-examples ``` #### Set up and start cluster A ##### Configure cluster A server properties 1. Copy `$CONFLUENT_CONFIG/server.properties` to use as a basis for `my-examples/a1.properties`. ```bash cp $CONFLUENT_CONFIG/server.properties $CONFLUENT_HOME/my-examples/a1.properties ``` 2. Update the port numbers. ```bash sed -i '' -e "s/9093/9094/g" my-examples/a1.properties ``` 3. Modify the listener configurations. ```bash echo "listeners=SASL_PLAINTEXT://:9092,PLAINTEXT://:9093,CONTROLLER://:9094" >> $CONFLUENT_HOME/my-examples/a1.properties ``` ```bash echo "advertised.listeners=SASL_PLAINTEXT://localhost:9092,CONTROLLER://localhost:9094,PLAINTEXT://localhost:9093" >> $CONFLUENT_HOME/my-examples/a1.properties ``` ```bash echo "controller.listener.names=CONTROLLER" >> $CONFLUENT_HOME/my-examples/a1.properties ``` 4. Update Kafka data directories. ```bash sed -i '' -e "s/kraft-combined-logs/kraft-combined-logs-a/g" $CONFLUENT_HOME/my-examples/a1.properties ``` 5. Append the following lines to the end of the server properties file to set license, telemetry, replication, and security configurations. ```bash echo "confluent.license.topic.replication.factor=1" >> $CONFLUENT_HOME/my-examples/a1.properties ``` ```bash echo "confluent.cluster.link.metadata.topic.replication.factor=1" >> $CONFLUENT_HOME/my-examples/a1.properties ``` ```bash echo "confluent.reporters.telemetry.auto.enable=false" >> $CONFLUENT_HOME/my-examples/a1.properties ``` ```bash echo "super.users=User:link;User:admin;User:ANONYMOUS;User:kafka" >> $CONFLUENT_HOME/my-examples/a1.properties ``` ```bash echo "broker.users=User:ANONYMOUS" >> $CONFLUENT_HOME/my-examples/a1.properties ``` ```bash echo "inter.broker.listener.name=PLAINTEXT" >> $CONFLUENT_HOME/my-examples/a1.properties ``` ```bash echo "sasl.enabled.mechanisms=SCRAM-SHA-512" >> $CONFLUENT_HOME/my-examples/a1.properties ``` ```bash echo "listener.name.sasl_plaintext.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required;" >> $CONFLUENT_HOME/my-examples/a1.properties ``` ```bash echo "authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer" >> $CONFLUENT_HOME/my-examples/a1.properties ``` 6. Open the file `$CONFLUENT_HOME/my-examples/a1.properties` in a text editor and manually comment out the first instances in the file of listeners and advertised listeners: ```bash # listeners=PLAINTEXT://:9092,CONTROLLER://:9094 # advertised.listeners=PLAINTEXT://localhost:9092,CONTROLLER://localhost:9094 ``` You have appended new configurations for these at the end of the file with the `echo` commands above, and commenting out these first occurrences will prevent any conflicts. This should result in the given key settings included in `a1.properties`: ```properties log.dirs=/tmp/kraft-combined-logs-a process.roles=broker,controller # inherited from server.properties listeners=SASL_PLAINTEXT://:9092,PLAINTEXT://:9093,CONTROLLER://:9094 advertised.listeners=SASL_PLAINTEXT://localhost:9092,CONTROLLER://localhost:9094,PLAINTEXT://localhost:9093 controller.listener.names=CONTROLLER confluent.license.topic.replication.factor=1 confluent.cluster.link.metadata.topic.replication.factor=1 confluent.reporters.telemetry.auto.enable=false super.users=User:link;User:admin;User:ANONYMOUS;User:kafka broker.users=User:ANONYMOUS inter.broker.listener.name=PLAINTEXT sasl.enabled.mechanisms=SCRAM-SHA-512 listener.name.sasl_plaintext.scram-sha-512.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required; authorizer.class.name=org.apache.kafka.metadata.authorizer.StandardAuthorizer ``` ##### Create credentials on cluster A Run the following commands from `$CONFLUENT_HOME`, the top of your Confluent Platform install directory. Run these commands in a new dedicated shell or terminal window. 1. Open a new command window and change directories to $CONFLUENT_HOME. ```bash cd $CONFLUENT_HOME ``` 2. Generate a `random-uuid` using the kafka-storage tool. ```bash KAFKA_CLUSTER_ID="$(bin/kafka-storage random-uuid)" ``` 3. Format log directories for this server, and create SASL SCRAM credentials for the cluster and the link. A user called “kafka” will be used by the cluster itself. For KRaft, both credentials must be applied together in a single command. ```bash kafka-storage format -t $KAFKA_CLUSTER_ID -c $CONFLUENT_HOME/my-examples/a1.properties -S 'SCRAM-SHA-512=[name=kafka,iterations=8192,password=kafka-secret]' -S 'SCRAM-SHA-512=[name=link,iterations=8192,password=link-secret]' --ignore-formatted --standalone ``` ##### Start cluster A In the command window dedicated to cluster A, start cluster A and leave it running. ```bash kafka-server-start $CONFLUENT_HOME/my-examples/a1.properties ``` #### Set up and start cluster B ##### Configure cluster B server properties 1. Copy `$CONFLUENT_HOME/my-examples/a1.properties` (the file you just created) to use as a basis for `my-examples/b1.properties`: ```bash cp $CONFLUENT_HOME/my-examples/a1.properties $CONFLUENT_HOME/my-examples/b1.properties ``` 2. Update the port numbers. ```bash sed -i '' -e "s/9092/9992/g" my-examples/b1.properties ``` ```bash sed -i '' -e "s/9094/9095/g" my-examples/b1.properties ``` ```bash sed -i '' -e "s/9093/9096/g" my-examples/b1.properties ``` ```bash echo "confluent.http.server.listeners=http://localhost:8091" >> $CONFLUENT_HOME/my-examples/b1.properties ``` (Note that cluster A uses default port numbers for the broker and for the HTTP server listeners, among others. The default port numbers are assumed in commented out lines. For cluster B, you must change some of the default port numbers to be different from cluster A, including for the broker and HTTP server listeners.) 3. Update Kafka data directories. ```bash sed -i '' -e "s/kraft-combined-logs-a/kraft-combined-logs-b/g" $CONFLUENT_HOME/my-examples/b1.properties ``` The `b1.properties` file should match `a1.properties` except for port numbers and log directories. ##### Create credentials on cluster B Run the following `kafka-storage` commands from `$CONFLUENT_HOME`, the top of your Confluent Platform install directory. Run these commands in a new dedicated shell or terminal window. 1. Open a new command window and change directories to $CONFLUENT_HOME. ```bash cd $CONFLUENT_HOME ``` 2. Generate a `random-uuid` using the kafka-storage tool. ```bash KAFKA_CLUSTER_ID="$(bin/kafka-storage random-uuid)" ``` 3. Format log directories for this server and create SASL SCRAM credentials on the cluster for a user called “kafka” that will be used by the Kafka cluster itself, and for the link (`link`). ```bash kafka-storage format -t $KAFKA_CLUSTER_ID -c $CONFLUENT_HOME/my-examples/b1.properties -S 'SCRAM-SHA-512=[name=kafka,iterations=8192,password=kafka-secret]' -S 'SCRAM-SHA-512=[name=link,iterations=8192,password=link-secret]' --ignore-formatted --standalone ``` ##### Start cluster B In the command window dedicated to cluster B, start cluster B, and leave it running. ```bash kafka-server-start $CONFLUENT_HOME/my-examples/b1.properties ``` #### Create credentials for the cluster link 1. Create a file with the admin credentials to authenticate when you run commands against the Confluent Platform cluster. Create a file called `command.config` in the `$CONFLUENT_HOME/my-examples` directory with the following content: ```bash sasl.mechanism=SCRAM-SHA-512 security.protocol=SASL_PLAINTEXT sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \ username="kafka" \ password="kafka-secret"; ``` 2. Get the cluster IDs for clusters A and B. ```bash kafka-cluster cluster-id --bootstrap-server localhost:9092 --config $CONFLUENT_HOME/my-examples/command.config ``` ```bash kafka-cluster cluster-id --bootstrap-server localhost:9992 --config $CONFLUENT_HOME/my-examples/command.config ``` Your output for each should resemble: ```bash Cluster ID: Abc123_Abc ``` In this example, `Abc123_Abc` is the Confluent Platform cluster ID, referred to in these examples as `$CP_CLUSTER_ID`. Optionally, set an environment variable for this either in the local shell or in a zsh or bash profile so that you can copy and paste commands directly in later steps. ```bash export A_CLUSTER_ID= ``` ```bash export B_CLUSTER_ID= ``` #### Populate and test cluster A These commands use the Confluent Platform CLI. 1. Create a topic on cluster A with a single partition so ordering is easier to see. ```bash kafka-topics --create --topic demo --partitions 1 --replication-factor 1 --bootstrap-server localhost:9092 --command-config $CONFLUENT_HOME/my-examples/command.config ``` You should get confirmation that the topic was successfully created. ```bash Created topic demo. ``` You can get a list of existing topics as follows: ```bash kafka-topics --list --bootstrap-server localhost:9092 --command-config $CONFLUENT_HOME/my-examples/command.config ``` And get detailed information on a topic with the `--describe` option: ```bash kafka-topics --describe --topic demo --bootstrap-server localhost:9092 --command-config $CONFLUENT_HOME/my-examples/command.config ``` 2. Send some messages to the demo topic on cluster A, and fill it with data. ```bash seq 1 5 | kafka-console-producer --topic demo --bootstrap-server localhost:9092 --producer.config $CONFLUENT_HOME/my-examples/command.config ``` The command should terminate without any output. 3. Consume from the topic on cluster A. Run a consumer to consume messages from the `demo` topic. ```bash kafka-console-consumer --topic demo --from-beginning --bootstrap-server localhost:9092 --consumer.config $CONFLUENT_HOME/my-examples/command.config ``` If the topic successfully consumes the messages, your output will be: ```bash 1 2 3 4 5 ``` Use keyboard command Ctrl+C to get the prompt back. #### Create the Principal and ACLs to allow the cluster link to read from cluster A The cluster link needs a principal that is authorized to read data from cluster A. You created the “link” principal in the cluster setup step, above, and now you will assign it the required privileges. 1. Give the link’s principal the **Describe:Cluster ACL**. ```bash $CONFLUENT_HOME/bin/kafka-acls --command-config my-examples/command.config --bootstrap-server localhost:9092 \ --add --allow-principal User:link --operation Describe --cluster ``` This ACL is specifically required for bidirectional mode. 2. At a minimum, give the link’s principal **Read:Topics** and **DescribeConfigs:Topics** on the topics that the cluster link is allowed to read from. This example allows the cluster link to read data from all topics. Alternatively, only specific topic names or prefixes can be given. These can be different from the topic ACLs given on the remote cluster. 3. (Recommended) Assign additional ACLs for syncing consumer offsets, which is a critical feature of a bidirectional cluster link. To learn about consumer offset sync configuration options, see `consumer.offset.sync.enable` and `consumer.offset.sync.ms` in [Configuration options specific to cluster links](#cp-cluster-link-config-options). - Grant the link’s principal **Describe** permissions on all topics. ```bash $CONFLUENT_HOME/bin/kafka-acls --command-config my-examples/command.config --bootstrap-server localhost:9092 --add --allow-principal User:link --operation Describe --topic "*" ``` Your output should resemble: ```bash Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=*, patternType=LITERAL)`: (principal=User:link, host=*, operation=DESCRIBE, permissionType=ALLOW) ``` - Grant the link’s principal **Describe** permissions on all consumer groups. ```bash $CONFLUENT_HOME/bin/kafka-acls --command-config my-examples/command.config --bootstrap-server localhost:9092 --add --allow-principal User:link --operation Describe --operation Read --group "*" ``` Your output should resemble: ```bash Adding ACLs for resource `ResourcePattern(resourceType=GROUP, name=*, patternType=LITERAL)`: (principal=User:link, host=*, operation=READ, permissionType=ALLOW) (principal=User:link, host=*, operation=DESCRIBE, permissionType=ALLOW) Current ACLs for resource `ResourcePattern(resourceType=GROUP, name=*, patternType=LITERAL)`: (principal=User:link, host=*, operation=DESCRIBE, permissionType=ALLOW) (principal=User:link, host=*, operation=READ, permissionType=ALLOW) ``` - Grant the link’s principal **DescribeConfigs** permissions on the cluster. ```bash $CONFLUENT_HOME/bin/kafka-acls --command-config my-examples/command.config --bootstrap-server localhost:9092 --add --allow-principal User:link --operation DescribeConfigs --cluster ``` Your output should resemble: ```bash Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER, name=kafka-cluster, patternType=LITERAL)`: (principal=User:link, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW) ``` 4. (Optional) Assign additional ACLs for [syncing (migrating) ACLs](security.md#cluster-link-acls-migrate) or using [prefixing](/platform/current/multi-dc-deployments/cluster-linking/mirror-topics-cp.html#prefix-mirror-topics-and-consumer-group-names) plus [auto-create mirror topics](/platform/current/multi-dc-deployments/cluster-linking/mirror-topics-cp.html#auto-create-mirror-topics). These can be different from the ACLs given on the remote cluster. #### Create the Principal and ACLs to allow the cluster link to read from cluster B The cluster link needs a principal that is authorized to read data from cluster B. 1. Give the link’s principal the **Describe:Cluster ACL**. ```bash $CONFLUENT_HOME/bin/kafka-acls --command-config my-examples/command.config --bootstrap-server localhost:9992 \ --add --allow-principal User:link --operation Describe --cluster ``` Your output should resemble: ```bash Adding ACLs for resource `ResourcePattern(resourceType=CLUSTER, name=kafka-cluster, patternType=LITERAL)`: (principal=User:link, host=*, operation=DESCRIBE, permissionType=ALLOW) Current ACLs for resource `ResourcePattern(resourceType=CLUSTER, name=kafka-cluster, patternType=LITERAL)`: (principal=User:link, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW) (principal=User:link, host=*, operation=DESCRIBE, permissionType=ALLOW) ``` This ACL is specifically required for bidirectional mode. 2. Give the link’s principal **Read:Topics** and **DescribeConfigs:Topics** on the topics that the cluster link is allowed to read from. ```bash $CONFLUENT_HOME/bin/kafka-acls --command-config my-examples/command.config --bootstrap-server localhost:9992 \ --add --allow-principal User:link --operation DescribeConfigs --operation Read --topic "*" ``` Your output should resemble: ```bash Adding ACLs for resource `ResourcePattern(resourceType=TOPIC, name=*, patternType=LITERAL)`: (principal=User:link, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW) (principal=User:link, host=*, operation=READ, permissionType=ALLOW) Current ACLs for resource `ResourcePattern(resourceType=TOPIC, name=*, patternType=LITERAL)`: (principal=User:link, host=*, operation=DESCRIBE_CONFIGS, permissionType=ALLOW) (principal=User:link, host=*, operation=READ, permissionType=ALLOW) ``` This example allows the cluster link to read data from all topics. Alternatively, only specific topic names or prefixes can be given. These can be different from the topic ACLs given on the remote cluster. 3. (Optional) Assign additional ACLs for syncing consumer offsets, syncing ACLs, or using prefixing and auto-create mirror topics. These can be different from the ACLs given on the remote cluster. #### Create the cluster link objects A cluster link object must be created on both clusters. When using the default security mode where both clusters can reach each other, either object may be created first. In this example, you will create the cluster link object on cluster B first; this is simply to demonstrate that order does not matter in this scenario. Both cluster link objects must use the **same name** and the **same link mode setting** for this special configuration: `link.mode=BIDIRECTIONAL` Apart from that, the [standard cluster link configurations](#cp-cluster-link-config-options) are valid and the standard Cluster Linking commands work the same. ##### Create the cluster link on cluster B 1. Create a config file called `b-link.config` that points to cluster A’s bootstrap servers (`9092`), and specifies `link.mode=BIDIRECTIONAL` along with authentication details. ```bash bootstrap.servers=localhost:9092 link.mode=BIDIRECTIONAL sasl.mechanism=SCRAM-SHA-512 security.protocol=SASL_PLAINTEXT sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \ username="link" \ password="link-secret"; ``` Note that more [cluster link configurations](#cp-cluster-link-config-options) 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. Run the following CLI command to create a link named “bidirectional-link” on cluster B. (You can name the link per your preferences, but these link names must be identical on both clusters.) ```bash $CONFLUENT_HOME/bin/kafka-cluster-links --create --link bidirectional-link \ --config-file my-examples/b-link.config --bootstrap-server localhost:9992 \ --command-config my-examples/command.config ``` Your output should resemble: ```bash Cluster link 'bidirectional-link' creation successfully completed. ``` ##### Create the cluster link on cluster A 1. Create a config file called `a-link.config` that points to cluster B’s bootstrap servers (`9992`), and specifies `link.mode=BIDIRECTIONAL` along with authentication details. ```bash bootstrap.servers=localhost:9992 link.mode=BIDIRECTIONAL sasl.mechanism=SCRAM-SHA-512 security.protocol=SASL_PLAINTEXT sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \ username="link" \ password="link-secret"; ``` Note that more [cluster link configurations](#cp-cluster-link-config-options) 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. Run the following CLI command to create a link named “bidirectional-link”. (You can name the link per your preferences, but these link names must be identical on both clusters.) ```bash $CONFLUENT_HOME/bin/kafka-cluster-links --create --link bidirectional-link \ --config-file my-examples/a-link.config --bootstrap-server localhost:9092 \ --command-config my-examples/command.config ``` Your output should resemble: ```bash Cluster link 'bidirectional-link' creation successfully completed. ``` #### Create topics and mirror topics 1. Create a topic on Cluster A: ```bash $CONFLUENT_HOME/bin/kafka-topics --topic from-a --create --partitions 6 --replication-factor 1 \ --command-config my-examples/command.config --bootstrap-server localhost:9092 ``` Expected output: ```bash Created topic from-a. ``` 2. Create a corresponding mirror topic on Cluster B: ```bash $CONFLUENT_HOME/bin/kafka-mirrors --create --mirror-topic from-a --link bidirectional-link \ --replication-factor 1 --command-config my-examples/command.config --bootstrap-server localhost:9992 ``` Expected output: ```bash Created topic from-a. ``` 3. Create a topic on Cluster B: ```bash $CONFLUENT_HOME/bin/kafka-topics --topic b-was-here --create --partitions 6 --replication-factor 1 \ --command-config my-examples/command.config --bootstrap-server localhost:9992 ``` Expected output: ```bash Created topic b-was-here. ``` 4. Create a corresponding mirror topic on Cluster A: ```bash $CONFLUENT_HOME/bin/kafka-mirrors --create --mirror-topic b-was-here --link bidirectional-link \ --replication-factor 1 --command-config my-examples/command.config --bootstrap-server localhost:9092 Expected output: ``` ```bash Created topic b-was-here. ``` ##### (Optional) List topics and test mirrors You have successfully created a bidirectional cluster link. At this point, you could do further testing and verification, if desired. 1. List topics on cluster A. ```bash kafka-topics --list --bootstrap-server localhost:9092 --command-config $CONFLUENT_HOME/my-examples/command.config ``` Expected output (which will include some internal topics): ```bash __consumer_offsets _confluent-command _confluent-link-metadata b-was-here demo ``` 2. List topics on cluster B. ```bash kafka-topics --list --bootstrap-server localhost:9992 --command-config $CONFLUENT_HOME/my-examples/command.config ``` Expected output (which will include some internal topics): ```bash __consumer_offsets _confluent-command _confluent-link-metadata b-was-here from-a ``` 3. Send messages to `from-a` source topic on cluster A, and fill it with data. ```bash echo "rain forest parrots are over here on cluster A" | kafka-console-producer --topic from-a --bootstrap-server localhost:9092 --producer.config $CONFLUENT_HOME/my-examples/command.config ``` The command should terminate without any output. 4. Consume from the corresponding mirror topic on cluster B. Run a consumer to consume messages from the mirror topic on cluster B. ```bash kafka-console-consumer --topic from-a --from-beginning --bootstrap-server localhost:9992 --consumer.config $CONFLUENT_HOME/my-examples/command.config ``` If the topic successfully consumes the messages, your output will be: ```bash rain forest parrots are over here on cluster A ``` Use keyboard command Ctrl+C to get the prompt back. ### Advanced options for bidirectional Cluster Linking 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. 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. 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 this case, a bidirectional cluster link can be configured such 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 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?](faqs-cp.md#faq-unidirectional-vs-bidirectional-source-based-links) in the FAQ. ![image](multi-dc-deployments/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: - The inbound link on the less privileged cluster must be created before the link on the remote cluster using `connection.mode=INBOUND` and `link.mode=BIDIRECTIONAL`. Otherwise, the links on both sides may have different IDs even though they have the same names. The link IDs, not the names, establish the connection between the two sides. For example, create a `b-link.config` file for cluster B as shown: ```bash link.mode=BIDIRECTIONAL connection.mode=INBOUND ``` Then, run the command to create an INBOUND only link on cluster B, including a call to your `b-link.config`: ```bash $CONFLUENT_HOME/bin/kafka-cluster-links --create --link bidirectional-link \ --config-file my-examples/b-link.config --bootstrap-server localhost:9992 \ --command-config my-examples/command-B.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 # bootstrap of the remote cluster bootstrap.servers=localhost:9992 link.mode=BIDIRECTIONAL # authentication for the link principal on the remote cluster sasl.mechanism=SCRAM-SHA-512 security.protocol=SASL_PLAINTEXT sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \ username="link" \ password="link-secret"; # authentication for the link principal on the local cluster local.sasl.mechanism=SCRAM-SHA-512 local.security.protocol=SASL_PLAINTEXT local.sasl.jaas.config=org.apache.kafka.common.security.scram.ScramLoginModule required \ username="link" \ password="link-secret"; ``` - Additional configurations can be included in this file as needed, per [cluster link configurations](#cp-cluster-link-config-options). - An authentication configuration (such as API key or OAuth) for a principal on its **remote cluster** with ACLs or RBAC role bindings giving permission to read topic data and metadata. - **Alter:Cluster ACL** – this is unique to the advanced mode - Describe:Cluster ACL - The required ACLs or RBAC role bindings for a cluster link, as described in [Manage Security for Cluster Linking on Confluent Platform](security.md#cluster-link-security) (for [a cluster link on a source cluster](security.md#cluster-link-acls-for-link-on-source) and for a [source-initiated link on the destination cluster](security.md#cluster-link-acls-for-source-initiated-links)). - **Local authentication** (unique to the unidirectional mode): An authentication configuration (such as API key or OAuth) for a principal **on its own cluster** with ACLs or RBAC role bindings giving permission to read topic data and the Describe:Cluster ACL - The required ACLs or RBAC role bindings as described giving permission to read topic data and the Describe:Cluster ACL, as described in [Manage Security for Cluster Linking on Confluent Platform](security.md#cluster-link-security). This authentication configuration never leaves this cluster. The lines for these configurations must be prefixed with `local.` to indicate that they belong to the local cluster. - Link mode set to `link.mode=BIDIRECTIONAL` 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 config file: ```bash $CONFLUENT_HOME/bin/kafka-cluster-links --create --link bidirectional-link \ --config-file my-examples/a-link.config \ --bootstrap-server localhost:9092 --command-config my-examples/command.config ``` ### Convert a mirror topic to a regular topic For the purposes of failover, `bidirectional` mode gives you the option to [reverse the mirroring relationship](mirror-topics-cp.md#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. You can [convert a mirror topic to a regular topic](mirror-topics-cp.md#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 `promote` operation, run `truncate-and-restore` on the original source topic to mirror from the newly promoted topic. The `truncate-and-restore` command is only available on [bidirectional links](mirror-topics-cp.md#bidirectional-linking-cp) in [KRaft mode](../../kafka-metadata/kraft.md#kraft-overview). ## Required Configurations for Control Center Cluster Linking requires embedded v3 Confluent REST Proxy to communicate with Confluent Control Center and properly display mirror topics on the Control Center UI. If the REST configurations are not implemented, mirror topics will display in Control Center as regular topics, showing inaccurate information. (To learn more, see [Known limitations and best practices](index.md#cluster-linking-limitations).) ### Configure REST Endpoints in the Control Center properties file If you want to use Control Center with Cluster Linking, you must configure the Control Center cluster with REST endpoints to enable HTTP servers on the brokers. If this is not configured properly for all brokers, Cluster Linking will not be accessible from Confluent Control Center. In the appropriate Control Center properties file (for example `$CONFLUENT_HOME/etc/confluent-control-center/control-center-dev.properties` or `control-center.properties`), use `confluent.controlcenter.streams.cprest.url` to define the REST endpoints for `controlcenter.cluster`. The default is `http://localhost:8090`, as shown below. ```bash # Kafka REST endpoint URL confluent.controlcenter.streams.cprest.url="http://localhost:8090" ``` Identify the associated URL for each broker. If you have multiple brokers in the cluster, use a comma-separated list. #### SEE ALSO `confluent.controlcenter.streams.cprest.url` in the [Control Center Configuration Reference](https://docs.confluent.io/control-center/current/installation/configuration.html#c3-connect-cluster-setting) ### Configure authentication for REST endpoints on Kafka brokers (Secure Setup) To run Cluster Linking in a secure setup, you must configure authentication for REST endpoints in each of the Kafka broker `server.properties` files. If the Kafka broker files are missing these configs, Control Center will not be able to access Cluster Linking in a secure setup. At a minimum, you will need the following configurations. ```bash # EmbeddedKafkaRest: HTTP Auth Configuration kafka.rest.kafka.rest.resource.extension.class=io.confluent.kafkarest.security.KafkaRestSecurityResourceExtension kafka.rest.rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.InstallBearerOrBasicSecurityHandler ``` Here is an example of an MDS client configuration for Kafka RBAC in a broker `server.properties` file . ```bash # EmbeddedKafkaRest: Kafka Client Configuration kafka.rest.bootstrap.servers=, , kafka.rest.client.security.protocol=SASL_PLAINTEXT # EmbeddedKafkaRest: HTTP Auth Configuration kafka.rest.kafka.rest.resource.extension.class=io.confluent.kafkarest.security.KafkaRestSecurityResourceExtension kafka.rest.rest.servlet.initializor.classes=io.confluent.common.security.jetty.initializer.InstallBearerOrBasicSecurityHandler kafka.rest.public.key.path= # EmbeddedKafkaRest: MDS Client configuration kafka.rest.confluent.metadata.bootstrap.server.urls=, , kafka.rest.ssl.truststore.location= kafka.rest.ssl.truststore.password= kafka.rest.confluent.metadata.http.auth.credentials.provider=BASIC kafka.rest.confluent.metadata.basic.auth.user.info= kafka.rest.confluent.metadata.server.urls.max.age.ms=60000 kafka.rest.client.confluent.metadata.server.urls.max.age.ms=60000 ``` #### SEE ALSO - [Configure Security for the Admin REST APIs for Confluent Server on Confluent Platform](../../kafka-rest/production-deployment/confluent-server/security.md#confluent-server-rest-security) - [Deploy Secure Standalone REST Proxy in Confluent Platform](../../kafka-rest/production-deployment/rest-proxy/security.md#kafkarest-security) - [Scripted Confluent Platform Demo](../../tutorials/cp-demo/index.md#cp-demo), On-Prem Tutorial, Security section provides examples of different types of configurations ## Disabling Cluster Linking To disable Cluster Linking on a cluster running Confluent Enterprise version 7.0.0 or later, add the following line to the broker configuration on the destination cluster (for example `$CONFLUENT_HOME/etc/server.properties`). ```bash confluent.cluster.link.enable=false ``` This disables the creation of cluster links with that cluster as the destination, or source-initiated cluster links with that cluster as the source. This does not disable creating a destination-initiated cluster link with this cluster as its source. Cluster Linking is not available as a dynamic configuration. It must either be enabled before starting the brokers, or to enable it on a running cluster where it was previously turned off, set the configuration `confluent.cluster.link.enable=true` on the brokers and restart them to perform a rolling update. ## Understanding Listeners in Cluster Linking For a forward connection, the target server knows which listener the connection came in on and associates the listener with that connection. When a metadata request arrives on that connection, the server returns metadata corresponding to the listener. For example, in Confluent Cloud, when a client on the external listener asks for the leader of `topicA`, it always gets the external endpoint of the leader and never the internal one, because the system knows the listener name from the connection. For reverse connections, the target server (that is, the source cluster) established the connection. When the connection is reversed, this target server needs to know which listener to associate the reverse connection with; that is, for example, which endpoint it should return to the destination for its leader requests. By default, the listener is associated based on the source cluster where the link was created. In most cases this is sufficient because typically a single external listener is used. On Confluent Cloud, this default is used and you cannot override it. On self-managed Confluent Platform, you have the option to override the default listener/connection association. This provides the flexibility to create the source link on an internal listener but associate the external listener with the reverse connection. The configuration `local.listener.name` refers to source cluster listener name. By default, this is the listener that was used to create the source link. If you want to use a different listener, you must explicitly configure it. If Confluent Cloud is the source, then it would be the external listener (default) and cannot be overridden. For the destination, the listener is determined by bootstrap.servers and cannot be overridden. ## Suggested Reading - [Command Reference for Cluster Linking on Confluent Platform](commands.md#cluster-link-commands) - [Manage Security for Cluster Linking on Confluent Platform](security.md#cluster-link-security) - [Monitor Cluster Metrics and Optimize Links for Cluster Linking on Confluent Platform](metrics.md#cluster-linking-metrics) - [Kafka Configuration Reference for Confluent Platform](../../installation/configuration/index.md#cp-config-reference)