Cluster Linking Automatic Failover Tutorial for Active-Passive Deployments

Automatic Failover lets your clients automatically follow the active cluster through a single global bootstrap URL when you fail over to the DR cluster with one command, minimizing downtime and data loss.

Important

Automatic Failover is a Limited Availability feature in Confluent Cloud.

You can use Automatic Failover in production except under any of the following disqualifying conditions:

• Region restriction: Clusters located in AWS us-west-2 (Oregon) are excluded.

• Consumer group scale limit: The cluster cannot exceed 1,000 active consumer groups.

Pricing and billing: Billing starts on December 1, 2026, as described on the Confluent Cloud product information website. Automatic Failover is available at no additional charge before December 1, 2026.

With Automatic Failover and Cluster Linking, you can implement an active-passive, cross-region Disaster Recovery (DR) setup with a low recovery time objective (RTO) and low recovery point objective (RPO). Failover from the primary to the secondary cluster requires only a single command on the “cluster pair”. Your clients automatically switch over to the secondary cluster without any manual intervention.

Automatic Failover is currently available on active-passive Cluster Linking setups.

What the tutorial covers

You build two Apache Kafka® clusters, use Cluster Linking to replicate data across the clusters, and pair them with Automatic Failover. You get a single global bootstrap URL for the cluster pair that your clients connect to. Next, you simulate normal operations by adding data load to the pair with a sample producer and consumer. Finally, you trigger a failover and fail back.

The examples in this tutorial use dedicated cluster with public networking and API Keys but more configurations are supported. For details on supported configurations, see the Prerequisites and considerations.

Initially, the global bootstrap URL of the cluster pair resolves to the primary cluster. Clients produce to and consume from “source” topics on the primary cluster, with private networking if configured.

In this steady state, the secondary cluster acts as a standby with “mirror topics” automatically managed by Cluster Linking.

When a disaster hits, the producers, consumers, and cluster link cannot interact with the original cluster.

To model your recovery strategy in a disaster, trigger a failover command on the cluster pair. The global bootstrap URL now resolves to the secondary cluster, which becomes the active cluster. This command also reverses the Cluster Linking data replication flow. Produce and consume target the secondary cluster, with private networking if configured.

The DR cluster becomes the new source of truth.

Prerequisites and considerations

Before you start working through this tutorial, verify you have the following prerequisites and understand the operational considerations for Automatic Failover.

Cluster and account prerequisites

• Cluster types: Dedicated or Enterprise clusters only on both primary and secondary clusters. Basic, Standard, and Freight are not supported. To learn more, see cluster types.

• Environment and organization scope: Both clusters must belong to the same Confluent Cloud organization, but can be in different environments.

• Cluster sizing and parity: Ideally, you should size both clusters identically with same Confluent Unit for Kafka (CKU), so the secondary cluster can absorb full primary traffic during a failover. If you use Enterprise clusters, their elastic CKUs (eCKUs) can scale down to a minimum while the DR cluster is passive. Dedicated clusters use a fixed number of provisioned CKUs.

• Secondary cluster initial state: The secondary cluster must be entirely empty of producers and consumers at the time of pairing.

• Bidirectional Cluster link: Automatic Failover requires a bidirectional cluster link between the primary and secondary clusters. You’ll create this link manually as part of this tutorial.

• Configure the cluster link to replicate all topics: Automatic Failover requires that the cluster link replicates all topics it manages. Topics not managed by the cluster link (like confluent-audit-log-events) are not replicated. For a full list of topics not replicated by the cluster link, see Topics not mirrored. Self-managed connector offsets are replicated and stored in regular topics. Fully-managed connector offsets are not replicated by the cluster link but you can retrieve them programmatically, as described in Manage Offsets for Fully Managed Connectors in Confluent Cloud.

• Outbound Cluster Link connections between clusters: The cluster pair’s bidirectional cluster link requires connection.mode=OUTBOUND (the default) on both clusters. INBOUND mode is not supported.

• No additional destination cluster link: Primary and secondary clusters should not be a destination of another cluster link and should not have another bidirectional cluster link. Using the global endpoint as part of a cluster link configuration causes all topics to fail.

• No additional source-initiated cluster link: A primary or secondary cluster should not be the source of a separate, source-initiated cluster link (connection.mode=INBOUND on the other cluster) to a third cluster. That link’s replication stops working whenever the shared cluster is the inactive member of the switchover pair.

• Transactions: Cluster link does not support transactions with exactly-once semantics and can leave hung transactions on failover. Because replication is asynchronous and partitions are replicated independently, transaction completion markers can reach some partitions but not others before failover occurs. To learn more, see Kafka transactions such as “exactly once” semantics not supported on mirror topics.

• Not all mirror topics properties are synchronized: Cluster link replicates critical topic configuration properties. For a list of configurations that are not replicated, see Mirror topic configurations not synced.

Networking and connectivity prerequisites

• Network parity: If using private networking, you must pair endpoints of the same type on both clusters. For example, pair a Private Link endpoint from your primary cluster to another Private Link endpoint on your secondary cluster.

• Cross-region connectivity: You are responsible for provisioning the cross-region network path between clusters. For example, clients in region A must be able to reach your Private Link / Private Service Connect in region B.

• Forward proxy SNI support: If client traffic passes through a forward proxy, the proxy must pass the TLS Server Name Indication (SNI) extension unmodified. Stripping or rewriting SNI hostnames breaks failover.

• Public DNS resolution path: Your DNS resolution path (including split-horizon or internal DNS resolvers) must be capable of resolving the global bootstrap endpoint over public DNS. For the global bootstrap URL patterns to allow, see Construct the global bootstrap endpoint URL.

Client and security prerequisites

• Client library version: Kafka client versions 4.0 or later. Clients written in Java, C/C++ (librdkafka), Python, Go, .Net/C#, JavaScript / Node.js are supported. REST API clients are not supported.

• Client authentication types: Clients must authenticate using OAuth or Global API keys. Cluster-scoped API keys cannot fail over. This tutorial uses API keys as an example. OAuth can also be used and pointers on how to do that are provided in this tutorial.

• Client and user authorization: When authorizing clients with role-based access control (RBAC), you must manually configure role bindings on both primary and secondary clusters for failover to work. When using ACLs, you can define ACLs on the primary cluster and have Cluster Linking automatically synchronize the ACLs to the secondary cluster.

• Encryption / security: Bring Your Own Key (BYOK) storage encryption, if enabled, must be configured on both primary and secondary clusters before pairing.

• Permissions required to read switchover pairs and endpoints: Principals that only read switchover pairs and switchover endpoints, such as to run describe or list commands or to check the active member, need the EnvironmentOperator, EnvironmentAdmin, OrganizationOperator, or OrganizationAdmin role. Switchover pairs are scoped to the environment, not to a cluster, so the CloudClusterAdmin role alone returns a 403 RBAC error, particularly when the clusters are in different environments. Grant EnvironmentOperator on the environment that hosts the pair.

• Permissions required to set up resources: You must create all Identity and Access Management (IAM), cluster, and cluster pair resources with OrganizationAdmin or EnvironmentAdmin permissions. The cluster link resource requires the CloudClusterAdmin role or higher. Depending on the interface that you use to run this tutorial, these permissions take different forms. For the Limited Availability version, the cluster pair and failover API does not support OAuth authentication.

Confluent CLI

Log in to the Confluent CLI to authenticate and set permissions associated with your user account. Make sure you use Confluent CLI version 4.78.0 or later.

confluent login

Confirm which binary you’re running and its version:

which confluent
confluent version

If needed, install or upgrade the Confluent CLI to get the latest version with the confluent switchover commands.

REST API

Set authentication and permissions using either a Confluent Cloud API key linked to a service account, or with an OAuth token from your identity provider linked to an identity pool.

To create resources in this tutorial, you need a Confluent Cloud API key with Cloud Resource management or Global scope. Its service account must have OrganizationAdmin or EnvironmentAdmin permissions. This tutorial assumes that you already have one, and refers to its ID as $CLOUD_API_KEY.

Terraform

Terraform manages the switchover pair and endpoint resources, but cannot trigger a failover, restore, or failback. To run those operations, use the Confluent CLI or REST API tab. Plan to use one of those interfaces alongside Terraform for the failover and failback sections of this tutorial.

Set credentials for the confluentinc/confluent Terraform provider via the CONFLUENT_CLOUD_API_KEY / CONFLUENT_CLOUD_API_SECRET environment variables (or the provider block’s cloud_api_key/cloud_api_secret arguments). They’re sent as HTTP Basic auth on every control-plane call, switchover included.

terraform {
  required_providers {
    confluent = {
      source  = "confluentinc/confluent"
      version = ">= 2.88.0"   # first version to include the switchover resources
    }
  }
}

provider "confluent" {
  endpoint = "https://api.confluent.cloud"
}

# Input variables referenced as var.* throughout this tutorial
variable "org" {}
variable "primary_env" {}
variable "secondary_env" {}
variable "pair_env" {}
variable "client_principal_id" {}

Declare these once in the provider-setup configuration so every later tab’s snippets work, and pass values with -var or TF_VAR_<name> environment variables.

Use the same Confluent Cloud API Key described in the REST API tab (Cloud Resource management or Global scope, service account with OrganizationAdmin or EnvironmentAdmin permissions).

Schema and governance constraints

• Schema Registry DR: Schema Registry failover is not managed by Automatic Failover. You must set up schema DR (such as schema linking) independently.

• Schema Linking and mirror topics: To use a mirror topic that has a schema with Confluent Cloud ksqlDB, broker-side schema ID validation, or the topic viewer, make sure that Schema Linking puts the schema in the default context of the Confluent Cloud Schema Registry. To learn more, see How Schemas work with Mirror Topics.

• Manual schema validation setup: Cluster Linking does not replicate topic-level schema validation configurations (confluent.key.schema.validation and confluent.value.schema.validation). You must apply these manually to the secondary cluster.

