Replicate Multi-Datacenter Topics Across Kafka Clusters in Confluent Platform

Confluent Replicator replicates topics from one Apache Kafka® cluster to another. In addition to copying the messages, Replicator creates topics as needed, preserving the topic configuration in the source cluster. This includes preserving the number of partitions, the replication factor, and any configuration overrides specified for individual topics. Replicator is implemented as a connector.

Important

For most use cases, the recommended approach is to use Cluster Linking and Schema Linking over Replicator.

Replicator features

Replicator provides these capabilities:

• Topic selection using whitelists, blacklists, and regular expressions.

• Dynamic topic creation in the destination cluster with matching partition counts, replication factors, and topic configuration overrides.

• Automatic resizing of topics when new partitions are added in the source cluster.

• Automatic reconfiguration of topics when topic configuration changes in the source cluster.

• Timestamp Preservation, Using provenance headers to prevent duplicates or cyclic message repetition, and Consumer Offset Translation (supported on Confluent Platform 5.0.1 and later).

• You can migrate from MirrorMaker to Replicator on existing datacenters (Confluent Platform 5.0.0 and later). Migration from MirrorMaker to Replicator is not supported in earlier versions of Confluent Platform (pre 5.5.0).

• At-least-once delivery. The Replicator connector guarantees that records are delivered at least once to the Kafka topic. If the connector restarts, the Kafka topic can contain duplicate records.

KRaft and ZooKeeper

As of Confluent Platform 8.0, ZooKeeper is no longer available for new deployments. As a best practice, use KRaft mode for new deployments. To learn more about running Kafka in KRaft mode, see KRaft Overview for Confluent Platform, KRaft Configuration for Confluent Platform, and the KRaft steps in the Platform Quick Start. To learn about migrating from older versions, see Migrate from ZooKeeper to KRaft on Confluent Platform.

Multi-datacenter use cases

Replicator can be deployed across clusters and in multiple datacenters. Multi-datacenter deployments enable use cases such as:

• Active-active geo-localized deployments: Lets you access a nearby datacenter and optimize your architecture for low latency and high performance.

• Active-passive disaster recovery deployments: In the event of a partial or complete datacenter disaster, allows failing over applications to use Confluent Platform in a different datacenter.

• Centralized analytics: Aggregate data from multiple Kafka clusters into one location for organization-wide analytics.

• Cloud migration: Use Kafka to synchronize data between on-premises applications and cloud deployments.

Replication of event records in Kafka topics from one cluster to another is the foundation of Confluent’s multi-datacenter architecture.

You can replicate data with Confluent Replicator or with the open source Kafka MirrorMaker. Use Replicator to replicate topic data and to migrate schemas in Schema Registry.

This documentation focuses on Replicator, including architecture, quick start tutorial, how to configure and run Replicator in different contexts, tuning and monitoring, and cross-cluster failover. This documentation also covers how to migrate from MirrorMaker to Replicator; see Migrate from Kafka MirrorMaker to Replicator in Confluent Platform.

Some of the general thinking on deployment strategies can also apply to MirrorMaker, but if you are primarily interested in MirrorMaker, see Mirroring data between clusters in the Kafka documentation.

Replicator architecture

The diagram below shows the Replicator architecture. Replicator uses the Kafka Connect APIs and workers for high availability, load balancing, and centralized management.

Replicator architecture

Tip

You can deploy Replicator near the destination cluster or the origin cluster, and either works. However, a best practice is to deploy Replicator closer to the destination cluster for reliability and performance over networks. Therefore, if the destination cluster is Confluent Cloud, deploy Replicator on an instance in the same region as your Confluent Cloud cluster. If the origin cluster does not permit external connections, you can deploy Replicator in the origin cluster. For more information, see Migrate Topics on Confluent Cloud Clusters.

Example Replicator deployment

In a typical multi-datacenter deployment, data from two geographically distributed Kafka clusters located in separate datacenters is aggregated in a separate cluster located in another datacenter. This documentation calls the origin of the copied data the source cluster and the target the destination cluster.

Each source cluster requires a separate instance of Replicator. For convenience, you can run them in the same Connect cluster, located in the aggregate datacenter.

Replication to an aggregate cluster

Guidelines for getting started with Replicator

