Configuration Options for Cluster Linking¶
Looking for Confluent Cloud Cluster Linking docs? You are currently viewing Confluent Platform documentation. If you are looking for Confluent Cloud docs, check out Cluster Linking on Confluent Cloud.
Using Cluster Linking with Confluent for Kubernetes¶
You can use Cluster Linking with Confluent Platform deployed with Confluent for Kubernetes.
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.
To configure Cluster Linking on earlier versions of CFK, use
configOverrides in the Kafka
custom resource. See Configuration Overrides.
in the CFK documentation for more information about using
Also, pre Confluent Platform 7.0.0 releases required that you include a
configOverrides section on the server
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.
apiVersion: platform.confluent.io/v1beta1 kind: Kafka metadata: name: kafka namespace: confluent spec: replicas: 3 image: application: confluentinc/cp-server:7.0.8 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.
Starting in Confluent Platform 7.0.0, Cluster Linking is enabled by default, so no changes are needed to the configuration file.
Pre Confluent Platform 7.0.0 releases required that you add a broker configuration
property to the
kafka_broker_custom_properties section in the inventory as described in
Configure Confluent Platform with Ansible,
confluent.cluster.link.enable: "true". If you are upgrading from an earlier release,
this configuration can be deleted, as it is no longer needed.
Several configurations are available for cluster links. The following sections describe how to set these using the CLI Commands, and then list the available properties.
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 using either:
--config-fileflag, when you first create the link,
- Or, the
--add-config-fileflag 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 either:
--configflag, when you first create the link,
- Or, the
--add-configflag to update configurations on an existing link.
When updating the configuration for an existing cluster link, pass in only those configs that change. Be especially mindful when you are using a config file with
--add-config--file(where it would be easy to pass in a full set of configs) that it contains only the configs you want to update. For example,
You can change several aspects of a cluster link configuration, but you cannot change its source cluster (source cluster ID), prefix, or the link name.
Examples and command syntax for specifying link properties in a file and at the command line are shown in in Creating a Cluster Link and in Altering a Cluster Link, and in the Tutorial: Using Cluster Linking for Topic Data Sharing.
These properties are available to specify for the cluster link.
If you disable a feature that has filters (consumer offset sync, auto create mirror topics) after having it enabled initially, then any existing filters will be cleared (deleted) from the cluster link.
- Whether or not 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.
- 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.
Whether or not the cluster link is running or paused. The default is false.
- Type: boolean
- Default: false
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)
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. 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.msworks in tandem availability.check.consecutive.failure.threshold.
The number of consecutive failed availability checks the source cluster is allowed before the cluster link status becomes
- 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 is also set to its default of 1 minute and there are 5 failed checks, then the cluster link will show as
SOURCE_UNAVAILABLEafter 5 minutes.
- Enables or disables Cluster Linking. In Confluent Platform 7.0.0 and later versions, the default is
true, Cluster Linking is enabled by default. To learn how to turn off Cluster Linking, see Disabling Cluster Linking.
Idle connections timeout. The server socket processor threads close any connections that idle longer than this.
- Type: int
- Default: 600000
JSON to denote the list of consumer groups to be migrated. To learn more, see Migrating Consumer Groups from Source to Destination Cluster.
- Type: string
- Default: “”
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.
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
How often to sync consumer offsets, in milliseconds, if enabled.
- Type: int
- Default: 30000
Number of fetcher threads used to replicate messages from source brokers in cluster links.
- Type: int
- Default: 1
How often to refresh the topic configs, in milliseconds.
- Type: int
- Default: 5000
- For a source initiated link, an alternative listener to used by the cluster link on the source cluster. For more, see Understanding Listeners in Cluster Linking
Common Configuration Options¶
The following subset of common properties, although not specific to Cluster Linking, may be particularly relevant to setting up and managing cluster links. These are common across Confluent Platform for clients, brokers, and security configurations, and are described in their respective sections per the links provided.
For a full list of
AdminClient configurations, see AdminClient Configurations.
Cluster Link Replication Configurations¶
These configuration options are fully described in Kafka Broker Configurations.
Client SASL and SSL Configurations¶
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. (See also, Known Limitations and Best Practices.)
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.controlcenter.streams.cprest.url to define the REST endpoints for
The default is
http://localhost:8090, as shown below.
# 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.
confluent.controlcenter.streams.cprest.url in the
Control Center Configuration Reference
Configure authentication for REST endpoints on Kafka brokers (Secure Setup)¶
- Cluster Linking does not require the Metadata Service (MDS) or security to run, but if you want to configure security, you can get started with the following example which shows an MDS client configuration for RBAC.
- You can use
confluent.metadata.server.listeners(which will enable the Metadata Service) instead of
confluent.http.server.listenersto listen for API requests. Use either
confluent.http.server.listeners, but not both. If a listener uses HTTPS, then appropriate SSL configuration parameters must also be set. To learn more, see Admin REST APIs Configuration Options.
To run Cluster Linking in a secure setup, you must configure authentication for REST endpoints in each of the Kafka broker
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.
# 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 .
# EmbeddedKafkaRest: Kafka Client Configuration kafka.rest.bootstrap.servers=<host:port>, <host:port>, <host:port> 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=<rbac_enabled_public_pem_path> # EmbeddedKafkaRest: MDS Client configuration kafka.rest.confluent.metadata.bootstrap.server.urls=<host:port>, <host:port>, <host:port> kafka.rest.ssl.truststore.location=<truststore_location> kafka.rest.ssl.truststore.password=<password> kafka.rest.confluent.metadata.http.auth.credentials.provider=BASIC kafka.rest.confluent.metadata.basic.auth.user.info=<user:password> kafka.rest.confluent.metadata.server.urls.max.age.ms=60000 kafka.rest.client.confluent.metadata.server.urls.max.age.ms=60000
- Admin REST APIs Security
- REST Proxy Security
- Confluent Platform Demo (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
This will disable creating cluster links with that cluster as the destination, or “source initiated cluster links” with that cluster as the source. Note: this will 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 (it is on by default starting with Confluent Platform 7.0.0),
or to enable it on a running cluster where it was previously turned off, set the
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.
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.