• Cloud tags gap: Cloud tags, including tags for client-side field level encryption (CSFLE), do not automatically synchronize across paired clusters. ACLs and cluster configurations do synchronize. Service accounts and Global API keys are organization-scoped, so they apply to both clusters without syncing.

Other products

• Managed ecosystem services: You must manually fail over managed connectors, Apache Flink®, and ksqlDB applications running on paired clusters. These are not a part of Automatic Failover.

• Stream processing applications: Automatic Failover does not fail over ksqlDB and Kafka Streams applications automatically. In an Automatic Failover switchover pair, applications can’t run against the destination cluster, so you can’t run a second ksqlDB or Kafka Streams application against the mirror topics there. Plan for re-running the application and reprocessing after failover, and factor the state rebuild time into your RTO. To rebuild a KTable after failover:

  • Use Tiered Storage to make sure that data is always retained. If you do this, the mirror topic has all history, and the KTable can be built from history. Configure this from the beginning so the input topic retention does not remove needed historical data.

  • Another option is to configure compaction on the topic on the primary cluster from the beginning. You can use the min.compaction.lag.ms and max.compaction.lag.ms configurations to preserve the full X-day history and then compact anything older. In other words, the compaction only runs in the part of the data that is older than the configured lag of X days.

  • If the KTable’s input topic has already purged data due to the retention period, rebuild from the topic on the secondary cluster (the former mirror topic) and lose the history in the KTable.

Performance limitations

Failover performance is largely dominated by cluster link’s performance to reverse links.

• Unplanned failover performance grows with topic count and to a minor extent partition count. Order of magnitude is in ~50s+1s/Topic.

• Planned failover performance scales in a similar way as unplanned failovers with topic count and partition count but also scales with the time it takes to get consumer lags down to 0 and the total number of consumer groups. During the EA, limit the number of consumer groups to under 1000.

Set up Automatic Failover

Automatic Failover pairs your clusters across regions, configures DR-optimized settings, and provisions a single global bootstrap URL using an existing cluster link.

The commands below use OAuth to authenticate the cluster link and the sample producer and consumer clients because a single OAuth identity can work across both clusters without managing per-cluster credentials. Global API keys, not cluster-scoped API keys, are also a supported option for both the cluster link and your client applications.

Create primary and secondary clusters

The primary cluster is the source cluster, and the secondary cluster is the DR cluster (destination). Both clusters must be Dedicated or Enterprise clusters. Create two Dedicated or Enterprise clusters in the respective regions in the same organization and same or different environments:

Confluent CLI

1. Create the primary cluster in your primary region:

   confluent kafka cluster create my-primary-cluster \
   --cloud aws --region us-east-1 --type dedicated --cku 2 \
   --availability multi-zone --environment $PRIMARY_ENV
   

   where: $PRIMARY_ENV is your Environment id hosting this cluster.

2. Create the secondary cluster in your secondary region:

   confluent kafka cluster create my-secondary-cluster \
   --cloud aws --region us-west-1 --type dedicated --cku 2 \
   --availability multi-zone --environment $SECONDARY_ENV
   

   where: $SECONDARY_ENV is your Environment id hosting this cluster.

REST API

1. Create the primary cluster in your primary region:

   curl -sS "https://api.confluent.cloud/cmk/v2/clusters" -X POST \
   -u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
   -d '{
      "spec": {
         "display_name": "my-primary-cluster",
         "cloud": "AWS",
         "region": "us-east-1",
         "availability": "MULTI_ZONE",
         "config": {"kind": "Dedicated", "cku": 2},
         "environment": {"id": "'"$PRIMARY_ENV"'"}
      }
   }' | jq .
   

   where:

   • $CLOUD_API_KEY and $CLOUD_API_SECRET are your Cloud API key ID and secret.

   • $PRIMARY_ENV is the environment ID hosting this cluster.

2. Create the secondary cluster in your secondary region. $SECONDARY_ENV is the Environment id hosting this cluster:

   curl -sS "https://api.confluent.cloud/cmk/v2/clusters" -X POST \
   -u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
   -d '{
      "spec": {
         "display_name": "my-secondary-cluster",
         "cloud": "AWS",
         "region": "us-west-1",
         "availability": "MULTI_ZONE",
         "config": {"kind": "Dedicated", "cku": 2},
         "environment": {"id": "'"$SECONDARY_ENV"'"}
      }
   }' | jq .
   

   where:

   • $CLOUD_API_KEY and $CLOUD_API_SECRET are your Cloud API key ID and secret.

   • $SECONDARY_ENV is the environment ID hosting this cluster.

Terraform

Create the primary and secondary clusters using the confluent_kafka_cluster resource.

resource "confluent_kafka_cluster" "primary" {
  display_name = "my-primary-cluster"
  availability = "MULTI_ZONE"
  cloud        = "AWS"
  region       = "us-east-1"

  dedicated {
    cku = 2
  }

  environment {
    id = var.primary_env
  }
}

resource "confluent_kafka_cluster" "secondary" {
  display_name = "my-secondary-cluster"
  availability = "MULTI_ZONE"
  cloud        = "AWS"
  region       = "us-west-1"

  dedicated {
    cku = 2
  }

  environment {
    id = var.secondary_env
  }
}

Apply the Terraform configuration to generate a plan, and run it.

terraform apply

where: var.primary_env and var.secondary_env map to $PRIMARY_ENV and $SECONDARY_ENV.

Create the cluster link

Set up identity, grant permissions, and create a bidirectional cluster link between the primary and secondary clusters.

You can use API keys or OAuth to authenticate and authorize your cluster link.

Create the cluster link identity

Create the service account for the cluster link identity.

Confluent CLI

Create the service account for the cluster link identity.

confluent iam service-account create dr-link-sa --description "Cluster link identity for DR pair"

REST API

Create the service account for the cluster link identity.

curl -sS "https://api.confluent.cloud/iam/v2/service-accounts" -X POST \
-u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
-d '{
   "display_name": "dr-link-sa",
   "description": "Cluster link identity for DR pair"
}' | jq .

where: $CLOUD_API_KEY is your Cloud API key ID.

Terraform

Create the service account for the cluster link identity using the confluent_service_account resource.

resource "confluent_service_account" "dr_link_sa" {
  display_name = "dr-link-sa"
  description  = "Cluster link identity for DR pair"
}

Apply the Terraform configuration to generate a plan, and run it.

terraform apply

If using OAuth, create an identity pool for the cluster link identity.

Grant the cluster link identity the CloudClusterAdmin role on both clusters

Grant the cluster link identity the CloudClusterAdmin role on both the primary and secondary clusters so it can manage link and topic resources during failover:

Confluent CLI

1. Grant the cluster link identity the CloudClusterAdmin role on both clusters.

   confluent iam rbac role-binding create --principal User:$CL_SA \
   --role CloudClusterAdmin --environment $PRIMARY_ENV --cloud-cluster $PRIMARY_LKC
   

   confluent iam rbac role-binding create --principal User:$CL_SA \
   --role CloudClusterAdmin --environment $SECONDARY_ENV --cloud-cluster $SECONDARY_LKC
   

   where:

   • $CL_SA is the Service Account ID of the cluster link.

   • $PRIMARY_LKC and $SECONDARY_LKC are the Cluster ID of your respective clusters.

   • $PRIMARY_ENV and $SECONDARY_ENV are the Environment ID of your respective clusters.

2. Create a Global API key scoped to cluster link identity

   confluent api-key create --resource global --service-account $CL_SA \
   --description "dr-link global key" -o json
   

   where: $CL_SA is the service account ID of the cluster link.

REST API

1. Grant the cluster link identity the CloudClusterAdmin role on both clusters.

   curl -sS "https://api.confluent.cloud/iam/v2/role-bindings" -X POST \
   -u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
   -d '{
      "principal": "User:'"$CL_SA"'",
      "role_name": "CloudClusterAdmin",
      "crn_pattern": "crn://confluent.cloud/organization='"$ORG"'/environment='"$PRIMARY_ENV"'/cloud-cluster='"$PRIMARY_LKC"'"
   }' | jq .
   

   curl -sS "https://api.confluent.cloud/iam/v2/role-bindings" -X POST \
   -u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
   -d '{
      "principal": "User:'"$CL_SA"'",
      "role_name": "CloudClusterAdmin",
      "crn_pattern": "crn://confluent.cloud/organization='"$ORG"'/environment='"$SECONDARY_ENV"'/cloud-cluster='"$SECONDARY_LKC"'"
   }' | jq .
   

   where:

   • $CLOUD_API_KEY is your Cloud API key ID.

   • $CLOUD_API_SECRET is your cloud API secret.

   • $ORG is your organization ID.

   • $CL_SA is the service account ID of the cluster link.

   • $PRIMARY_LKC and $SECONDARY_LKC are the cluster ID of your respective clusters.

   • $PRIMARY_ENV and $SECONDARY_ENV are the environment ID of your respective clusters.

2. Create a Global API key scoped to cluster link identity

   curl -sS --request POST \
   --url 'https://api.confluent.cloud/iam/v2/api-keys' \
   -u "$CLOUD_API_KEY:$CLOUD_API_SECRET" \
   --header 'Content-Type: application/json' \
   --data '{
      "spec": {
         "display_name": "dr-link global key",
         "description": "dr-link global key",
         "owner": {"id": "'"$CL_SA"'"},
         "resource": {"id": "global"}
      }
   }' | jq .
   

   where:

   • $CLOUD_API_KEY is your Cloud API key ID.

   • $CLOUD_API_SECRET is your Cloud API secret.

   • $CL_SA is the Service Account id of the cluster link.

Terraform