Follow these guidelines to configure a multi-datacenter deployment using Replicator:

1. Use the Replicator quick start to set up replication between two Kafka clusters.

2. Learn how to install and configure Replicator and other Confluent Platform components in multi-datacenter environments.

3. Before running Replicator in production, make sure you read the tuning guide and the monitoring guide.

4. For a practical guide to designing and configuring multiple Kafka clusters to be resilient in case of a disaster scenario, see the Disaster Recovery whitepaper. The whitepaper outlines a plan for failover, failback, and ultimately successful recovery.

Replicator demos and examples

After completing the Replicator quick start, explore these working examples of Replicator in multi-datacenter deployments, which you can download from GitHub and run yourself. Refer to the diagram below to determine the Replicator examples that correspond to your deployment scenario.

Replicator examples by deployment scenario

1. Kafka on-premises to Kafka on-premises

   • Example: Replicate Data in an Active-Active Multi-DataCenter Deployment on Confluent Platform: fully automated example of an active-active multi-datacenter design with two instances of Replicator copying data bidirectionally between the datacenters.

   • Schema translation: transfers schemas stored in Schema Registry from one cluster to another using Replicator.

   • Confluent Platform demo: deploy a Kafka streaming extract, transform, load (ETL) pipeline, along with Replicator to replicate data.

2. Kafka on-premises to Confluent Cloud

   • Hybrid On-premises and Confluent Cloud: on-premises Kafka cluster and Confluent Cloud cluster, and data copied between them with Replicator.

   • Connect Cluster Backed to Destination: Replicator configuration with Kafka Connect backed to destination cluster.

   • On-premises to Cloud with Connect Backed to Origin: Replicator configuration with Kafka Connect backed to origin cluster.

3. Confluent Cloud to Confluent Cloud

   • Cloud to Cloud with Connect Backed to Destination: Replicator configuration with Kafka Connect backed to destination cluster.

   • Cloud to Cloud with Connect Backed to Origin: Replicator configuration with Kafka Connect backed to origin cluster.

   • Migrate Topics on Confluent Cloud Clusters: migrate topics from the origin Confluent Cloud cluster to the destination Confluent Cloud cluster.

Topic renaming

By default, Replicator uses the same topic name in both the source and destination clusters. If you replicate from a single cluster, this works fine. When copying data from multiple clusters to a single destination (that is, the aggregate use case), use a separate topic for each source cluster in case the source topics have configuration differences.

You can use the same Kafka cluster as the source and destination as long as you ensure that the replicated topic name is different. This pattern is not a best practice because you should generally prefer the built-in replication in Kafka within the same cluster, but it can be useful in some cases (for example, testing).

Replicator protects against circular replication by using provenance headers. This guarantees that if two Replicator instances are configured to run, one replicating from datacenter 1 (DC1) to datacenter 2 (DC2) and the second instance configured to replicate from DC2 to DC1, Replicator ensures that messages replicated to DC2 are not replicated back to DC1, and vice versa. As a result, Replicator safely runs in each direction.

Although Replicator can enable applications in different datacenters to access topics with the same names, design client applications with a topic naming strategy that considers several factors.

If you plan to have the same topic name span datacenters, be aware that in this configuration:

• Producers don’t wait for commit acknowledgment from the remote cluster, and Replicator asynchronously copies the data between datacenters after it has been committed locally.

• If there are producers in each datacenter writing to topics of the same name, Replicator provides no global ordering. This means there are no message ordering guarantees for data that originated from producers in different datacenters.

• If there are consumer groups in each datacenter with the same group ID reading from topics of the same name, in steady state, they reprocess the same event records in each datacenter.

You might not want the same topic name in each datacenter when:

• Replicator is running a version earlier than 5.0.1.

• Kafka brokers are running a version before Kafka 0.11 that does not yet support message headers.

• Kafka brokers are running Kafka version 0.11 or later but are configured with a log.message.format.version earlier than the required 2.0 for using headers.

• Client applications are not designed to handle topics with the same name across datacenters.

In these cases, refer to the appendix on “Topic Naming Strategies to Prevent Cyclic Repetition” in the Disaster Recovery whitepaper.

Periodic metadata updates

