<a id="co-schema-registry-switchover"></a>

# Set up Schema Registry Switchover with Unified Stream Manager

The Schema Registry switchover streamlines and automates the management of Schema Registry clusters
within the Confluent ecosystem in order to facilitate automated migration,
switchover, and rollback between different Schema Registry deployments, specifically
between self-managed Confluent Platform clusters and Confluent Cloud clusters. The Confluent Cloud Schema Registry
serves as the authoritative system of record, while the Confluent Platform Schema Registry functions as a
cache for on-prem applications using Schema Linking. This setup allows read
operations to be handled by the on-prem Schema Registry, while all write operations must be
directed to the cloud Schema Registry.

Using Unified Stream Manager, Confluent for Kubernetes (CFK) provides declarative workflows that trigger and monitor the
progress of switching a Confluent Platform Schema Registry from being the primary registry to using
Confluent Cloud Schema Registry as the primary. This enables seamless migration of schema
management from on-premises to cloud, while minimizing manual intervention.

CFK manages the automation workflows to minimize downtime and manual effort
during Schema Registry migration or linking. It ensures issues are surfaced via status
fields, and error states are handled promptly.

This topic describes how to automate the Schema Registry switchover process in CFK.

The high-level steps for configuring Schema Registry switchover from Confluent Platform to Confluent Cloud are:

1. [Configure Confluent Platform Schema Registry with Unified Stream
   Manager Extension](#co-configure-cp-sr-extensions).

   The Unified Stream Manager extension enables Confluent Platform Schema Registry to forward schema write operations to a
   remote (Confluent Cloud) Schema Registry endpoint—effectively allowing centralized schema
   management. This is achieved by configuring Confluent Platform Schema Registry with Unified Stream Manager extension
   settings, including authentication for Confluent Cloud Schema Registry and endpoint details.
2. [Create and deploy a schema exporter using a SchemaExporter custom
   resource](#co-setup-schema-exporter).

   (CR) to export existing schemas from Confluent Platform Schema Registry to Confluent Cloud Schema Registry, ensuring
   consistency between the two environments.
3. CFK switches over from the Confluent Platform Schema Registry to the Confluent Cloud Schema Registry as the primary
   Schema Registry.
4. [Create and deploy a schema importer using a SchemaImporter CR](#co-setup-schema-importer).

   A schema importer pulls schemas from Confluent Cloud Schema Registry back to Confluent Platform Schema Registry,
   maintaining updates for applications running locally or in the cloud.

#### NOTE
CFK only enables migration. It does not perform full orchestration.

## Requirements

* A Kubernetes cluster with CFK and Confluent Platform installed.
* Access to Confluent Cloud Schema Registry enabled
* The kubectl CLI tool configured to access your cluster
* Confluent Cloud API credentials for Confluent Cloud Schema Registry access
* Before triggering the switchover, clean all mode settings at the subject and
  context levels for Confluent Platform Schema Registry to avoid any context synchronization failures.

  For an example of settings and steps, see the [GitHub example for cleaning mode settings](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/hybrid/sr-automation-workflow/clean-mode-settings).

<a id="co-configure-cp-sr-extensions"></a>

## Enable Schema Registry extensions for switchover automation

Configure the Schema Registry cluster with the required extensions enabled:

* Unified Stream Manager extension

  This extension allows your on-prem Confluent Platform Schema Registry to interact with the remote
  Schema Registry in Confluent Cloud.
* Schema exporter extension
* Schema importer extension

Enable the Unified Stream Manager extension in the Schema Registry CR and apply the changes with the
`kubectl apply` command.

```yaml
kind: SchemaRegistry
spec:
  unifiedStreamManager:
    enabled:                 --- [1]
    remoteSchemaRegistry:
      context:               --- [2]
      endpoint:              --- [3]
      authentication:
        type:                --- [4]
        basic:
          secretRef:         --- [5]
  enableSchemaImporter:      --- [6]
  enableSchemaExporter:      --- [7]
```

* [1] Required. Set to `true` to enable the Unified Stream Manager extension.
* [2] Optional. Specify the context of the remote Schema Registry. If not specified, the default context is used.
* [3] Required. Specify the private endpoint of the remote Schema Registry, for example:
  `http://lsrc-abc.us-west-2.aws.private.confluent.cloud`. Public Schema Registry endpoints are not supported.
* [4] Required. Specify the authentication method to use for the remote Schema Registry.
  Only basic authentication is currently supported for the Confluent Cloud endpoint in
  the Unified Stream Manager extension.
* [5] Required. Specify the name of the secret that contains the credentials
  for the remote Confluent Cloud Schema Registry.
* [6] Set to `true` to enable the schema importer.
* [7] Set to `true` to enable the schema exporter.

<a id="co-setup-schema-exporter"></a>

## Set up a schema exporter

When the Schema Registry is running, set up a schema exporter on the Confluent Platform Schema Registry to sync the
Confluent Platform Schema Registry to the Confluent Cloud Schema Registry.

For details about the SchemaExporter CR, see [Create schema exporter](co-link-schemas.md#co-create-schema-exporter).

1. Set up a SchemaExporter CR and apply it on the Confluent Platform Schema Registry to sync the Confluent Platform
   Schema Registry to the Confluent Cloud Schema Registry.

   For more information about the SchemaExporter CR properties, see
   [Create schema exporter](co-link-schemas.md#co-create-schema-exporter).
   ```yaml
   kind: SchemaExporter
   metadata:
     name:                   --- [1]
   spec:
     sourceCluster:
       schemaRegistryClusterRef:
         name:               --- [2]
         namespace:          --- [3]
     destinationCluster:
       schemaRegistryRest:
         endpoint:           --- [4]
         authentication:
           type:             --- [5]
           basic:
             secretRef:      --- [6]
     subjects:               --- [7]
     contextName:            --- [8]
   ```

   * [1] Required. The name of the schema exporter. This name must match the
     `exporterName` value in the SchemaImporter CR
     (`SchemaImporter.spec.exporterName`) you set in
     [Set up a schema importer](#co-setup-schema-importer).
   * [2] The name of the source Schema Registry cluster.
   * [3] The namespace of the source Schema Registry cluster.
   * [4] The endpoint of the Confluent Cloud Schema Registry.
   * [5] The authentication type of the Confluent Cloud Schema Registry. The supported type is
     `basic`.
   * [6] The API key of the Confluent Cloud Schema Registry.
   * [7] The subjects to export. The default value of subjects is `"*"`, which
     denotes all subjects in the default context.

     To export all subjects from Confluent Platform to Confluent Cloud, you can set the subjects to
     `:*:` and contextType as `NONE`.

     You can choose to create multiple exporters and importers if using Scenario
     2 listed in [Use Schema Registry in a hybrid setup](http://docs.confluent.io/platform/current/usm/usm-schema.html#configuration-scenarios).
   * [8] The context name of the Schema Registry cluster. Use an empty string for the
     default context.

     This value must match `SchemaRegistry.spec.unifiedStreamManager.context` for the Unified Stream Manager Schema Registry to work as expected.

   An example SchemaExporter CR snippet:
   ```yaml
   apiVersion: platform.confluent.io/v1beta1
   kind: SchemaExporter
   metadata:
     name: schema-exporter
     namespace: my-namespace-1
   spec:
     sourceCluster:
       schemaRegistryClusterRef:
         name: schemaregistry
         namespace: my-namespace-2
     destinationCluster:
       schemaRegistryRest:
         endpoint: https://<ccloud-sr-endpoint>
         authentication:
           type: basic
           basic:
             secretRef: cc-sr-credential
     subjects: [":*:"]
     contextName: my-context
   ```
2. Wait until the exporter enters the RUNNING state, which confirms that schemas
   are synced; in-flight changes are handled automatically.
3. Ensure that the exporter destination URL and any forwarding settings match
   those used by the forwarder/Unified Stream Manager configuration.

<a id="co-setup-schema-importer"></a>

## Set up a schema importer

A schema importer triggers and automates the setup and state management for
schema import from Confluent Cloud to Confluent Platform, coordinating migration steps and serving as
a key trigger for reversing synchronization direction. It makes hybrid schema
governance workflows easier to automate and manage in Kubernetes environments.

This workflow is triggered after schemas are initially exported from Confluent Platform Schema Registry
to Confluent Cloud Schema Registry using a SchemaExporter CR.

The automation includes changing modes on both Schema Registry clusters to ensure cloud
is the system of record and on-prem is a read cache and reverse importer.

If the `schemaRegistryAutomation` block is not specified in the SchemaImporter
CR, the automation workflow is not triggered. It would simply deploy an importer
that would sync schemas from the Confluent Cloud Schema Registry to the Confluent Platform Schema Registry.

1. Deploy your schema importer using a SchemaImporter CR.

   Set up a SchemaImporter CR and apply it on the Confluent Platform Schema Registry to sync the Confluent Cloud Schema Registry to
   the Confluent Platform Schema Registry.
   ```yaml
   kind: SchemaImporter
   spec:
     contextName:              --- [1]
     subjects:                 --- [2]
     destinationCluster:       --- [3]
       schemaRegistryClusterRef:
         name:                 --- [4]
         namespace:            --- [5]
     sourceCluster:
       schemaRegistryRest:     --- [6]
         endpoint:             --- [7]
         authentication:
           type:               --- [8]
           basic:
             secretRef:        --- [9]
     schemaRegistryAutomation: --- [10]
       exporterName:           --- [11]
       enableForwardWriteMode: --- [12]
   ```

   * [1] The destination context in Confluent Platform Schema Registry.

     `SchemaRegistry.spec.unifiedStreamManager.context` must match the
     destination context of the exporter for which the automation workflow
     is triggered.
   * [2] Required for Schema Registry automation workflow. Only a single element is
     allowed. Supported values are:
     * `":*:"` representing all subjects in all contexts
     * `":.<my-site>:*"` representing all subjects in the `<my-site>`
       context. For example, `":.my-cfk:*"`.
   * [3] Points to your on-premises Confluent Platform Schema Registry.
   * [4] The name of the destination Schema Registry cluster.
   * [5] The namespace of the destination Schema Registry cluster.
   * [6] Contains the REST endpoint and authentication for the Confluent Cloud Schema Registry.
   * [7] The REST endpoint of the Confluent Cloud Schema Registry.
   * [8] The authentication type of the Confluent Cloud Schema Registry. The supported type is
     `basic`.
   * [9] The API key of the Confluent Cloud Schema Registry.
   * [10] Required to trigger the automation workflow.

     When the `schemaRegistryAutomation` block is specified in the
     SchemaImporter CR, the automation workflow is triggered for CFK to perform
     the switchover and import.

     If `schemaRegistryAutomation` block is not specified, the automation workflow is not triggered. It would simply deploy an importer that would sync schemas from the Confluent Cloud Schema Registry to the Confluent Platform Schema Registry.
   * [11] or [12] Either `exporterName` ([11]) is specified or
     `enableForwardWriteMode` ([12]) is set to `true` to trigger the
     automation workflow based on the requirement.
   * [11] Specify the name of the schema exporter. Must match the name of the
     SchemaExporter CR (`SchemaExporter.metadata.name`) you set in
     [Set up a schema exporter](#co-setup-schema-exporter) that handled the export from Confluent Platform Schema Registry to
     Confluent Cloud Schema Registry.
   * [12] Set to `true` to trigger the automation workflow without specifying
     the exporter.

   An example SchemaImporter CR snippet:
   ```yaml
   kind: SchemaImporter
   metadata:
     name: schema-importer
     namespace: source
   spec:
     sourceCluster:
       schemaRegistryRest:
         endpoint: https://psrc-stgc48jdn27.us-east-1.aws.stag.cpdev.cloud
         authentication:
           type: basic
           basic:
             secretRef: source-credential
     destinationCluster:
       schemaRegistryClusterRef:
         name: schemaregistry
         namespace: source
     schemaRegistryAutomation:
       exporterName: "schema-exporter"
     contextName: ""
     subjects: [":*:"]
   ```
2. Wait for the schema importer status to be in RUNNING state.
3. If the automation workflow is triggered (by specifying
   `schemaRegistryAutomation` in the SchemaImporter CR in the previous step),
   CFK transitions Confluent Platform Schema Registry to FORWARD mode at the global level (not per
   subject) and transitions Confluent Cloud Schema Registry to READWRITE mode to establish
   Confluent Cloud Schema Registry as the authoritative write destination.

   If `schemaRegistryAutomation` is not specified and the automation workflow is not triggered, CFK will simply deploy an importer that syncs schemas from the Confluent Cloud Schema Registry to the Confluent Platform Schema Registry.
4. Monitor workflow status and handle errors.
   1. The status field (`AutomationWorkflowStatus`) from Schema Registry/CFK indicates
      workflow progress and errors. If any API-based transitions fail (FORWARD
      mode, READWRITE mode), CFK will revert to previous states and exit the
      workflow.

      Example error cases are:
      * `AUTOMATION_STATUS_PRECHECKS_FAILED`: The precheck conditions for triggering the automation workflow failed.
      * `AUTOMATION_STATUS_SKIPPED`: The automation workflow was skipped as importer already exists.
   2. Fix the root causes for failures before retrying.

### Schema importer operations

The following schema importer operations are supported by CFK. Add the specified annotation to the SchemaImporter CR with the `kubectl annotate` command.

* Pause Schema Importer
  ```bash
  kubectl annotate schemaimporter schema-importer \
    platform.confluent.io/pause-schema-importer="true" \
    -n <namespace>
  ```
* Resume Schema Importer
  ```bash
  kubectl annotate schemaimporter schema-importer \
    platform.confluent.io/resume-schema-importer="true" \
    -n <namespace>
  ```

<a id="co-schema-registry-switchover-examples"></a>

## Configuration scenarios

Select and configure one of the following scenarios that matches your requirements.

You can get more information about the configuration scenarios at [Unified Stream Manager - Schema Configuration scenarios](http://docs.confluent.io/platform/current/usm/usm-schema.html#configuration-scenarios).

### Mirror schemas

Choose this path to ensure your Schema Registry on Confluent Platform environments have exact replicas of the schemas in your Confluent Cloud Schema Registry. This approach is ideal for maintaining identical schema sets across both environments. To avoid schema ID collisions, ensure the Confluent Cloud Schema Registry either has no schemas to start with or already contains the same set of schemas as your Confluent Platform registry.

See the [sample configuration for mirror schemas scenario](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/hybrid/sr-automation-workflow/mirror-schemas) in GitHub.

### Use distinct schemas and contexts across Confluent Platform environments

Choose this path if you have multiple, separate Confluent Platform environments and you want to sync them all to the same Confluent Cloud instance. Using a distinct schema context for each Confluent Platform environment prevents schema ID collisions and is ideal for centralizing governance across a multi-cluster estate.

See the [sample configuration for distinct schemas and contexts scenario](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/hybrid/sr-automation-workflow/distinct-schemas-and-contexts) in GitHub.

### Greenfield Confluent Platform (no initial schemas in Confluent Platform)

Choose this path if you are starting with a greenfield Confluent Platform environment with no initial schema in Confluent Platform and you want to sync it with Confluent Cloud Schema Registry.

See the [sample configuration for greenfield scenario](https://github.com/confluentinc/confluent-kubernetes-examples/tree/master/hybrid/sr-automation-workflow/greenfield-cp) in GitHub.

## Troubleshooting

The following are a few of the common issues that you may encounter in the Schema Registry switchover workflow:

* Schema Registry not starting: Check if Kafka is fully ready before deploying Schema Registry.
* Export/import failing: Verify Confluent Cloud credentials and network connectivity.
* Permission errors: Ensure that the Confluent Cloud API key has the necessary permissions
  for Schema Registry operations.