1. Grant the cluster link identity the CloudClusterAdmin role on both clusters using the confluent_role_binding resource.

   resource "confluent_role_binding" "dr_link_sa_primary" {
   principal   = "User:${confluent_service_account.dr_link_sa.id}"
   role_name   = "CloudClusterAdmin"
   crn_pattern = confluent_kafka_cluster.primary.rbac_crn
   }
   
   resource "confluent_role_binding" "dr_link_sa_secondary" {
   principal   = "User:${confluent_service_account.dr_link_sa.id}"
   role_name   = "CloudClusterAdmin"
   crn_pattern = confluent_kafka_cluster.secondary.rbac_crn
   }
   

   Apply the Terraform configuration to generate a plan, and run it.

   terraform apply
   

2. Create the Global API key for the cluster link identity with the confluent_api_key resource, using a managed_resource block of kind Global.

   resource "confluent_api_key" "dr_link_global_key" {
   display_name = "dr-link global key"
   description  = "dr-link global key"
   
   owner {
      id          = confluent_service_account.dr_link_sa.id
      api_version = confluent_service_account.dr_link_sa.api_version
      kind        = confluent_service_account.dr_link_sa.kind
   }
   
   managed_resource {
      id          = "global"
      api_version = "global/v1"
      kind        = "Global"
   }
   
   lifecycle {
      prevent_destroy = true
   }
   }
   

   The lifecycle block protects this key from accidental deletion, because both confluent_cluster_link resources (and the data source lookups later in this tutorial) authenticate with it: destroying it would break the bidirectional link between both clusters. To intentionally destroy this resource, remove the lifecycle block first.

   Apply the Terraform configuration to generate a plan, and run it.

   terraform apply
   

Note

The cluster link identity (service account or identity pool) on the secondary cluster must have permissions to LIST and DESCRIBE all topics and mirror topics. Insufficient permissions can cause data loss.

A new role binding can take a few seconds to take effect. If a command fails with an authorization error (such as cluster authorization failure) right after you create a role binding, wait a few seconds and retry. For more information about role bindings, see Manage RBAC role bindings on Confluent Cloud.

Specify configurations for the cluster link

The cluster link must be bidirectional. If you create the link from only one side, the other side is left as a mismatched, unmanaged link object.

As a best practice, configure the cluster link with these recommended settings to ensure optimal DR performance and consumer offset preservation during failover and failback:

Recommended DR Cluster Link Configurations

Setting	Recommended Value	Purpose
Auto-create Mirror Topics	Enabled (All topics)	Ensures new topics created on the active cluster are automatically protected by DR. Set with auto.create.mirror.topics.enable=true. The default is false.
Auto-create Mirror Topics Filters	INCLUDE filter with name *	Mirrors all current and future topics on the active cluster, except topics that Cluster Linking never mirrors, such as internal topics and the Schema Registry schemas topic. Set with auto.create.mirror.topics.filters and requires auto.create.mirror.topics.enable=true.
Consumer Offset Sync	Enabled (LOCAL_MIRROR)	Preserves consumer group positions across failover, resuming execution near the last committed offset. Set with consumer.offset.sync.enable=true. The default is false.
ACL Sync	Enabled	Ensures that ACLs are synchronized across the active and DR clusters. Set with acl.sync.enable=true. The default is false.

Create a single link configuration file for both the primary cluster and secondary clusters. This configuration syncs offsets for all consumer groups (*) for these mirror topics.

Copy the following lines into a file called dr-link.config, and replace the placeholder variables with your details.

API KEY

link.mode=BIDIRECTIONAL

auto.create.mirror.topics.enable=true
auto.create.mirror.topics.filters={"topicFilters": [{"name": "*","patternType": "LITERAL","filterType": "INCLUDE"}]}
topic.config.sync.ms=1000

consumer.offset.sync.enable=true
consumer.offset.group.filters={"groupFilters": [{"name": "*","patternType": "LITERAL","filterType": "INCLUDE","topicTypes": ["LOCAL_MIRROR"]}]}
consumer.offset.sync.ms=1000

acl.sync.enable=true
acl.filters={ "aclFilters": [ { "resourceFilter": { "resourceType": "any", "patternType": "any" }, "accessFilter": { "operation": "any", "permissionType": "any" } } ] }
acl.sync.ms=1000

security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$CL_API_KEY" password="$CL_API_KEY_SECRET";

local.security.protocol=SASL_SSL
local.sasl.mechanism=PLAIN
local.sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username="$CL_API_KEY" password="$CL_API_KEY_SECRET";

where:

• $CL_API_KEY is your cluster link API key ID.

• $CL_API_KEY_SECRET is your cluster link API key secret.

OAUTH

link.mode=BIDIRECTIONAL

auto.create.mirror.topics.enable=true
auto.create.mirror.topics.filters={"topicFilters": [{"name": "*","patternType": "LITERAL","filterType": "INCLUDE"}]}
topic.config.sync.ms=1000

consumer.offset.sync.enable=true
consumer.offset.group.filters={"groupFilters": [{"name": "*","patternType": "LITERAL","filterType": "INCLUDE","topicTypes": ["LOCAL_MIRROR"]}]}
consumer.offset.sync.ms=1000

acl.sync.enable=true
acl.filters={ "aclFilters": [ { "resourceFilter": { "resourceType": "any", "patternType": "any" }, "accessFilter": { "operation": "any", "permissionType": "any" } } ] }
acl.sync.ms=1000

security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler
sasl.oauthbearer.token.endpoint.url=$OAUTH_TOKEN_ENDPOINT
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId='$OAUTH_CLIENT_ID' clientSecret='$OAUTH_CLIENT_SECRET' scope='' extension_identityPoolId='$OAUTH_IDENTITY_POOL_ID';

local.security.protocol=SASL_SSL
local.sasl.mechanism=OAUTHBEARER
local.sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler
local.sasl.oauthbearer.token.endpoint.url=$OAUTH_TOKEN_ENDPOINT
local.sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId='$OAUTH_CLIENT_ID' clientSecret='$OAUTH_CLIENT_SECRET' scope='' extension_identityPoolId='$OAUTH_IDENTITY_POOL_ID';

where:

• $OAUTH_TOKEN_ENDPOINT is your OAuth token endpoint.

• $OAUTH_CLIENT_ID is your OAuth client ID.

• $OAUTH_CLIENT_SECRET is your OAuth client secret.

• $OAUTH_IDENTITY_POOL_ID is your OAuth identity pool ID.

Note

extension_logicalCluster is no longer required and must even be removed for the same config file to be used in both cluster link directions.

Create a bidirectional cluster link

Automatic Failover requires a bidirectional Cluster Linking link between the two clusters, so mirror topics, ACLs, and consumer group offsets replicate in both directions and either side can be promoted without a resync.

Important

A bidirectional cluster link must have the same link name and the same link ID on both clusters. Create the link on one cluster first. When that command completes successfully, create the link on the other cluster. If you create both sides at about the same time, they can end up with different link IDs, and the pair later moves to the FAILED phase with REMOTE_LINK_NOT_FOUND in its status.

Confluent CLI

1. Create the link on the primary cluster.

   confluent kafka link create dr-link --remote-cluster $SECONDARY_LKC \
   --remote-bootstrap-server $SECONDARY_LKC_BOOTSTRAP \
   --config dr-link.config \
   --cluster $PRIMARY_LKC --environment $PRIMARY_ENV
   

   where:

   • $SECONDARY_LKC, $PRIMARY_LKC, $PRIMARY_ENV are your respective cluster and environment ids.

   • $SECONDARY_LKC_BOOTSTRAP is your secondary cluster bootstrap endpoint.

2. Create the link on the secondary cluster.

   confluent kafka link create dr-link --remote-cluster $PRIMARY_LKC \
   --remote-bootstrap-server $PRIMARY_LKC_BOOTSTRAP \
   --config dr-link.config \
   --cluster $SECONDARY_LKC --environment $SECONDARY_ENV
   

   where:

   • $PRIMARY_LKC, $SECONDARY_LKC, $SECONDARY_ENV are your respective cluster and environment ids.

   • $PRIMARY_LKC_BOOTSTRAP is your primary cluster bootstrap endpoint.

REST API