Replicator periodically checks topics in the source cluster for new topics that need to be replicated, and for configuration changes (for example, an increase in the number of partitions). The metadata.max.age.ms setting in the connector configuration controls how often Replicator checks. The default is two minutes, which is intended to provide reasonable responsiveness to configuration changes while ensuring that the connector does not add any unnecessary load on the source cluster. You can lower this setting to detect changes more quickly, but doing so isn’t advisable as long as topic creation or reconfiguration is relatively rare.

Security and ACL configurations

ACLs overview

Replicator supports communication with secure Kafka over TLS/SSL for both the source and destination clusters. Replicator also supports TLS/SSL or Simple Authentication and Security Layer (SASL) for authentication. You can use different security configurations on the source and destination clusters.

All properties documented here are additive (that is, you can apply both TLS/SSL encryption and SASL/PLAIN authentication properties) except for security.protocol. Use the following table to determine the correct value for this:

Encryption	Authentication	security.protocol
TLS/SSL	None	SSL
TLS/SSL	TLS/SSL	SSL
TLS/SSL	SASL	SASL_SSL
Plaintext	SASL	SASL_PLAINTEXT

You can configure Replicator connections to source and destination Kafka with:

• TLS/SSL Encryption. You can use different TLS/SSL configurations on the source and destination clusters.

• SSL Authentication.

• SASL/SCRAM.

• SASL/GSSAPI.

• SASL/PLAIN.

To configure security on the source cluster, see the connector configurations for Source Kafka: Security. To configure security on the destination cluster, see the connector configurations for Destination Kafka: Security and the general security configuration for Connect workers in Connect security.

See also

To see the required security configuration parameters for Replicator consolidated in one place, use the Docker Compose environments in GitHub confluentinc/examples.

When you use SASL or TLS/SSL authentication and access control lists (ACLs) are enabled on the source cluster, the destination cluster, or both, Replicator requires the ACLs described in the following sections.

This section provides ACL commands for both Confluent Platform (Confluent Platform CLI Command Reference) and Confluent Cloud.

For more information on configuring ACLs, see Use Access Control Lists (ACLs) for Authorization in Confluent Platform.

Principal users (Confluent Platform) and service accounts (Confluent Cloud)

The following sections show ACL commands for both Confluent Platform and Confluent Cloud.

On Confluent Platform, associate ACLs with a service principal.

On Confluent Cloud, associate ACLs with a Confluent Cloud service account.

To create a service account for Confluent Cloud, run the following command:

confluent iam service-account create <service-account-name> --description "<description of the service account>"

For example:

confluent iam service-account create my-first-cluster-test-acls --description "test ACLs on Cloud"

+-------------+----------------------------+
| Id          |                     123456 |
| Resource ID | ab-123abc                  |
| Name        | my-first-cluster-test-acls |
| Description | test ACLs on Cloud         |
+-------------+----------------------------+

Save the service account ID to use in the following commands to create the ACLs.

Tip

You can retrieve service account IDs by listing them with confluent iam service-account list.

ACLs for license management

For license management, you need the following ACLs:

Cluster	Resource	Operation
Destination (or other cluster configured with confluent.topic.bootstrap.servers)	TOPIC - _confluent-command	All

Confluent Platform

Commands to configure the preceding ACLs on Confluent Platform:

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<destination principal> --operation ALL --topic _confluent-command

Confluent Cloud

Commands to configure ACLs for the _confluent-command license topic on Confluent Cloud:

confluent kafka acl create --allow --service-account <service-account-id> --operations CREATE --topic _confluent-command
confluent kafka acl create --allow --service-account <service-account-id> --operations WRITE --topic _confluent-command
confluent kafka acl create --allow --service-account <service-account-id> --operations READ --topic _confluent-command
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE --topic _confluent-command
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE-CONFIGS --topic _confluent-command
confluent kafka acl create --allow --service-account <service-account-id> --operations ALTER-CONFIGS --topic _confluent-command

ACLs to read from the source cluster

To read from the source cluster, you need the following ACLs:

Cluster	Resource	Operation
Source	CLUSTER	Describe
Source	TOPIC - all topics Replicator replicates	Describe
Source	TOPIC - all topics Replicator replicates	Read
Source	GROUP - The consumer group name is determined by the Replicator name or by the src.consumer.group.id property	Read

Confluent Platform

