# Use Replicator to Migrate Topics on Confluent Cloud Clusters Confluent Cloud offers multiple [Kafka cluster types](cluster-types.md#cloud-cluster-types) (Basic, Standard, Enterprise, Dedicated, and Freight). If you want to migrate topic data from one Confluent Cloud cluster to another, you have these two options for doing so: - Use Cluster Linking as described in [Migrate data with Cluster Linking on Confluent Cloud](../multi-cloud/cluster-linking/migrate-cc.md#cluster-linking-migrate-cc) (**preferred**) - Use Confluent Replicator, as described in the following sections (**legacy**) ## Replicator Deployment Options Replicator supports three deployment modes for migrating topic data: as a connector within a distributed Connect cluster, as a packaged executable on a VM, or as a packaged executable on Kubernetes. The modes are functionally equivalent; choose one based on your starting point. | Replicator Mode | Advantages and Scenarios | |---------------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | As a connector within a distributed Connect cluster (on a VM) | Ideal if you already have a Connect cluster in use with the destination cluster. | | As a packaged executable on a VM | Isolates three easy-to-use configuration files (for replicator, consumer, producer), and avoids having to explicitly configure the Connect cluster.

The [Quick Start](replicator-cloud-quickstart.md#cloud-replicator-quickstart) walks through an example of running Replicator as this type of executable. | | As a packaged executable on Kubernetes | Similar to running Replicator as a packaged executable on a VM, but might be easier to start as a single isolated task. Ideal if you are already managing tasks within Kubernetes. | ## Prerequisite Considerations - Confluent Platform is required for these workflows because Replicator is bundled with Confluent Platform. - Before you start topic migration, be sure to plan for network and security considerations, save non-topic metadata, and gather cluster information. These prerequisites are described in the following sections. - These procedures cover topic data only. You must migrate other aspects of the cluster separately, such as running connectors, ACLs, Streams apps, ksqlDB streams, and schemas. This is noted again at applicable points in the workflow. ### Choose a location The virtual machine (VM) or [Kubernetes (k8s)](https://kubernetes.io/) instances must be deployed in a virtual private cloud (VPC) that has network access to your Confluent Cloud clusters. As a best practice, run Replicator as close to the destination cluster as possible. For Confluent Cloud, this means the Replicator cluster should run in the same region as the destination Confluent Cloud deployment. If one of your Confluent Cloud clusters is VPC-peered, then Replicator must be deployed in one of the VPCs that is peered to Confluent Cloud. ### Determine cluster information Replicator configuration requires that you know: - API keys and secrets for the source and destination clusters. - The `bootstrap.servers` URL. To get this, navigate to **Cluster Settings** on the Cloud Console, and note down the URL next to `bootstrap.servers`. ### Save non-topic metadata and set up your new cluster before cutover Replicator does not sync ksqlDB jobs, schemas, ACLs, connectors, or service accounts to your new cluster. Make note of all such non-topic metadata that you want to migrate, and manually recreate these on the new cluster before turning on Replicator to replicate topic data. ### Configure consumers with timestamp interceptor for offset translation The `ConsumerTimestampsInterceptor` automatically translates consumer offsets using timestamps, so consumers resume in the destination cluster where they left off on the origin cluster. Configure this interceptor on your Java consumer applications. ```none interceptor.classes=io.confluent.connect.replicator.offsets.ConsumerTimestampsInterceptor ``` #### IMPORTANT - You can only set the `timestamp-interceptor` for consumers. - Do not set the interceptor for Replicator. - Use discretion when setting the interceptor on Connect workers; for the interceptor to be useful, the worker must be running sink connectors and these must use Apache Kafka® for their offset management. This interceptor is located under `kafka-connect-replicator` in (`timestamp-interceptor-.jar`), where `` is the Confluent Platform version. This JAR must be available on the `CLASSPATH` of the consumer. The location of the JAR file is dependent on your platform and type of [Confluent install](/platform/current/installation/index.html). For example, on macOS install with zip.tar, by default, `timestamp-interceptor-.jar` is in the Confluent directory in `/share/java/kafka-connect-replicator`. The `timestamp-interceptor` is located in the Confluent Maven repository: ```xml confluent https://packages.confluent.io/maven/ ``` For Maven projects, include the following dependency, where `release` is the Confluent Platform version: ```xml io.confluent timestamp-interceptor release ``` For example, for Confluent Platform version 7.5.0: ```none io.confluent timestamp-interceptor 7.5.0 ``` The `ConsumerTimestampsInterceptor` writes to a new topic `__consumer_timestamps` to preserve the required timestamp information for the offset translation. Replicator reads from this new topic and uses the timestamps to determine the correct offsets in the destination cluster. #### IMPORTANT The `ConsumerTimestampsInterceptor` is a producer to the `__consumer_timestamps` topic on the source cluster and as such requires appropriate security configurations. These should be provided with the `timestamps.producer.` prefix. for example, `timestamps.producer.security.protocol=SSL`. For more information on security configurations see: - [SSL Encryption](/platform/current/kafka/encryption.html#encryption-ssl-clients) - [SSL Authentication](/platform/current/kafka/authentication_ssl.html#authentication-ssl-clients) - [SASL/SCRAM](/platform/current/kafka/authentication_sasl/authentication_sasl_scram.html#sasl-scram-clients) - [SASL/GSSAPI](/platform/current/kafka/authentication_sasl/authentication_sasl_gssapi.html#sasl-gssapi-clients) - [SASL/PLAIN](/platform/current/kafka/authentication_sasl/authentication_sasl_plain.html#sasl-plain-clients) The interceptor also requires ACLs for the `__consumer_timestamps` topic. The consumer principal requires WRITE and DESCRIBE operations on the `__consumer_timestamps` topic. To learn more, see: - [Understanding Consumer Offset Translation](/platform/current/multi-dc-deployments/replicator/replicator-failover.html#consumer-offset-translation-feature) - Discussion on consumer offsets and timestamp preservation in the whitepaper on [Disaster Recovery for Multi-Datacenter Apache Kafka Deployments](https://www.confluent.io/white-paper/disaster-recovery-for-multi-datacenter-apache-kafka-deployments/). ### Configure ACLs Replicator must have authorization to read Kafka data from the origin cluster and write Kafka data in the destination Confluent Cloud cluster. Replicator should be run as a Confluent Cloud service account, not with super user credentials, so use the Confluent CLI to configure the appropriate ACLs for the service account id corresponding to Replicator in Confluent Cloud. For more details on Replicator ACLs, see [Security](/platform/current/multi-dc-deployments/replicator/index.html#replicator-security-overview). ```bash confluent kafka acl create --allow --service-account --operations create --topic confluent kafka acl create --allow --service-account --operations write --topic confluent kafka acl create --allow --service-account --operations read --topic confluent kafka acl create --allow --service-account --operations describe --topic confluent kafka acl create --allow --service-account --operations describe-configs --topic confluent kafka acl create --allow --service-account --operations alter-configs --topic confluent kafka acl create --allow --service-account --operations describe --cluster-scope ``` Additionally, you need to set the following license topic ACLs on the internal topic, `_confluent-command`. ```bash confluent kafka acl create --allow --service-account --operations read --topic _confluent-command confluent kafka acl create --allow --service-account --operations write --topic _confluent-command confluent kafka acl create --allow --service-account --operations describe --topic _confluent-command ``` The license topic should be created by default, but if for some reason this topic does not already exist, you must set CREATE along with DESCRIBE ACLs on the cluster. You already set DESCRIBE in the preceding ACL commands, but it is mentioned again to clarify its purpose. Set these ACLs on the `--cluster-scope`; this flag takes no arguments. It sets the ACLs on whichever cluster is in use. (To determine which cluster is in use, type `confluent kafka cluster list`. The cluster with the asterisk by it is in use. To use a different cluster, type `confluent kafka cluster use `.) ```bash confluent kafka acl create --allow --service-account --operations create --cluster-scope confluent kafka acl create --allow --service-account --operations describe --cluster-scope ``` ## Configure and Run Replicator to migrate topics After you complete the prerequisites, run Replicator to migrate topics. Choose one of the following methods to do so, based on your preference for [Replicator modes](#replicator-deploy-modes-cloud). - [Run Replicator as an executable](#cloud-cluster-migrate-replicator-mode-executable) - [Run Replicator as a connector](#cloud-cluster-migrate-replicator-mode-connector) - [Run Replicator Docker Container with Kubernetes](#cloud-cluster-migrate-replicator-mode-docker-k8) For any of these deployment methods, Replicator can be further tuned. See the [Replicator tuning guide](/platform/current/multi-dc-deployments/replicator/replicator-tuning.html). ### Run Replicator as an executable This applies to the VM and Kubernetes-based approaches. You can run Replicator executable on any virtual or cloud node or on a physical machine (desktop or laptop) that is connected to the network. #### Configure properties There are three configuration files for the executable (consumer, producer, and replication). The minimal configuration changes for these follow. * `consumer.properties` ```none ssl.endpoint.identification.algorithm=https sasl.mechanism=PLAIN bootstrap.servers= retry.backoff.ms=500 sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="" password=""; security.protocol=SASL_SSL ``` * `producer.properties` ```none ssl.endpoint.identification.algorithm=https sasl.mechanism=PLAIN bootstrap.servers= sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="" password=""; security.protocol=SASL_SSL ``` * `replication.properties` - No special configuration is required in `replication.properties`. #### Run Replicator Run the Replicator executable to migrate topics. ```bash ./bin/replicator --cluster.id replicator --consumer.config consumer.properties --producer.config producer.properties --topic.regex '.*' ``` To learn more, see [Run Replicator as an executable](/platform/current/multi-dc-deployments/replicator/replicator-run.html#run-crep-as-an-executable) and [Command Line Parameters of Replicator Executable](/platform/current/multi-dc-deployments/replicator/replicator-run.html#command-line-parameters-of-crep-executable) in the [Replicator documentation](/platform/current/multi-dc-deployments/index.html). ### Run Replicator as a connector The connector JSON should look like this: ```json { "name": "replicate-topic", "config": { "connector.class": "io.confluent.connect.replicator.ReplicatorSourceConnector", "key.converter": "io.confluent.connect.replicator.util.ByteArrayConverter", "value.converter": "io.confluent.connect.replicator.util.ByteArrayConverter", "src.kafka.ssl.endpoint.identification.algorithm":"https", "src.kafka.sasl.mechanism":"PLAIN", "src.kafka.request.timeout.ms":"20000", "src.kafka.bootstrap.servers":"", "src.kafka.retry.backoff.ms":"500", "src.kafka.sasl.jaas.config":"org.apache.kafka.common.security.plain.PlainLoginModule required username=\"\" password=\"\";", "src.kafka.security.protocol":"SASL_SSL", "dest.kafka.ssl.endpoint.identification.algorithm":"https", "dest.kafka.sasl.mechanism":"PLAIN", "dest.kafka.request.timeout.ms":"20000", "dest.kafka.bootstrap.servers":"", "dest.kafka.retry.backoff.ms":"500", "dest.kafka.sasl.jaas.config":"org.apache.kafka.common.security.plain.PlainLoginModule required username=\"\" password=\"\";", "dest.kafka.security.protocol":"SASL_SSL", "dest.topic.replication.factor":"3", "topic.regex":".*" } } ``` If you have not already done so, configure the distributed Connect cluster with the following configuration. ```none ssl.endpoint.identification.algorithm=https sasl.mechanism=PLAIN bootstrap.servers= sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="" password=""; security.protocol=SASL_SSL producer.ssl.endpoint.identification.algorithm=https producer.sasl.mechanism=PLAIN producer.bootstrap.servers= producer.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="" password=""; producer.security.protocol=SASL_SSL key.converter=org.apache.kafka.connect.storage.StringConverter value.converter=org.apache.kafka.connect.json.JsonConverter group.id=connect-replicator config.storage.topic=connect-configs1 offset.storage.topic=connect-offsets1 status.storage.topic=connect-statuses1 plugin.path=/share/java ``` To learn more, see [Run Replicator as a Connector](/platform/current/multi-dc-deployments/replicator/replicator-run.html#run-crep-as-a-connector) in the [Replicator documentation](/platform/current/multi-dc-deployments/index.html). ### Run Replicator Docker Container with Kubernetes 1. Delete existing secret (if it exists). ```bash kubectl delete secret replicator-secret-props ``` 2. Regenerate configurations, if changed. (See the [Quick Start](replicator-cloud-quickstart.md#cloud-replicator-quickstart).) 3. Upload the new secret. ```bash kubectl create secret generic replicator-secret-props --from-file=/tmp/replicator/ ``` 4. Reload pods. ```bash kubectl apply -f container/replicator-deployment.yaml ``` Here is an example `replicator-deployment.yaml`. ```yaml apiVersion: extensions/v1beta1 kind: Deployment metadata: name: repl-exec-connect-cluster spec: replicas: 1 template: metadata: labels: app: replicator-app spec: containers: - name: confluent-replicator image: confluentinc/cp-enterprise-replicator-executable env: - name: CLUSTER_ID value: "replicator-k8s" - name: CLUSTER_THREADS value: "1" - name: CONNECT_GROUP_ID value: "containerized-repl" # Note: This is to avoid _overlay errors_ . You could use /etc/replicator/ here instead. - name: REPLICATION_CONFIG value: "/etc/replicator-config/replication.properties" - name: PRODUCER_CONFIG value: "/etc/replicator-config/producer.properties" - name: CONSUMER_CONFIG value: "/etc/replicator-config/consumer.properties" volumeMounts: - name: replicator-properties mountPath: /etc/replicator-config/ volumes: - name: replicator-properties secret: secretName: "replicator-secret-props" defaultMode: 0666 ``` 5. Verify status. ```bash kubectl get pods kubectl logs -f ``` ## Cutover clients to new cluster After Replicator is set up, cut over clients to the new cluster as follows. 1. Confirm that data is replicating from the source to the destination by viewing the replication lag on the **Replication** tab on Confluent Cloud Console. 2. Stop producers. 3. Wait for consumer lag to go to zero. To monitor this, run `bin/kafka-consumer-groups` as described in the [consumer group command tool documentation](/platform/current/clients/consumer.html#ak-consumer-group-command-tool), or navigate to [Consumer Lag](../monitoring/monitor-lag.md#cloud-monitoring-lag) on the Cloud Console and select a consumer group. 4. Wait for Replicator lag to go to zero by looking at the Replicator consumer group lag, then restart consumers to point at the new cluster. - To monitor Replicator lag, run `bin/kafka-consumer-groups` as described in the [consumer group command tool documentation](/platform/current/clients/consumer.html#ak-consumer-group-command-tool), or navigate to [Consumer Lag](../monitoring/monitor-lag.md#cloud-monitoring-lag) on the Cloud Console and select a consumer group. - After restarting consumers, use `last-heartbeat-seconds-ago` as described in [Consumer Group Metrics](/platform/current/kafka/monitoring.html#kafka-monitoring-metrics-consumer-group) to verify that consumers are running. 5. Wait until all consumers are up. 6. Start producers in the new cluster. Monitor `outgoing-byte-rate` as described in [producer outgoing-byte-rate metrics](/platform/current/kafka/monitoring.html#producer-global-request-metrics). 7. Monitor lag on the destination cluster. - To monitor this, use the activity monitor for the cluster on the Cloud Console.  - You can also run `bin/kafka-consumer-groups` on the destination cluster as described in [Consumer Group Metrics](/platform/current/kafka/monitoring.html#kafka-monitoring-metrics-consumer-group). ## Related content ### Cluster Linking Cluster Linking is the preferred method of copying data on Confluent Cloud. (Replicator is the legacy option.) To learn more about migrating data with Cluster Linking, see: - [Geo-replication with Cluster Linking on Confluent Cloud](../multi-cloud/cluster-linking/index.md#cloud-cluster-linking) - [Migrate data with Cluster Linking on Confluent Cloud](../multi-cloud/cluster-linking/migrate-cc.md#cluster-linking-migrate-cc) ### Replicator - [Replicator Quick Start to Migrate Topic Data on Confluent Cloud](replicator-cloud-quickstart.md#cloud-replicator-quickstart) - [Confluent Replicator to Confluent Cloud Configurations](/platform/current/tutorials/examples/ccloud/docs/replicator-to-cloud-configuration-types.html) - [On-Premises to Confluent Cloud example](/platform/current/tutorials/cp-demo/docs/index.html) - [Multi-DC Deployment Architectures](/platform/current/multi-dc-deployments/index.html) - [Replicator for Multi-Datacenter Replication](/platform/current/multi-dc-deployments/replicator/index.html) - [Tutorial: Replicating Data Between Clusters](/platform/current/multi-dc-deployments/replicator/replicator-quickstart.html#replicator-quickstart) - [Configure and Run Replicator](/platform/current/multi-dc-deployments/replicator/replicator-run.html#replicator-run) - [Disaster Recovery for Multi-Datacenter Apache Kafka Deployments](https://www.confluent.io/white-paper/disaster-recovery-for-multi-datacenter-apache-kafka-deployments/).