1. Transform the config file into REST arguments

   CONFIGS_JSON=$(jq -c -Rn '
      [inputs
         | select(length > 0 and (startswith("#") | not))
         | capture("^(?<name>[^=]+)=(?<value>.*)$")]
      ' dr-link.config)
   

2. Create the link on the primary cluster.

   curl -sS "https://$PRIMARY_REST_ENDPOINT/kafka/v3/clusters/$PRIMARY_LKC/links?link_name=dr-link" -X POST \
   -u "$CL_API_KEY:$CL_API_KEY_SECRET" \
   -H "Content-Type: application/json" \
   -d '{
      "remote_cluster_id": "'"$SECONDARY_LKC"'",
      "configs": '"${CONFIGS_JSON%]}"',{"name": "bootstrap.servers", "value": "'"$SECONDARY_LKC_BOOTSTRAP"'"}]
   }' | jq .
   

   where:

   • $PRIMARY_LKC, $SECONDARY_LKC are your respective cluster ids.

   • $PRIMARY_REST_ENDPOINT is your primary cluster REST endpoint (without the URL protocol https://).

   • $CL_API_KEY is your cluster link API key ID.

   • $CL_API_KEY_SECRET is your cluster link API key secret.

   • $SECONDARY_LKC_BOOTSTRAP is the bootstrap of your secondary cluster.

3. Create the link on the secondary cluster.

   curl -sS "https://$SECONDARY_REST_ENDPOINT/kafka/v3/clusters/$SECONDARY_LKC/links?link_name=dr-link" -X POST \
   -u "$CL_API_KEY:$CL_API_KEY_SECRET" \
   -H "Content-Type: application/json" \
   -d '{
      "remote_cluster_id": "'"$PRIMARY_LKC"'",
      "configs": '"${CONFIGS_JSON%]}"',{"name": "bootstrap.servers", "value": "'"$PRIMARY_LKC_BOOTSTRAP"'"}]
   }' | jq .
   

   where:

   • $SECONDARY_LKC, $PRIMARY_LKC are your respective cluster ids.

   • $SECONDARY_REST_ENDPOINT is your secondary cluster REST endpoint (without the URL protocol https://).

   • $CL_API_KEY is your cluster link API key id.

   • $CL_API_KEY_SECRET is your cluster link API key secret.

   • $PRIMARY_LKC_BOOTSTRAP is the bootstrap of your primary cluster.

Terraform

Create the link on both clusters using the confluent_cluster_link resource: one resource per direction, sharing the same config.

Terraform creates independent resources in parallel. Make secondary_to_primary depend on primary_to_secondary so the two links are created one after the other and get the same link ID. If you add other dependencies to secondary_to_primary, such as a delay after the role bindings, keep confluent_cluster_link.primary_to_secondary in the depends_on list.

locals {
  dr_link_config = {
    "auto.create.mirror.topics.enable"  = "true"
    "auto.create.mirror.topics.filters" = jsonencode({ topicFilters = [{ name = "*", patternType = "LITERAL", filterType = "INCLUDE" }] })
    "topic.config.sync.ms"              = "1000"
    "consumer.offset.sync.enable"       = "true"
    "consumer.offset.group.filters"     = jsonencode({ groupFilters = [{ name = "*", patternType = "LITERAL", filterType = "INCLUDE", topicTypes = ["LOCAL_MIRROR"] }] })
    "consumer.offset.sync.ms"           = "1000"
    "acl.sync.enable"                   = "true"
    "acl.filters"                        = jsonencode({ aclFilters = [{ resourceFilter = { resourceType = "any", patternType = "any" }, accessFilter = { operation = "any", permissionType = "any" } }] })
    "acl.sync.ms"                        = "1000"
  }
}

resource "confluent_cluster_link" "primary_to_secondary" {
  link_name = "dr-link"
  link_mode = "BIDIRECTIONAL"

  local_kafka_cluster {
    id            = confluent_kafka_cluster.primary.id
    rest_endpoint = confluent_kafka_cluster.primary.rest_endpoint
    credentials {
      key    = confluent_api_key.dr_link_global_key.id
      secret = confluent_api_key.dr_link_global_key.secret
    }
  }

  remote_kafka_cluster {
    id                 = confluent_kafka_cluster.secondary.id
    bootstrap_endpoint = confluent_kafka_cluster.secondary.bootstrap_endpoint
    credentials {
      key    = confluent_api_key.dr_link_global_key.id
      secret = confluent_api_key.dr_link_global_key.secret
    }
  }

  config = local.dr_link_config

  depends_on = [
    confluent_role_binding.dr_link_sa_primary,
    confluent_role_binding.dr_link_sa_secondary,
  ]
}

resource "confluent_cluster_link" "secondary_to_primary" {
  link_name = "dr-link"
  link_mode = "BIDIRECTIONAL"

  local_kafka_cluster {
    id            = confluent_kafka_cluster.secondary.id
    rest_endpoint = confluent_kafka_cluster.secondary.rest_endpoint
    credentials {
      key    = confluent_api_key.dr_link_global_key.id
      secret = confluent_api_key.dr_link_global_key.secret
    }
  }

  remote_kafka_cluster {
    id                 = confluent_kafka_cluster.primary.id
    bootstrap_endpoint = confluent_kafka_cluster.primary.bootstrap_endpoint
    credentials {
      key    = confluent_api_key.dr_link_global_key.id
      secret = confluent_api_key.dr_link_global_key.secret
    }
  }

  config = local.dr_link_config

  depends_on = [
    confluent_cluster_link.primary_to_secondary,
    confluent_role_binding.dr_link_sa_primary,
    confluent_role_binding.dr_link_sa_secondary,
  ]
}

Apply the Terraform configuration to generate a plan, and run it.

terraform apply

where: confluent_api_key.dr_link_global_key is the Global API key resource created in the previous step (see Grant the cluster link identity the CloudClusterAdmin role on both clusters).

Verify the cluster link

Before you create the pair, confirm that the bidirectional cluster link is ACTIVE on both clusters and has the same link ID on both. If the IDs differ, delete the link on both clusters and create it again, one cluster at a time.

Tip

If auto-create has already mirrored existing topics over the link, the delete fails with in use by topics. Promote any mirror topics first with confluent kafka mirror promote, then delete the link.

Confluent CLI

1. List the cluster links on the primary cluster. The ID column shows the link ID, and State should be ACTIVE.

   confluent kafka link list --cluster $PRIMARY_LKC --environment $PRIMARY_ENV
   

2. List the cluster links on the secondary cluster, and confirm that the ID is the same as on the primary cluster.

   confluent kafka link list --cluster $SECONDARY_LKC --environment $SECONDARY_ENV
   

REST API

1. List the details and status of the cluster link on the primary cluster.

   curl -sS "https://$PRIMARY_REST_ENDPOINT/kafka/v3/clusters/$PRIMARY_LKC/links/dr-link" \
   -u "$CL_API_KEY:$CL_API_KEY_SECRET" | jq .
   

   where:

   • $PRIMARY_REST_ENDPOINT is your primary cluster REST endpoint (without the URL protocol https://).

   • $PRIMARY_LKC is your primary cluster ID.

   • $CL_API_KEY is your cluster link API key ID.

   • $CL_API_KEY_SECRET is your cluster link API key secret.

2. List the details and status of the cluster link on the secondary cluster.

   curl -sS "https://$SECONDARY_REST_ENDPOINT/kafka/v3/clusters/$SECONDARY_LKC/links/dr-link" \
   -u "$CL_API_KEY:$CL_API_KEY_SECRET" | jq .
   

   where:

   • $SECONDARY_REST_ENDPOINT is your secondary cluster REST endpoint (without the URL protocol https://).

   • $SECONDARY_LKC is your secondary cluster ID.

   • $CL_API_KEY is your cluster link API key ID.

   • $CL_API_KEY_SECRET is your cluster link API key secret.

The response includes the link ID twice, as link_id (a UUID) and cluster_link_id (a shorter form of the same ID). Compare either field, but use the same one on both clusters.

Terraform

Look up the link’s status on both clusters with the confluent_cluster_link data source.

The cluster_link_id output shows the link ID, so you can confirm that the IDs match on both clusters.

data "confluent_cluster_link" "primary" {
  link_name     = "dr-link"
  rest_endpoint = confluent_kafka_cluster.primary.rest_endpoint
  kafka_cluster {
    id = confluent_kafka_cluster.primary.id
  }
  credentials {
    key    = confluent_api_key.dr_link_global_key.id
    secret = confluent_api_key.dr_link_global_key.secret
  }
}

data "confluent_cluster_link" "secondary" {
  link_name     = "dr-link"
  rest_endpoint = confluent_kafka_cluster.secondary.rest_endpoint
  kafka_cluster {
    id = confluent_kafka_cluster.secondary.id
  }
  credentials {
    key    = confluent_api_key.dr_link_global_key.id
    secret = confluent_api_key.dr_link_global_key.secret
  }
}

output "cluster_link_state" {
  value = {
    primary   = data.confluent_cluster_link.primary.link_state
    secondary = data.confluent_cluster_link.secondary.link_state
  }
}

output "cluster_link_id" {
  value = {
    primary   = data.confluent_cluster_link.primary.cluster_link_id
    secondary = data.confluent_cluster_link.secondary.cluster_link_id
  }
}

Apply the Terraform configuration to generate a plan, and run it.

terraform apply

where: confluent_api_key.dr_link_global_key is the Global API key resource created in Grant the cluster link identity the CloudClusterAdmin role on both clusters.

Create the cluster pair

Pair the clusters and retrieve the global bootstrap endpoint URL.

Create the cluster pair

Before you pair the clusters, verify that:

• The bidirectional cluster link is ACTIVE.

• The secondary cluster has no active producers or consumers connected. There should be no consumer groups active in the secondary cluster, otherwise mirror topics might enter the PENDING_STOPPED state.

Create the switchover pair, referencing both cluster resource names:

Confluent CLI

Create the switchover pair named my-cluster-pair with the primary and secondary clusters as members, and set the primary cluster as the active member.

confluent switchover pair create my-cluster-pair \
   --member name=primary,crn=$PRIMARY_CLUSTER_CRN \
   --member name=secondary,crn=$SECONDARY_CLUSTER_CRN \
   --active-member primary \
   --environment $PAIR_ENV

where:

• $ORG is your organization ID.

• $PRIMARY_ENV, $PRIMARY_LKC, $SECONDARY_ENV, $SECONDARY_LKC are your respective environment and cluster ids.

• $PRIMARY_CLUSTER_CRN is your primary cluster’s Confluent Resource Name (CRN). You can construct using the following format: crn://confluent.cloud/organization=$ORG/environment=$PRIMARY_ENV/cloud-cluster=$PRIMARY_LKC.

• $SECONDARY_CLUSTER_CRN is your secondary cluster’s CRN, constructed the same way: crn://confluent.cloud/organization=$ORG/environment=$SECONDARY_ENV/cloud-cluster=$SECONDARY_LKC.

• $PAIR_ENV is the environment that hosts the switchover pair itself, not necessarily the environment of either member cluster. You can set it to $PRIMARY_ENV as an example.

REST API

Create the switchover pair named my-cluster-pair with the primary and secondary clusters as members, and set the primary cluster as the active member.

curl -sS "https://api.confluent.cloud/switchover/v1/switchover-pairs" -X POST \
   -u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
   -d '{
      "spec": {
      "display_name": "my-cluster-pair",
      "environment_crn": "crn://confluent.cloud/organization='"$ORG"'/environment='"$PAIR_ENV"'",
      "members": [
         {"name": "primary", "member_crn": "'"$PRIMARY_CLUSTER_CRN"'"},
         {"name": "secondary", "member_crn": "'"$SECONDARY_CLUSTER_CRN"'"}
      ],
      "active_member": "primary"
      }
   }' | jq .

where:

• $CLOUD_API_KEY is your Cloud API key ID.

• $CLOUD_API_SECRET is your Cloud API key secret.

• $ORG is your organization ID.

• $PRIMARY_ENV, $PRIMARY_LKC, $SECONDARY_ENV, $SECONDARY_LKC are your respective environment and cluster ids.

• $PRIMARY_CLUSTER_CRN is your primary cluster’s CRN. You can construct using the following format: crn://confluent.cloud/organization=$ORG/environment=$PRIMARY_ENV/cloud-cluster=$PRIMARY_LKC.

• $SECONDARY_CLUSTER_CRN is your secondary cluster’s CRN, constructed the same way: crn://confluent.cloud/organization=$ORG/environment=$SECONDARY_ENV/cloud-cluster=$SECONDARY_LKC.

• $PAIR_ENV is the id of the environment that hosts the switchover pair itself, not necessarily the environment of either member cluster. You can set it to $PRIMARY_ENV as an example.

Terraform

Create the switchover pair named my-cluster-pair with the primary and secondary clusters as members, and set the primary cluster as the active member.

resource "confluent_switchover_pair" "my_cluster_pair" {
  display_name  = "my-cluster-pair"
  active_member = "primary"

  members {
    name       = "primary"
    member_crn = confluent_kafka_cluster.primary.rbac_crn
  }
  members {
    name       = "secondary"
    member_crn = confluent_kafka_cluster.secondary.rbac_crn
  }

  environment_crn = "crn://confluent.cloud/organization=${var.org}/environment=${var.pair_env}"

  depends_on = [
    confluent_cluster_link.primary_to_secondary,
    confluent_cluster_link.secondary_to_primary,
  ]
}

Apply the Terraform configuration to generate a plan, and run it.

terraform apply

where: confluent_kafka_cluster.primary/confluent_kafka_cluster.secondary are the cluster resources created earlier, and var.org/var.pair_env map to $ORG/$PAIR_ENV.

display_name is the only attribute updatable in place: changing members or environment_crn after create is rejected by the provider (destroy and recreate, or apply -replace=confluent_switchover_pair.my_cluster_pair, to change one). apply blocks until the pair reaches READY_TO_FAILOVER.

Verify the cluster pair

Check the status of the pair, including async validation and polling for READY_TO_FAILOVER.

Note

If the command returns a 403 RBAC error, the principal lacks permission to read switchover pairs. Grant it the EnvironmentOperator role. For details, see Prerequisites and considerations.

Confluent CLI

Get the status of the cluster pair.

confluent switchover pair describe $SW \
   --environment $PAIR_ENV

where:

• $SW is your switchover pair ID.

• $PAIR_ENV is the ID of the environment that hosts the switchover pair.

REST API

Get the status of the cluster pair.

curl -sS "https://api.confluent.cloud/switchover/v1/switchover-pairs/$SW?environment=$PAIR_ENV" \
-u "$CLOUD_API_KEY:$CLOUD_API_SECRET" | jq .

where:

• $CLOUD_API_KEY is your Cloud API key ID.

• $CLOUD_API_SECRET is your Cloud API key secret.

• $SW is your switchover pair id.

• $PAIR_ENV is the ID of the environment that hosts the switchover pair.

Terraform

apply above already blocks until the pair reaches READY_TO_FAILOVER. This data source lets you reference the pair’s state elsewhere in this same configuration, such as in another resource or output:

data "confluent_switchover_pair" "my_cluster_pair" {
  id              = confluent_switchover_pair.my_cluster_pair.id
  environment_crn = "crn://confluent.cloud/organization=${var.org}/environment=${var.pair_env}"
}

output "pair_phase" {
  value = data.confluent_switchover_pair.my_cluster_pair.phase
}

Apply the Terraform configuration to generate a plan, and run it.

terraform apply

where: id is your switchover pair ID ($SW) and environment_crn is the environment that hosts the switchover pair ($PAIR_ENV). A data source has nothing to change: apply simply refreshes and shows the output.

If the pair’s phase becomes FAILED, read status.conditions in the CLI output (-o json) or the REST response for the reason. For example, run confluent switchover pair describe $SW --environment $PAIR_ENV -o json. The Terraform phase output shows FAILED but not the reason, so use the CLI or REST API to read status.conditions.

Create an endpoint pair

For each primary cluster endpoint, you can have a corresponding endpoint on your secondary cluster and pair these endpoints together. Each of these endpoint pairs provides a global endpoint that your clients can connect to. During a disaster, you trigger failover on the cluster pair, and all its child endpoint pairs follow.

Public and Private Link / Private Service Connect endpoints are supported. Endpoint pairs must be of matching networking type. For example, you can only pair a Private Link endpoint with another Private link endpoint.

In this tutorial, you pair together a set of public endpoints: one from the primary and one from the secondary cluster.

Create an endpoint pair bound to your cluster pair:

Confluent CLI

Create the switchover endpoint for the pair.

confluent switchover endpoint create my-cluster-pair-endpoint \
   --parent-resource-crn crn://confluent.cloud/organization=$ORG/environment=$PAIR_ENV/switchover-pair=$SW \
   --endpoint name=primary-ep,type=public \
   --endpoint name=secondary-ep,type=public

where:

• $ORG is your organization ID.

• $PAIR_ENV is the ID of the environment that hosts the switchover pair.

• $SW is the switchover pair ID.

REST API

Create the switchover endpoint for the pair.

curl -sS "https://api.confluent.cloud/switchover/v1/switchover-endpoints" -X POST \
-u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
-d '{
   "spec": {
      "display_name": "my-cluster-pair-endpoint",
      "parent_resource_crn": "crn://confluent.cloud/organization='"$ORG"'/environment='"$PAIR_ENV"'/switchover-pair='"$SW"'",
      "endpoints": [
      {"name": "primary-ep", "endpoint_filter": {"type": "public"}},
      {"name": "secondary-ep", "endpoint_filter": {"type": "public"}}
      ]
   }
}' | jq .

where:

• $CLOUD_API_KEY is your Cloud API key ID.

• $CLOUD_API_SECRET is your Cloud API key secret.

• $ORG is your organization ID.

• $PAIR_ENV is the ID of the environment that hosts the switchover pair.

• $SW is the switchover pair ID.

Terraform

Create the switchover endpoint for the pair. parent_resource_crn references the pair resource directly. target is server-owned (it follows the active member), so there’s no input for it.

resource "confluent_switchover_endpoint" "my_cluster_pair_endpoint" {
  display_name        = "my-cluster-pair-endpoint"
  parent_resource_crn = "${confluent_switchover_pair.my_cluster_pair.environment_crn}/switchover-pair=${confluent_switchover_pair.my_cluster_pair.id}"

  endpoints {
    name = "primary-ep"
    endpoint_filter {
      type = "public"
    }
  }
  endpoints {
    name = "secondary-ep"
    endpoint_filter {
      type = "public"
    }
  }
}

Apply the Terraform configuration to generate a plan, and run it.

terraform apply

Same rule as the pair resource: display_name is the only attribute updatable in place. Changing endpoints or parent_resource_crn after create is rejected by the provider (destroy and recreate, or apply -replace=confluent_switchover_endpoint.my_cluster_pair_endpoint, to change one).

Verify the endpoint pair

Check the status of the endpoint pair, including async validation and polling for READY. Reading switchover endpoints requires the same role as reading switchover pairs. For role details, see Prerequisites and considerations.

Confluent CLI

Get the status of the endpoint pair.

confluent switchover endpoint describe $SE \
   --environment $PAIR_ENV

where:

• $SE is your switchover endpoint ID.

• $PAIR_ENV is the ID of the environment that hosts the switchover pair.

REST API

Get the status of the endpoint pair.

curl -sS "https://api.confluent.cloud/switchover/v1/switchover-endpoints/$SE?environment=$PAIR_ENV" \
-u "$CLOUD_API_KEY:$CLOUD_API_SECRET" | jq .

where:

• $CLOUD_API_KEY is your Cloud API key ID.

• $CLOUD_API_SECRET is your Cloud API key secret.

• $SE is your switchover endpoint ID.

• $PAIR_ENV is the id of the environment that hosts the switchover pair.

Terraform

data "confluent_switchover_endpoint" "my_cluster_pair_endpoint" {
  id              = confluent_switchover_endpoint.my_cluster_pair_endpoint.id
  environment_crn = "crn://confluent.cloud/organization=${var.org}/environment=${var.pair_env}"
}

output "endpoint_target" {
  value = data.confluent_switchover_endpoint.my_cluster_pair_endpoint.target
}

Apply the Terraform configuration to generate a plan, and run it.

terraform apply

where: id is your switchover endpoint ID ($SE) and environment_crn is the environment that hosts the switchover pair ($PAIR_ENV).

Construct the global bootstrap endpoint URL

No Confluent CLI command, REST API response, or Terraform resource returns the global bootstrap URL directly. Using the switchover pair ID ($SW) and switchover endpoint ID ($SE) retrieved previously, construct it yourself using the following format:

glkc-$SW-$SE.global.glb.confluent.cloud:9092

where

• $SW is formatted as sw-<12 chars> (such as, sw-a1111bc23d45)

• $SE is formatted as se-<12 chars> (such as, se-12a3456b78c9)

For example:

glkc-sw-a1111bc23d45-se-12a3456b78c9.global.glb.confluent.cloud:9092

Keep this global bootstrap as you’ll need to give it to your clients for them to failover properly.

Produce and consume sample data

Start by creating sample produce and consume applications in the language of your choice. Use either API keys or OAuth by following one of these tutorials: Kafka Client Examples for Confluent Cloud. By default, these applications connect to a single cluster that is not set up for DR. Point them towards your primary cluster for now.

This tutorial uses the Java produce/consume sample as an example.

The following steps show how to set up your clients for DR and test the setup using sample data.

Create the topic on the primary cluster

Create the purchases topic on the primary cluster. The cluster link automatically creates a mirror topic for it on the secondary cluster, typically within about five minutes, regardless of whether you created the topic before or after the cluster link.

If you already created the purchases topic on the primary cluster while following the sample client tutorial, you don’t need to repeat this step.

Confluent CLI

confluent kafka topic create purchases --partitions 6 \
--cluster $PRIMARY_LKC --environment $PRIMARY_ENV

where: $PRIMARY_LKC and $PRIMARY_ENV are your primary cluster and environment IDs.

REST API

curl -sS "https://$PRIMARY_REST_ENDPOINT/kafka/v3/clusters/$PRIMARY_LKC/topics" -X POST \
-u "$CL_API_KEY:$CL_API_KEY_SECRET" -H "Content-Type: application/json" \
-d '{
   "topic_name": "purchases",
   "partitions_count": 6
}' | jq .