Commands to configure the preceding ACLs on Confluent Platform:

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<source principal> --operation DESCRIBE --cluster
kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<source principal> --operation DESCRIBE --topic <source topic>
kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<source principal> --operation READ --topic <source topic>

Confluent Cloud

Commands to configure the preceding ACLs on Confluent Cloud:

confluent kafka cluster use <cluster-id>
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE --cluster-scope
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE --cluster <cluster-id>
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE --topic <source topic>
confluent kafka acl create --allow --service-account <service-account-id> --operations READ --topic <source topic>

ACLs to write to the destination cluster

To write to the destination cluster, you need the following ACLs:

Cluster	Resource	Operation
Destination	CLUSTER	Describe
Destination	TOPIC - all topics Replicator replicates	Describe
Destination	TOPIC - all topics Replicator replicates	Write
Destination	TOPIC - all topics Replicator replicates	Read

Confluent Platform

Commands to configure the preceding ACLs on Confluent Platform:

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<destination principal> --operation DESCRIBE --cluster
kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<destination principal> --operation DESCRIBE --topic <destination topic>
kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<destination principal> --operation WRITE --topic <destination topic>
kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<destination principal> --operation READ --topic <destination topic>

Confluent Cloud

Commands to configure the preceding ACLs on Confluent Cloud:

confluent kafka cluster use <cluster-id>
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE --cluster-scope
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE --cluster <cluster-id>
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE --topic <destination topic>
confluent kafka acl create --allow --service-account <service-account-id> --operations WRITE --topic <destination topic>
confluent kafka acl create --allow --service-account <service-account-id> --operations READ --topic <destination topic>

ACLs for topic creation and configuration sync

If using the topic creation and configuration sync features of Replicator (enabled by default), you need the following ACLs:

Cluster	Resource	Operation
Source	TOPIC - all topics Replicator replicates	DescribeConfigs
Destination	TOPIC - all topics Replicator replicates	Create
Destination	TOPIC - all topics Replicator replicates	DescribeConfigs
Destination	TOPIC - all topics Replicator replicates	AlterConfigs

For configuration options relating to topic creation and configuration sync, see Destination Topics.

Confluent Platform

Commands to configure the preceding ACLs on Confluent Platform:

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<source principal> --operation DESCRIBECONFIGS --topic <source topic>
kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<destination principal> --operation DESCRIBECONFIGS --topic <destination topic>
kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<destination principal> --operation CREATE --cluster
kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<destination principal> --operation ALTERCONFIGS --cluster

Confluent Cloud

Commands to configure the preceding ACLs on Confluent Cloud:

confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBECONFIGS --topic <source topic>
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBECONFIGS --topic <destination topic>
confluent kafka acl create --allow --service-account <service-account-id> --operations CREATE --cluster <cluster id>
confluent kafka acl create --allow --service-account <service-account-id> --operations ALTERCONFIGS --cluster <cluster id>
confluent kafka acl create --allow --service-account <service-account-id-destination> --operations DESCRIBE --cluster-scope

ACLs for offset translation

If using the offset translation feature of Replicator (enabled by default), you need the following ACLs:

Cluster	Resource	Operation
Source	TOPIC - __consumer_timestamps	All
Destination	GROUP - All translated consumer groups	All

For configuration options relating to offset translation, see Consumer offset translation.

Confluent Platform

Commands to configure the preceding ACLs on Confluent Platform:

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<source principal> --operation ALL --topic {_consumer_timestamps}
kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<source principal> --operation ALL --topic {<group name>}

Confluent Cloud

Commands to configure the preceding ACLs on Confluent Cloud:

confluent kafka acl create --allow --service-account <service-account-id> --operations CREATE --topic __consumer_timestamps
confluent kafka acl create --allow --service-account <service-account-id> --operations WRITE --topic __consumer_timestamps
confluent kafka acl create --allow --service-account <service-account-id> --operations READ --topic __consumer_timestamps
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE --topic __consumer_timestamps
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE-CONFIGS --topic __consumer_timestamps
confluent kafka acl create --allow --service-account <service-account-id> --operations ALTER-CONFIGS --topic __consumer_timestamps