where:

• $PRIMARY_REST_ENDPOINT is your primary cluster REST endpoint (without the URL protocol https://).

• $PRIMARY_LKC is your primary cluster ID.

• $CL_API_KEY is your cluster link API key ID.

• $CL_API_KEY_SECRET is your cluster link API key secret.

Terraform

Create the topic on the primary cluster with the confluent_kafka_topic resource.

resource "confluent_kafka_topic" "purchases_primary" {
  kafka_cluster {
    id = confluent_kafka_cluster.primary.id
  }

  topic_name       = "purchases"
  partitions_count = 6
  rest_endpoint    = confluent_kafka_cluster.primary.rest_endpoint

  credentials {
    key    = confluent_api_key.dr_link_global_key.id
    secret = confluent_api_key.dr_link_global_key.secret
  }

  depends_on = [
    confluent_cluster_link.primary_to_secondary,
    confluent_cluster_link.secondary_to_primary,
  ]
}

Apply the Terraform configuration to generate a plan, and run it.

terraform apply

Confirm the mirror topic exists on the secondary cluster

Confirm that the cluster link automatically created a mirror topic for the purchases topic you created in Create the topic on the primary cluster. Auto-create mirror topics runs about every five minutes, so the mirror topic can take up to five minutes to appear after you create the topic:

Confluent CLI

List topics on the secondary cluster to verify that the mirror topic for purchases exists.

confluent kafka topic list --cluster $SECONDARY_LKC --environment $SECONDARY_ENV

REST API

List topics on the secondary cluster to verify that the mirror topic for purchases exists.

curl -sS "https://$SECONDARY_REST_ENDPOINT/kafka/v3/clusters/$SECONDARY_LKC/topics" \
-u "$CL_API_KEY:$CL_API_KEY_SECRET" | jq .

Terraform

Look up the mirror topic on the secondary cluster with the confluent_kafka_topic data source.

data "confluent_kafka_topic" "purchases_secondary" {
  kafka_cluster {
    id = confluent_kafka_cluster.secondary.id
  }

  topic_name    = "purchases"
  rest_endpoint = confluent_kafka_cluster.secondary.rest_endpoint

  credentials {
    key    = confluent_api_key.dr_link_global_key.id
    secret = confluent_api_key.dr_link_global_key.secret
  }
}

output "purchases_mirror_topic_config" {
  value = data.confluent_kafka_topic.purchases_secondary.config
}

Apply the Terraform configuration to generate a plan, and run it.

terraform apply

A lookup that succeeds confirms the mirror topic exists. terraform apply fails if it doesn’t (yet).

Update client authentication configuration

For failover to work, clients (both producers and consumers) must be able to authenticate against your primary and secondary clusters. Update their authentication configuration whether they use API key or OAuth.

API KEY

When using API keys, clients must use Global API keys. Follow this Global API keys tutorial to create new keys and update your authentication config:

security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required username='$CLIENT_API_KEY' password='$CLIENT_API_KEY_SECRET';

where:

• $CLIENT_API_KEY is your client’s Global API key id.

• $CLIENT_API_KEY_SECRET is your client’s Global API key secret.

OAUTH

When using OAuth, clients must remove extension_logicalCluster. Follow this Identity Pool tutorial to create identity pools if needed and verify your authentication config:

security.protocol=SASL_SSL
sasl.mechanism=OAUTHBEARER
sasl.login.callback.handler.class=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginCallbackHandler
sasl.oauthbearer.token.endpoint.url=$OAUTH_TOKEN_ENDPOINT
sasl.jaas.config=org.apache.kafka.common.security.oauthbearer.OAuthBearerLoginModule required clientId='$OAUTH_CLIENT_ID' clientSecret='$OAUTH_CLIENT_SECRET' scope='' extension_identityPoolId='$OAUTH_IDENTITY_POOL_ID';

where:

• $OAUTH_TOKEN_ENDPOINT is your OAuth token endpoint.

• $OAUTH_CLIENT_ID is your OAuth client id.

• $OAUTH_CLIENT_SECRET is your OAuth client secret.

• $OAUTH_IDENTITY_POOL_ID is your OAuth Identity Pool Id.

Grant permissions to your clients on both clusters

For failover to work, clients (both producers and consumers) must be authorized on your primary and secondary clusters.

If using ACLs, you don’t need to do anything else since they are automatically replicated by cluster link.

If using RBAC, make sure to set up role bindings on your primary and secondary clusters. Grant DeveloperWrite to produce and DeveloperRead to consume on topics, plus DeveloperRead on consumer groups, on both clusters. These bindings are identical whether you use API keys or OAuth. Only the principal differs: a service account for API keys, or an identity pool for OAuth.

Confluent CLI

1. Grant write access to the topic, to produce, on both clusters.

   confluent iam rbac role-binding create --principal User:$CLIENT_PRINCIPAL \
   --role DeveloperWrite --resource Topic:purchases \
   --environment $PRIMARY_ENV --cloud-cluster $PRIMARY_LKC --kafka-cluster $PRIMARY_LKC
   

   confluent iam rbac role-binding create --principal User:$CLIENT_PRINCIPAL \
   --role DeveloperWrite --resource Topic:purchases \
   --environment $SECONDARY_ENV --cloud-cluster $SECONDARY_LKC --kafka-cluster $SECONDARY_LKC
   

2. Grant read access to the topic, to consume, on both clusters.

   confluent iam rbac role-binding create --principal User:$CLIENT_PRINCIPAL \
   --role DeveloperRead --resource Topic:purchases \
   --environment $PRIMARY_ENV --cloud-cluster $PRIMARY_LKC --kafka-cluster $PRIMARY_LKC
   

   confluent iam rbac role-binding create --principal User:$CLIENT_PRINCIPAL \
   --role DeveloperRead --resource Topic:purchases \
   --environment $SECONDARY_ENV --cloud-cluster $SECONDARY_LKC --kafka-cluster $SECONDARY_LKC
   

3. Grant read access to its consumer group, to consume, on both clusters.

   confluent iam rbac role-binding create --principal User:$CLIENT_PRINCIPAL \
   --role DeveloperRead --resource Group:dr-demo-consumer \
   --environment $PRIMARY_ENV --cloud-cluster $PRIMARY_LKC --kafka-cluster $PRIMARY_LKC
   

   confluent iam rbac role-binding create --principal User:$CLIENT_PRINCIPAL \
   --role DeveloperRead --resource Group:dr-demo-consumer \
   --environment $SECONDARY_ENV --cloud-cluster $SECONDARY_LKC --kafka-cluster $SECONDARY_LKC
   

   where:

   • $CLIENT_PRINCIPAL is your client’s service account id (API keys) or identity pool id (OAuth).

   • $PRIMARY_ENV, $PRIMARY_LKC, $SECONDARY_ENV, $SECONDARY_LKC are your respective environment and cluster ids.

REST API

1. Grant write access to the topic, to produce, on both clusters.

   curl -sS "https://api.confluent.cloud/iam/v2/role-bindings" -X POST \
   -u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
   -d '{
      "principal": "User:'"$CLIENT_PRINCIPAL"'",
      "role_name": "DeveloperWrite",
      "crn_pattern": "crn://confluent.cloud/organization='"$ORG"'/environment='"$PRIMARY_ENV"'/cloud-cluster='"$PRIMARY_LKC"'/kafka='"$PRIMARY_LKC"'/topic=purchases"
   }' | jq .
   

   curl -sS "https://api.confluent.cloud/iam/v2/role-bindings" -X POST \
   -u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
   -d '{
      "principal": "User:'"$CLIENT_PRINCIPAL"'",
      "role_name": "DeveloperWrite",
      "crn_pattern": "crn://confluent.cloud/organization='"$ORG"'/environment='"$SECONDARY_ENV"'/cloud-cluster='"$SECONDARY_LKC"'/kafka='"$SECONDARY_LKC"'/topic=purchases"
   }' | jq .
   

2. Grant read access to the topic, to consume, on both clusters.

   curl -sS "https://api.confluent.cloud/iam/v2/role-bindings" -X POST \
   -u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
   -d '{
      "principal": "User:'"$CLIENT_PRINCIPAL"'",
      "role_name": "DeveloperRead",
      "crn_pattern": "crn://confluent.cloud/organization='"$ORG"'/environment='"$PRIMARY_ENV"'/cloud-cluster='"$PRIMARY_LKC"'/kafka='"$PRIMARY_LKC"'/topic=purchases"
   }' | jq .
   

   curl -sS "https://api.confluent.cloud/iam/v2/role-bindings" -X POST \
   -u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
   -d '{
      "principal": "User:'"$CLIENT_PRINCIPAL"'",
      "role_name": "DeveloperRead",
      "crn_pattern": "crn://confluent.cloud/organization='"$ORG"'/environment='"$SECONDARY_ENV"'/cloud-cluster='"$SECONDARY_LKC"'/kafka='"$SECONDARY_LKC"'/topic=purchases"
   }' | jq .
   

3. Grant read access to its consumer group, to consume, on both clusters.

   curl -sS "https://api.confluent.cloud/iam/v2/role-bindings" -X POST \
   -u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
   -d '{
      "principal": "User:'"$CLIENT_PRINCIPAL"'",
      "role_name": "DeveloperRead",
      "crn_pattern": "crn://confluent.cloud/organization='"$ORG"'/environment='"$PRIMARY_ENV"'/cloud-cluster='"$PRIMARY_LKC"'/kafka='"$PRIMARY_LKC"'/group=dr-demo-consumer"
   }' | jq .
   

   curl -sS "https://api.confluent.cloud/iam/v2/role-bindings" -X POST \
   -u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
   -d '{
      "principal": "User:'"$CLIENT_PRINCIPAL"'",
      "role_name": "DeveloperRead",
      "crn_pattern": "crn://confluent.cloud/organization='"$ORG"'/environment='"$SECONDARY_ENV"'/cloud-cluster='"$SECONDARY_LKC"'/kafka='"$SECONDARY_LKC"'/group=dr-demo-consumer"
   }' | jq .
   

   where:

   • $CLIENT_PRINCIPAL is your client’s service account id (API keys) or identity pool id (OAuth).

   • $CLOUD_API_KEY / $CLOUD_API_SECRET are your Cloud API key ID/secret.

   • $ORG is your organization id.

   • $PRIMARY_ENV, $PRIMARY_LKC, $SECONDARY_ENV, $SECONDARY_LKC are your respective environment and cluster ids.