confluent kafka acl create --allow --service-account <service-account-id> --operations CREATE --consumer-group <group name>
confluent kafka acl create --allow --service-account <service-account-id> --operations WRITE --consumer-group <group name>
confluent kafka acl create --allow --service-account <service-account-id> --operations READ --consumer-group <group name>
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE --consumer-group <group name>
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE-CONFIGS --consumer-group <group name>
confluent kafka acl create --allow --service-account <service-account-id> --operations ALTER-CONFIGS --consumer-group <group name>

ACLs for the timestamp interceptor

Any clients instrumented with the Replicator timestamp interceptor must also have the following ACLs:

Cluster	Resource	Operation
Source	TOPIC - __consumer_timestamps	Write
Source	TOPIC - __consumer_timestamps	Describe

Confluent Platform

Commands to configure the preceding ACLs on Confluent Platform:

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<source principal> --operation WRITE --topic __consumer_timestamps
kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<source principal> --operation DESCRIBE --topic __consumer_timestamps

Confluent Cloud

Commands to configure the preceding ACLs on Confluent Cloud:

confluent kafka acl create --allow --service-account <service-account-id> --operations WRITE --topic __consumer_timestamps
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE --topic __consumer_timestamps

ACLs for source offset management

If using the source offset management feature of Replicator (enabled by default), you need the following ACLs:

Cluster	Resource	Operation
Source	GROUP - The consumer group name is determined by the Replicator name or by the src.consumer.group.id property	All

For configuration options relating to offset management, see Offset Management in the Replicator configuration reference.

Confluent Platform

Commands to configure the preceding ACLs on Confluent Platform:

kafka-acls --bootstrap-server localhost:9092 --command-config adminclient-configs.conf --add --allow-principal User:<source principal> --operation ALL --group <group name>

Confluent Cloud

Commands to configure the preceding ACLs on Confluent Cloud:

confluent kafka acl create --allow --service-account <service-account-id> --operations CREATE --consumer-group <group name>
confluent kafka acl create --allow --service-account <service-account-id> --operations WRITE --consumer-group <group name>
confluent kafka acl create --allow --service-account <service-account-id> --operations READ --consumer-group <group name>
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE --consumer-group <group name>
confluent kafka acl create --allow --service-account <service-account-id> --operations DESCRIBE-CONFIGS --consumer-group <group name>
confluent kafka acl create --allow --service-account <service-account-id> --operations ALTER-CONFIGS --consumer-group <group name>

Replicator with RBAC

When using role-based access control (RBAC), Replicator clients must use token authentication as described in Configure Clients for SASL/OAUTHBEARER Authentication in Confluent Platform. Prefix these configurations with the usual Replicator prefixes of src.kafka. and dest.kafka.. The following example shows a configuration for RBAC-enabled source and destination clusters:

src.kafka.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  username="sourceUser \
  password="xxx" \
  metadataServerUrls="http://sourceHost:8090";
src.kafka.security.protocol=SASL_PLAINTEXT
src.kafka.sasl.mechanism=OAUTHBEARER
src.kafka.sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler

dest.kafka.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  username="destUser \
  password="xxx" \
  metadataServerUrls="http://destHost:8090";
dest.kafka.security.protocol=SASL_PLAINTEXT
dest.kafka.sasl.mechanism=OAUTHBEARER
dest.kafka.sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler

For the Replicator executable, don’t prefix these configurations. Place them in the files referenced by --consumer.config and --producer.config, as shown in the following example:

# in --consumer.config
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  username="sourceUser \
  password="xxx" \
  metadataServerUrls="http://sourceHost:8090";
security.protocol=SASL_PLAINTEXT
sasl.mechanism=OAUTHBEARER
sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler

# in --producer.config
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required \
  username="destUser \
  password="xxx" \
  metadataServerUrls="http://destHost:8090";
security.protocol=SASL_PLAINTEXT
sasl.mechanism=OAUTHBEARER
sasl.login.callback.handler.class=io.confluent.kafka.clients.plugins.auth.token.TokenUserLoginCallbackHandler

The preceding configuration requires that the Metadata Service (MDS) is running. For more information, see Configure Metadata Service (MDS) in Confluent Platform.

Important

The role bindings in this section are based on the following assumptions:

• There are two independent RBAC deployments, one in each datacenter.

• Each RBAC deployment supports all other services local to it, for example, Connect and Schema Registry.

• Replicator doesn’t replicate RBAC role bindings.