Terraform

1. Grant write access to the topic, to produce, on both clusters.

   resource "confluent_role_binding" "client-primary-write" {
   principal   = "User:${var.client_principal_id}"
   role_name   = "DeveloperWrite"
   crn_pattern = "${confluent_kafka_cluster.primary.rbac_crn}/kafka=${confluent_kafka_cluster.primary.id}/topic=purchases"
   }
   
   resource "confluent_role_binding" "client-secondary-write" {
   principal   = "User:${var.client_principal_id}"
   role_name   = "DeveloperWrite"
   crn_pattern = "${confluent_kafka_cluster.secondary.rbac_crn}/kafka=${confluent_kafka_cluster.secondary.id}/topic=purchases"
   }
   

2. Grant read access to the topic, to consume, on both clusters.

   resource "confluent_role_binding" "client-primary-read-topic" {
   principal   = "User:${var.client_principal_id}"
   role_name   = "DeveloperRead"
   crn_pattern = "${confluent_kafka_cluster.primary.rbac_crn}/kafka=${confluent_kafka_cluster.primary.id}/topic=purchases"
   }
   
   resource "confluent_role_binding" "client-secondary-read-topic" {
   principal   = "User:${var.client_principal_id}"
   role_name   = "DeveloperRead"
   crn_pattern = "${confluent_kafka_cluster.secondary.rbac_crn}/kafka=${confluent_kafka_cluster.secondary.id}/topic=purchases"
   }
   

3. Grant read access to its consumer group, to consume, on both clusters.

   resource "confluent_role_binding" "client-primary-read-group" {
   principal   = "User:${var.client_principal_id}"
   role_name   = "DeveloperRead"
   crn_pattern = "${confluent_kafka_cluster.primary.rbac_crn}/kafka=${confluent_kafka_cluster.primary.id}/group=dr-demo-consumer"
   }
   
   resource "confluent_role_binding" "client-secondary-read-group" {
   principal   = "User:${var.client_principal_id}"
   role_name   = "DeveloperRead"
   crn_pattern = "${confluent_kafka_cluster.secondary.rbac_crn}/kafka=${confluent_kafka_cluster.secondary.id}/group=dr-demo-consumer"
   }
   

   where:

   • var.client_principal_id is your client’s service account id (API keys) or identity pool id (OAuth).

   • confluent_kafka_cluster.primary/confluent_kafka_cluster.secondary are the cluster resources created earlier.

Apply the Terraform configuration to generate a plan, and run it.

terraform apply

Configure your consumer client with the same group.id that you granted access to above (dr-demo-consumer). The default group.id in the sample client tutorials isn’t authorized on either cluster.

Update client bootstrap configuration

Similarly, for failover to work, your clients (both producers and consumers) must point at the currently active cluster. You can do this by using the global bootstrap and clients internal logic available in versions Kafka 4.0 or later, including the metadata.recovery.strategy setting, which defaults to rebootstrap and makes the client go back to the global bootstrap URL to rediscover the active cluster when its cached cluster metadata becomes stale after a failover. This tutorial sets it explicitly so the behavior doesn’t depend on a client’s default.

bootstrap.servers=$GLOBAL_BOOTSTRAP_URL
metadata.recovery.strategy=rebootstrap

where:

• $GLOBAL_BOOTSTRAP_URL is the global bootstrap URL for your switchover pair, constructed earlier: glkc-$SW-$SE.global.glb.confluent.cloud:9092.

Recovery time differs by client type and failover situation. In a planned failover, when the primary cluster is still available, clients can receive a command to initiate a rebootstrap immediately. In an unplanned failover, when the primary cluster is not available, the clients must detect that a reboostrap is required. A consumer in a consumer group typically detects the change within about a minute, through its normal group-rebalance heartbeat. A producer has no equivalent mechanism and instead waits for metadata.recovery.rebootstrap.trigger.ms (default 300000 ms, 5 minutes) to elapse before it re-bootstraps. To recover producers faster, lower this value:

metadata.recovery.rebootstrap.trigger.ms=30000

Lowering this value makes the client rebootstrap sooner whenever it can’t get metadata from any known broker, so a brief network interruption can also trigger an unnecessary rebootstrap. Choose a value that balances faster failover against that risk.

Apply your client configuration

To finish setting up your clients, restart them so that they pick up the changes.

For this tutorial, start your producer and consumer in separate terminals so you can watch messages during a failover. The exact command depends on which sample client tutorial you followed. For example, for the Java sample client:

1. In one terminal, start the producer:

   java -cp build/libs/kafka-java-getting-started-0.0.1.jar io.confluent.developer.ProducerExample
   

2. In another terminal, start the consumer:

   java -cp build/libs/kafka-java-getting-started-0.0.1.jar io.confluent.developer.ConsumerExample
   

Your Automatic Failover setup is now complete. On failover, clients automatically follow the active cluster without restarts or further configuration changes, though how quickly they do so differs by client type and failover type, as described in Update client bootstrap configuration.

Monitor mirroring lag

Because Cluster Linking is an asynchronous process, there might be mirroring lag between the source cluster and the destination cluster.

You can see what your mirroring lag is on a per-partition basis for your DR (mirror) topic:

Confluent CLI

Check the mirroring lag for your DR topic.

confluent kafka mirror describe purchases --link dr-link --cluster $SECONDARY_LKC --environment $SECONDARY_ENV

REST API

Check the mirroring lag for your DR topic.

curl -sS "https://$SECONDARY_REST_ENDPOINT/kafka/v3/clusters/$SECONDARY_LKC/links/dr-link/mirrors/purchases" \
-u "$CL_API_KEY:$CL_API_KEY_SECRET" | jq .

where:

• $SECONDARY_REST_ENDPOINT is your secondary cluster REST endpoint (without the URL protocol https://).

• $SECONDARY_LKC is your secondary cluster id.

• $CL_API_KEY / $CL_API_KEY_SECRET are your cluster link’s Global API key id/secret.

Terraform

N/A

The following example shows the output of the mirroring lag for each partition of your DR topic.

  LinkName  | MirrorTopicName | Partition | PartitionMirrorLag | SourceTopicName | MirrorStatus | StatusTimeMs
+-----------+-----------------+-----------+--------------------+-----------------+--------------+---------------+
   dr-link  | purchases       |         0 |                  0 |  purchases      | ACTIVE       | 1624030963587
   dr-link  | purchases       |         1 |                  0 |  purchases      | ACTIVE       | 1624030963587
   dr-link  | purchases       |         2 |                  0 |  purchases      | ACTIVE       | 1624030963587

You can also monitor mirroring lag and metrics through the Confluent Cloud Metrics. These two metrics are exposed:

• MaxLag shows the maximum lag (in number of messages) among the partitions that are being mirrored. It is available on a per-topic and a per-link basis. This gives you a sense of how much data is on the original cluster only at the point of failover.

• Mirroring Throughput shows on a per-link or per-topic basis how much data is being mirrored.

Fail over (Automatic Failover)

Trigger a failover on the switchover pair to promote the secondary cluster to active. Two failover types are supported:

• PLANNED: use this for a graceful, voluntary failover (such as for scheduled maintenance) where the primary cluster is still reachable and can cooperate in the handoff. Because the primary can finish syncing before the secondary is promoted, this is the lower-risk option that should not involve data loss, and you can fail back directly afterward without a restore step.

• UNPLANNED: use this when the primary cluster is genuinely unavailable (such as in a real outage) and cannot cooperate in the handoff. The secondary is promoted immediately without waiting on the primary, so any data not yet mirrored at the time of the outage is not reflected on the newly active cluster. Because of this, an UNPLANNED failover must be followed by a restore step before failing back to the original primary.

You can start with a PLANNED failover and escalate to an UNPLANNED one if it’s taking too long. But escalating means accepting the data loss and required restore step described for UNPLANNED failover.

Caution

Duplicate messages

Consumers can reprocess duplicate messages after a failover or failback, because consumer offsets sync across the link on an interval (consumer.offset.sync.ms, which is 1000 ms in the recommended configuration and defaults to 30 seconds) rather than instantly.

For a PLANNED failover, or for failback, you control the timing: wait until at least one full sync interval passes after the last committed offset before initiating it, to prevent duplicates.

For an UNPLANNED failover, the primary is already unavailable, so there’s no opportunity to wait first. Expect some duplicate reprocessing in this case . You can prevent it, but your consumer’s own deduplication logic can handle it if that matters for your use case.

Confluent CLI

Trigger a planned failover to promote the secondary cluster to active.

confluent switchover pair failover $SW \
--active-member secondary \
--failover-type PLANNED \
--environment $PAIR_ENV

Trigger an unplanned failover (primary cluster unavailable).

confluent switchover pair failover $SW \
--active-member secondary \
--failover-type UNPLANNED \
--environment $PAIR_ENV

where:

• $SW is your switchover pair ID.

• $PAIR_ENV is the ID of the environment that hosts the switchover pair.

The confluent switchover pair failover command prompts for confirmation: Do you want to proceed? (y/n). In a script or CI pipeline, where the command can’t read input, it fails with Error: failed to read input. To skip the prompt, add the --force flag. This applies to the unplanned failover, restore, and failback commands as well.

REST API

Trigger a planned failover to promote the secondary cluster to active.

curl -sS "https://api.confluent.cloud/switchover/v1/switchover-pairs/${SW}:failover" -X POST \
-u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
-d '{
   "spec": {
      "environment_crn": "crn://confluent.cloud/organization='"$ORG"'/environment='"$PAIR_ENV"'",
      "active_member": "secondary",
      "failover_type": "PLANNED"
   }
}' | jq .

Trigger an unplanned failover (primary cluster unavailable).

curl -sS "https://api.confluent.cloud/switchover/v1/switchover-pairs/${SW}:failover" -X POST \
-u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
-d '{
   "spec": {
      "environment_crn": "crn://confluent.cloud/organization='"$ORG"'/environment='"$PAIR_ENV"'",
      "active_member": "secondary",
      "failover_type": "UNPLANNED"
   }
}' | jq .

where:

• $CLOUD_API_KEY is your Cloud API key ID.

• $CLOUD_API_SECRET is your Cloud API key secret.

• $ORG is your organization ID.

• $SW is your switchover pair ID.

• $PAIR_ENV is the ID of the environment that hosts the switchover pair.

Terraform

Terraform manages the switchover pair and endpoint resources (create, read, update, destroy) but does not support triggering a failover. Use the Confluent CLI or REST API tab instead.

When failover runs, the global bootstrap of the cluster pair resolves client connections to the newly promoted active cluster.

Mirror topics become regular topics on the promoted secondary during failover.

During an outage, make topic configuration changes directly on the active cluster with the API, CLI (confluent kafka topic), or Confluent Cloud Console. While in a failed-over state, mirror topics are not visible through Automatic Failover interfaces.

Restore and fail back

When your original primary cluster is recovered and you are ready to return traffic to it, perform the following steps. The restore step is required only after an UNPLANNED failover. If you performed a PLANNED failover, skip the restore step and go directly to the failback step.

Restore the original primary cluster

This section describes how to restore the original primary cluster after an unplanned failover.

Warning

• Recover any lagged data from the original primary cluster *before* you run the restore step. The restore truncates unreplicated data. In a switchover pair, the original primary cluster is the passive cluster, so consumers can’t read from it until Confluent Support allow-lists them. To recover lagged data, contact Confluent Support to allow-list your consumers to read from the original primary cluster.

• Data loss risk: The restore step truncates unreplicated data on the original primary cluster to align its state with the active (secondary) cluster. This permanently deletes any unreplicated writes made to the original primary cluster before the outage. If you need those records, recover them before you run restore, as described in Recover lagged data (Recover lagged data).

• Data loss risk: Make sure that the restore step does not need to truncate into tiered storage. Otherwise, the topic enters a PENDING_RESTORE_MIRROR state that requires deleting and recreating it.

Truncate and catch up the original primary from the currently active cluster:

Confluent CLI

Trigger a restore operation to catch up the original primary cluster from the currently active cluster.

confluent switchover pair failover $SW \
--failover-type RESTORE \
--environment $PAIR_ENV

where:

• $SW is your switchover pair ID.

• $PAIR_ENV is the ID of the environment that hosts the switchover pair.

REST API

Trigger a restore operation to catch up the original primary cluster from the currently active cluster.

curl -sS "https://api.confluent.cloud/switchover/v1/switchover-pairs/${SW}:failover" -X POST \
-u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
-d '{
   "spec": {
      "environment_crn": "crn://confluent.cloud/organization='"$ORG"'/environment='"$PAIR_ENV"'",
      "failover_type": "RESTORE"
   }
}' | jq .

where:

• $CLOUD_API_KEY is your Cloud API key ID.

• $CLOUD_API_SECRET is your Cloud API key secret.

• $ORG is your organization ID.

• $SW is your switchover pair ID.

• $PAIR_ENV is the ID of the environment that hosts the switchover pair.

Terraform

Terraform does not support triggering a restore: use the Confluent CLI or REST API tab instead.

Fail back to the original primary cluster

When the primary cluster is fully restored and ready to resume its role as the active cluster, you can initiate a planned failback. Before triggering a planned failover, make sure that your cluster link lag is as small as possible to speed up failover.

After failback, the topics on the original primary cluster become regular (writable) topics again, and the topics on the secondary cluster revert to mirror topics.

Caution

Duplicate messages

As with failover, consumers can reprocess duplicate messages after failback. Wait until at least one full consumer.offset.sync.ms interval passes after the last committed offset before initiating failback. For details, see the duplicate messages caution in the Fail over section.

After that sync interval, trigger a planned switchover to return active status to the primary:

Confluent CLI

Trigger a planned switchover to return active status to the primary cluster.

confluent switchover pair failover $SW \
--active-member primary \
--failover-type PLANNED \
--environment $PAIR_ENV

where:

• $SW is your switchover pair ID.

• $PAIR_ENV is the ID of the environment that hosts the switchover pair.

REST API

Trigger a planned switchover to return active status to the primary cluster.

curl -sS "https://api.confluent.cloud/switchover/v1/switchover-pairs/${SW}:failover" -X POST \
-u "$CLOUD_API_KEY:$CLOUD_API_SECRET" -H "Content-Type: application/json" \
-d '{
   "spec": {
      "environment_crn": "crn://confluent.cloud/organization='"$ORG"'/environment='"$PAIR_ENV"'",
      "active_member": "primary",
      "failover_type": "PLANNED"
   }
}' | jq .

where:

• $CLOUD_API_KEY is your Cloud API key ID.

• $CLOUD_API_SECRET is your Cloud API key secret.

• $ORG is your organization ID.

• $SW is your switchover pair ID.

• $PAIR_ENV is the ID of the environment that hosts the switchover pair.

Terraform

Terraform does not support triggering a failback: use the Confluent CLI or REST API tab instead.

Teardown

When you no longer need the Automatic Failover setup from this tutorial, delete the switchover endpoint, then the switchover pair, and then the primary and secondary clusters. Delete them in this order, because the switchover pair references the clusters, and the switchover endpoint references the pair. Deleting the clusters also deletes the topics, mirror topics, and cluster links on them, so you don’t need to delete those separately.

Delete the switchover endpoint and pair

Deleting a switchover pair doesn’t automatically delete its switchover endpoints. Delete all switchover endpoints associated with the pair first, because they reference the pair, and then delete the pair. This tutorial creates one endpoint. Immediately after you delete the endpoint, deleting the pair can fail with an error that the endpoint must be deleted first. If that happens, wait a moment and retry.

Confluent CLI

1. Delete the switchover endpoint.

   confluent switchover endpoint delete $SE --environment $PAIR_ENV --force
   

2. Delete the switchover pair.

   confluent switchover pair delete $SW --environment $PAIR_ENV --force
   

where:

• $SE is your switchover endpoint ID.

• $SW is your switchover pair ID.

• $PAIR_ENV is the ID of the environment that hosts the switchover pair.

REST API

1. Delete the switchover endpoint.

   curl -sS "https://api.confluent.cloud/switchover/v1/switchover-endpoints/$SE?environment=$PAIR_ENV" -X DELETE \
   -u "$CLOUD_API_KEY:$CLOUD_API_SECRET"
   

2. Delete the switchover pair.

   curl -sS "https://api.confluent.cloud/switchover/v1/switchover-pairs/$SW?environment=$PAIR_ENV" -X DELETE \
   -u "$CLOUD_API_KEY:$CLOUD_API_SECRET"
   

where:

• $CLOUD_API_KEY is your Cloud API key ID.

• $CLOUD_API_SECRET is your Cloud API key secret.

• $SE is your switchover endpoint ID.

• $SW is your switchover pair ID.

• $PAIR_ENV is the ID of the environment that hosts the switchover pair.

Terraform

Destroy the switchover endpoint and the switchover pair. The endpoint references the pair, so Terraform destroys the endpoint first.

terraform destroy \
-target=confluent_switchover_endpoint.my_cluster_pair_endpoint \
-target=confluent_switchover_pair.my_cluster_pair

Delete the clusters

Delete the primary and secondary clusters. Deleting a cluster also deletes the topics and cluster links on it, including the purchases topic and its mirror topic, so you don’t need to delete them separately.

Confluent CLI

confluent kafka cluster delete $PRIMARY_LKC --environment $PRIMARY_ENV --force
confluent kafka cluster delete $SECONDARY_LKC --environment $SECONDARY_ENV --force

REST API

curl -sS "https://api.confluent.cloud/cmk/v2/clusters/$PRIMARY_LKC?environment=$PRIMARY_ENV" -X DELETE \
-u "$CLOUD_API_KEY:$CLOUD_API_SECRET"

curl -sS "https://api.confluent.cloud/cmk/v2/clusters/$SECONDARY_LKC?environment=$SECONDARY_ENV" -X DELETE \
-u "$CLOUD_API_KEY:$CLOUD_API_SECRET"

where:

• $CLOUD_API_KEY is your Cloud API key ID.

• $CLOUD_API_SECRET is your Cloud API key secret.

Terraform

The Global API key for the cluster link identity has a lifecycle { prevent_destroy = true } block (see Grant the cluster link identity the CloudClusterAdmin role on both clusters). Remove that block from the confluent_api_key resource before you destroy the setup, or the destroy fails.

Then destroy all remaining resources in the configuration, including the cluster links, the clusters, and the service account, role bindings, API key, and purchases topic you created in this tutorial:

terraform destroy

Related content

• For a comparison of approaches to disaster recovery, see Choose your DR deployment strategy.

• To learn more about mirror topics, see Mirror Topics concepts.

• For metrics and monitoring information, see Cluster Linking Metrics and Lag Monitoring.