Replicator requires the role bindings listed below.

For the backing Connect cluster:

Cluster	Resource	Role
Destination	GROUP - name given by the group.id property	ResourceOwner
Destination	TOPIC - name given by the offset.storage.topic property (default connect-offsets)	ResourceOwner
Destination	TOPIC - name given by the config.storage.topic property (default connect-configs)	ResourceOwner
Destination	TOPIC - name given by the status.storage.topic property (default connect-status)	ResourceOwner

For license management:

Cluster	Resource	Role
Destination (or other cluster configured with confluent.topic.bootstrap.servers)	TOPIC - _confluent-command	DeveloperRead
Destination (or other cluster configured with confluent.topic.bootstrap.servers)	TOPIC - _confluent-command	DeveloperWrite
Destination (or other cluster configured with confluent.topic.bootstrap.servers)	TOPIC - _confluent-command	DeveloperManage

To read from the source cluster:

Cluster	Resource	Role
Source	TOPIC - all topics Replicator replicates	DeveloperRead
Source	TOPIC - all topics Replicator replicates	DeveloperManage

To write to the destination cluster:

Cluster	Resource	Role
Destination	TOPIC - all topics Replicator replicates	ResourceOwner

Important

If not using the topic configuration sync feature of Replicator (enabled by default), you can use the following roles in place of ResourceOwner:

Cluster	Resource	Role
Destination	TOPIC - all topics Replicator replicates	DeveloperRead
Destination	TOPIC - all topics Replicator replicates	DeveloperWrite
Destination	TOPIC - all topics Replicator replicates	DeveloperManage

If using the offset translation feature of Replicator (enabled by default):

Cluster	Resource	Role
Source	TOPIC - __consumer_timestamps	DeveloperRead
Source	TOPIC - __consumer_timestamps	DeveloperManage
Destination	GROUP - All translated consumer groups (if you don’t know these, use the literal * to allow all)	DeveloperRead

Also, any consumers on the source cluster using the Replicator timestamp interceptor require the following:

Cluster	Resource	Role
Source	TOPIC - __consumer_timestamps	DeveloperWrite
Source	TOPIC - __consumer_timestamps	DeveloperManage

For configuration options relating to offset translation, see Consumer offset translation.

If using the source offset management feature of Replicator (enabled by default):

Cluster	Resource	Role
Source	GROUP - The consumer group name is determined by the Replicator name or by the src.consumer.group.id property	ResourceOwner

If using the schema migration feature of Replicator (disabled by default):

Cluster	Resource	Role
Source	TOPIC - underlying Schema Registry topic (default _schemas)	DeveloperRead
Destination	CLUSTER - Schema Registry cluster	ClusterAdmin

For more information on configuring RBAC, see Use Role-Based Access Control (RBAC) for Authorization in Confluent Platform.

Replicating messages with schemas

Replicator doesn’t support an active-active Schema Registry setup. Replicator supports only migration, either one-time or continuous, from an active Schema Registry to a passive Schema Registry.

Schema Linking is available in preview on Confluent Platform, as described in Schema Linking on Confluent Platform. Schema Linking is the recommended method of migrating schemas. For migrating schemas from one Confluent Cloud cluster to another, use the cloud-specific Schema Linking.

For Confluent Platform releases earlier than 7.0.0, use Replicator with Schema Translation to migrate schemas from a self-managed cluster to a target cluster that is either self-managed or in Confluent Cloud. Schema Translation was first available in Confluent Platform 5.2.0.

To learn more about schema migration, see Use Schema Registry to Migrate Schemas in Confluent Platform and Schema Linking on Confluent Platform.

Schema ID Validation and Replicator

By default, Replicator is configured with topic.config.sync=true. If the source cluster has a topic with schema validation enabled (confluent.value.schema.validation=true), then Replicator copies this property to the destination cluster’s replicated topic.

When using Replicator to replicate data from one cluster of brokers to another, avoid the overhead of a second validation on the secondary cluster.

Therefore, either disable Schema ID Validation on the source cluster before replicating to the destination, or set topic.config.sync=false on Replicator and explicitly set the configurations you want in the destination cluster’s server.properties file.

Requirements for running Replicator

At a high level, Replicator works like a consumer group with the partitions of the replicated topics from the source cluster divided between the connector’s tasks. Replicator periodically polls the source cluster for changes to the configuration of replicated topics and the number of partitions, and updates the destination cluster accordingly by creating topics or updating configuration. For this to work correctly, you must meet the following requirements:

• The origin and destination clusters must be Kafka or Confluent Platform. For version compatibility, see connector interoperability.

• The Replicator version must match the Kafka Connect version it is deployed on. For instance, deploy Replicator 8.3 only to Kafka Connect 8.3.

• Replicator requires the ACLs described in Security and ACL configurations.

• The default topic configurations in the source and destination clusters must match. In general, aside from any broker-specific settings (such as broker.id), use the same broker configuration in both clusters.

• The destination Kafka cluster must have a capacity similar to the source cluster. Replicator preserves the replication factor of source topics, so the destination cluster must have at least as many brokers as the highest replication factor in use. If not, topic creation fails until the destination cluster has the capacity to support the same replication factor. In this case, the connector retries topic creation automatically, so replication begins as soon as the destination cluster has enough brokers.

• The dest.kafka.bootstrap.servers destination connection setting in the Replicator properties file must specify a single destination cluster, even when using multiple source clusters. For example, the “Replication to an aggregate cluster” figure in Example deployment shows two source clusters in different datacenters targeting a single aggregate destination cluster. The aggregate destination cluster must have a capacity similar to the total of all associated source clusters.

• On Confluent Platform versions 5.3.0 and later, Confluent Replicator requires the enterprise edition of Kafka Connect for Confluent Platform. Replicator does not support the community edition of Connect. You can install the enterprise edition of Connect as part of the Confluent Platform on-premises bundle, as described in Production Environments and in the Quick Start for Confluent Platform (choose self-managed Confluent Platform). Demos of the enterprise edition of Connect are available in the quick start and on Docker Hub at confluentinc/cp-server-connect.

• The timestamp-interceptor for consumers supports only Java clients, as described in Configuring the consumer for failover (timestamp preservation).

Tip

For best performance, run Replicator as close to the destination cluster as possible for a low-latency connection for Kafka Connect operations within Replicator.

Replicator version compatibility

For data transfer Replicator maintains the same compatibility matrix as Java clients, detailed in Kafka Java clients. However some Replicator features have different compatibility requirements:

• Schema Translation requires that both source and destination clusters are running Confluent 5.2.0 or later.

• Offset Translation requires that both source and destination clusters are running Confluent 5.1.0 or later.

• Automatic topic creation and config sync requires that the destination cluster is at a later version than the source cluster.

• The timestamp-interceptor for consumers supports only Java clients, as described in Configuring the consumer for failover (timestamp preservation).

• Newer versions of Replicator cannot be used to replicate data from early version Kafka clusters to Confluent Cloud. Specifically, Replicator version 5.4.0 or later cannot be used to replicate data from clusters Apache Kafka® v0.10.2 or earlier nor from Confluent Platform v3.2.0 or earlier, to Confluent Cloud. If you have clusters on these earlier versions, use Replicator 5.0.x to replicate to Confluent Cloud until you can upgrade. Keep in mind the following, and plan your upgrades accordingly:

  • Kafka Connect workers included in Confluent Platform 3.2 and later are compatible with any Kafka broker that is included in Confluent Platform 3.0 and later as documented in Cross-component compatibility.

  • Confluent Platform 5.0.x has an end-of-support date of July 31, 2020 as documented in Supported Versions and Interoperability for Confluent Platform.

Known issues, limitations, and best practices

• Replicator does not support the Kafka Connect first-class offsets APIs introduced by Kafka Improvement Proposal (KIP) 875. In particular, when using ReplicatorSourceConnector with offset.topic.commit=false, the GET /connectors/{name}/offsets REST endpoint returns null offsets for all partitions, and DELETE /connectors/{name}/offsets doesn’t clear Replicator’s stored offsets. Replicator still tracks and restores offsets correctly by using the internal __connect-offsets topic, but Replicator doesn’t support the external KIP-875 offset management endpoints. For descriptions of the REST endpoints mentioned, see Kafka Connect REST Interface for Confluent Platform.

• Using custom converters at the Replicator level to do schema migration is no longer supported. As a best practice, use Schema Linking plus Replicator with the byte array converter.

• Replicator supports Single Message Transformations (SMTs), but using them is not a best practice. Instead, use Apache Flink® or Kafka Streams, which scale better and are easier to debug.

• Don’t use Replicator for serialization changes. Instead, use ksqlDB. To learn more, see the documentation on ksqlDB and the tutorial on How to convert a stream’s serialization format on the Confluent Developer site.

• When running Replicator version 5.3.0 or later, set connect.protocol=eager because of a known issue in which using the default of connect.protocol=compatible or connect.protocol=sessioned can cause task rebalancing problems and duplicate records.

• If you encounter RecordTooLargeException when you use compressed records, set the record batch size for the Replicator producer to the highest possible value. When Replicator decompresses records while consuming from the source cluster, it checks the size of the uncompressed batch on the producer before recompressing it and can throw RecordTooLargeException. Setting the record batch size mitigates the exception, and compression proceeds as expected when the record is sent to the destination cluster.

• Replicator calculates the latency metric by subtracting the time the record was produced to the source from the time it was replicated on the destination. This works in the real-time case, when there is active production in the source cluster and the calculation happens in real time. However, if you are replicating old data, you see large latency due to the old record timestamps. In the historical data case, the latency does not indicate how long Replicator is taking to replicate data. It indicates how much time has passed since the original event record was produced, for the record that Replicator is currently replicating. As Replicator proceeds over historical data, the latency metric should decrease quickly.

• The Replicator lag metric reports the value NaN when no lag sample is recorded in a given time window. This can happen if you have limited production in the source cluster or if Replicator isn’t flushing data fast enough to the destination cluster, so Replicator can’t record enough samples in the given time window. The Java Management Extensions (JMX) metrics then report NaN for the Replicator metrics. NaN doesn’t necessarily mean the lag is zero. It means Replicator doesn’t have enough samples in the given time window to report lag.

Replicator connector

Replicator is implemented as a Kafka connector, and is listed in Supported Connectors. Some general connector information applies to Replicator, but most of the information you need to work with Replicator is in this Replicator-specific documentation.

Important

This connector is bundled natively with Confluent Platform. If you have Confluent Platform installed and running, there are no additional steps required to install.

If you are using Confluent Platform using only Confluent Community components, you can install the connector using the Confluent Marketplace Client (recommended) or you can manually download the ZIP file.

MirrorMaker

MirrorMaker is a standalone tool for copying data between two Kafka clusters. To learn more, see Mirroring data between clusters in the Kafka documentation.

Confluent supports MirrorMaker 2 as a standalone executable but not as a connector.

Confluent Replicator replicates topic configuration and data, and integrates with Kafka Connect and Confluent Control Center to improve availability, scalability, and ease of use. To learn more, see the quick start and Migrate from Kafka MirrorMaker to Replicator in Confluent Platform.

Related content

Blog posts and whitepapers

• Confluent Blog: 15 Things Every Apache Kafka Engineer Should Know About Confluent Replicator

• Confluent Blog: Enterprise Streaming Multi-Datacenter Replication using Apache Kafka

• Confluent Whitepaper: Disaster Recovery for Multi-Datacenter Apache Kafka Deployments

Kafka basics

• Confluent Developer: Kafka 101: Producers

• Confluent Developer: Kafka 101: Consumers

• Confluent Developer: Kafka 101: Replication

Getting started guides

• Tutorial: Replicate Data Across Kafka Clusters in Confluent Platform shows you how to get started using Replicator.

• Example: Replicate Data in an Active-Active Multi-DataCenter Deployment on Confluent Platform provides a Docker-based quick start.

Replicator references

• Replicator Configuration Reference for Confluent Platform provides a full reference for configuration.

• Confluent Platform and Cloud CLI Command Reference

Security

• Use Access Control Lists (ACLs) for Authorization in Confluent Platform

• ACL operations and allowed APIs

• Manage Confluent Cloud Accounts and Access

• Service Accounts for Confluent Cloud

Migration guides

• Migrate Topics on Confluent Cloud Clusters describes how to use Replicator to migrate topic data from one cloud cluster to another.

• Use Schema Registry to Migrate Schemas in Confluent Platform describes how to use Replicator to migrate a self-managed Schema Registry to Confluent Cloud.