# Schema Linking on Confluent Cloud Use Schema Linking to keep schemas in sync across two Schema Registry clusters, typically alongside Cluster Linking for full data synchronization. The quick start below walks through creating and using exporters for Schema Linking on your clusters, followed by reference material for schema contexts and exporters. Contexts are also useful outside of Schema Linking to organize schemas into purpose-specific groups and create virtual “sub-registries.” ![An exporter copies a schema from source to destination |sr|.](images/schema-linking.png) ## Contexts and exporters Schema Linking uses two Schema Registry concepts, schema contexts and schema exporters: - **Contexts** - A [context](#schema-contexts) is an independent scope in Schema Registry. Use contexts to create separate “sub-registries” within one Schema Registry cluster, each with its own schema IDs and subject names. The same schema ID in different contexts can represent different schemas. A single dot, `.`, represents the default context. An explicit context starts with a dot and uses more dots as separators, such as `.mycontext.subcontext`, like absolute Unix paths. Two contexts that share a prefix have no implicit relationship. - **Exporters** - A [schema exporter](#schema-exporters) runs inside Schema Registry and exports schemas from one Schema Registry cluster to another. Use the Schema Registry REST API to create, pause, resume, and destroy exporters. An exporter acts like a “mini-connector” that performs change data capture for schemas. ## Limitations and considerations - On Confluent Cloud, the exporter limit per environment is 10 with the Essentials package and 100 with the Advanced package. Refer to the [Packages documentation](/cloud/current/stream-governance/packages.html#overview). - A single exporter can transfer any number of schemas; there is no upper limit on schemas per exporter. For in-depth descriptions, refer to [Contexts](#schema-contexts) and [Exporters](#schema-exporters). ## Quick start Before you begin, install the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/install.html). ### Set up two clusters in source and destination environments For this Quick Start, you set up a source and destination Kafka cluster, each in its own environment. Because Schema Registry automatically provisions one cluster per environment, this gives you two registries for testing Schema Linking across subjects. 1. Log on to [Confluent Cloud](https://confluent.cloud/login) with your user account, and create two new environments: one named “SOURCE” and the other named “DESTINATION”. Navigate to **Environments** (top right menu), click **Add cloud environment**, and follow the steps. When you create an environment, you must choose a Stream Governance package. For Schema Linking, both **Essentials** and **Advanced** work. 2. Add a Kafka cluster to each environment. These can be any [cluster type](../clusters/cluster-types.md#cloud-cluster-types). For example, name them `my-source-cluster` and `my-destination-cluster`. 3. Note the Schema Registry cluster ID and API endpoint for each environment, using either the Confluent Cloud Console or the Confluent CLI. You will need these in the next steps. - On the Confluent Cloud Console, navigate to an environment and select the cluster. The Schema Registry cluster ID and API endpoint are shown on the right panel of the environment level display under **Stream Governance API**. - On the Confluent CLI, [log in](https://docs.confluent.io/confluent-cli/current/command-reference/confluent_login.html), then select an environment with [confluent environment list](https://docs.confluent.io/confluent-cli/current/command-reference/environment/confluent_environment_list.html) and [confluent environment use ](https://docs.confluent.io/confluent-cli/current/command-reference/environment/confluent_environment_use.html). Run [confluent schema-registry cluster describe](https://docs.confluent.io/confluent-cli/current/command-reference/schema-registry/cluster/confluent_schema-registry_cluster_describe.html) to get details for the Schema Registry cluster in that environment. For example, navigate to each environment and run `confluent schema-registry cluster describe` to see output similar to the following: ```bash my-laptop:~ me$ confluent env use env-src Using environment "env-src". my-laptop:~ me$ confluent schema-registry cluster describe +-------------------------+--------------------------------------------------+ | Name | Always On Stream Governance | | | Package | | Cluster | lsrc-x6612x | | Endpoint URL | https://psrc-22y2ny.us-west2.gcp.confluent.cloud | | Used Schemas | 0 | | Available Schemas | 100 | | Free Schemas Limit | 100 | | Global Compatibility | BACKWARD | | Mode | READWRITE | | Cloud | GCP | | Region | us-west2 | | Package | ESSENTIALS | +-------------------------+--------------------------------------------------+ ``` ```bash my-laptop:~ me$ confluent env use Using environment "env-dest". my-laptop:~ me$ confluent schema-registry cluster describe +-------------------------+--------------------------------------------------+ | Name | Always On Stream Governance | | | Package | | Cluster | lsrc-jzzx6w | | Endpoint URL | https://psrc-22y2ny.us-west2.gcp.confluent.cloud | | Used Schemas | 0 | | Available Schemas | 100 | | Free Schemas Limit | 100 | | Global Compatibility | BACKWARD | | Mode | READWRITE | | Cloud | GCP | | Region | us-west2 | | Package | ESSENTIALS | +-------------------------+--------------------------------------------------+ ``` ### Configure credentials on source and destination In the next steps, you will set up permissions to allow access to the source and destination registries in support of Schema Linking. #### Configure credentials on the source Follow the steps in this section to create an API key and secret to authenticate your user account to the source Schema Registry. In this section, you will: - Create an API key and secret to authenticate the user account to the Schema Registry on the source. - Give the user account “ResourceOwner” permissions ([RBAC roles](#cloud-schema-linking-rbac)) on all Schema Registry subjects on the source. These are the permissions needed to create an exporter on the source, for the purposes of the examples in this Quick Start. In general, you only need permissions for the subjects you want to export; this will vary per each use case. 1. Navigate to the SOURCE environment and create an API key associated with the Schema Registry on the SOURCE. For example: ```bash confluent api-key create --resource ``` Save the SOURCE API key and secret in a secure location. The secret cannot be retrieved later. 2. Review source access information. By now you should have a triplet of Schema Registry access details for the source cluster: - Schema Registry URL (the API endpoint) - API key - API secret 3. Assign the RBAC role “ResourceOwner” on the Schema Registry on the source as follows to give the user account permissions on all schema subjects under all contexts: `--subjects ":*:"`. (For a broader scope, you could also use “OrganizationAdmin” which is applicable to the organization level, or “EnvironmentAdmin” or “DataSteward” roles, which are scoped to the environment level.) ```bash confluent iam rbac role-binding create --principal User: --role ResourceOwner --environment --schema-registry-cluster --resource "Subject:*" ``` Your output should resemble: ```bash +---------------+----------------+ | Principal | User:u-qqddpm | | Role | ResourceOwner | | Resource Type | Subject | | Name | * | | Pattern Type | LITERAL | +---------------+----------------+ ``` #### Configure credentials on the destination #### IMPORTANT - These are the credentials the exporter will use to access the destination. - For Schema Linking exporters in production, always use API keys tied to service accounts, not users. For now, you must use the CLI to create API keys tied to service accounts, as the Confluent Cloud Console does not currently support service accounts for Schema Linking. To learn more, see [Add an API key](../security/authenticate/workload-identities/service-accounts/api-keys/manage-api-keys.md#create-resource-specific-api-key) and [Best Practices for Using API Keys on Confluent Cloud](../security/authenticate/workload-identities/service-accounts/api-keys/best-practices-api-keys.md#api-keys-best-practices). In this section, you will: - Create a service account on the destination. - Create an API key and secret to authenticate the service account to the Schema Registry on the destination. - Give the service account permissions ([RBAC roles](#cloud-schema-linking-rbac)) on Schema Registry subjects on the destination. These are the permissions needed to copy from SOURCE to DESTINATION. 1. Navigate to the DESTINATION environment and run the command [confluent iam service-account create](https://docs.confluent.io/confluent-cli/current/command-reference/iam/service-account/confluent_iam_service-account_create.html). For example: ```bash confluent iam service-account create schema-link-destination --description "Service account for Schema Registry on the Destination cluster" ``` Your output should resemble: ```bash +-------------+--------------------------------+ | ID | sa-123abc | | Name | sr-desintation-demo | | Description | Service account for Schema | | | Registry on the Destination | | | cluster | +-------------+--------------------------------+ ``` 2. Create an API key and secret, and associate it with the service account you just created to authenticate to the destination Schema Registry. You will need the resource ID for the DESTINATION Schema Registry, which you can find on the Confluent Cloud UI in the destination environment on the right panel under “Stream Governance API”, or on the Confluent CLI with the command `confluent schema-registry cluster describe`. For example: ```bash confluent api-key create --service-account sa-123abc --resource --description "Destination Schema Registry API Key" ``` Save the DESTINATION API key and secret in a secure location. The secret cannot be retrieved later. 3. Review destination access information. By now you should have the following Schema Registry access details for the destination Schema Registry: - Schema Registry URL (the API endpoint) - Schema Registry cluster ID (resource ID) - Service account ID - API key - API secret 4. Assign the RBAC role “ResourceOwner” on the Schema Registry as follows to give the service account permissions on all schema subjects. ```bash confluent iam rbac role-binding create --principal User: --role ResourceOwner --environment --schema-registry-cluster --resource "Subject:*" ``` Your output should resemble: ```bash +---------------+----------------+ | Principal | User:sa-qqddpm | | Role | ResourceOwner | | Resource Type | Subject | | Name | * | | Pattern Type | LITERAL | +---------------+----------------+ ``` ### Create credentials for the exporter Create a `config.txt` file with the URL and credentials your exporter uses to access the destination cluster. The exporter copies schemas from the source and writes linked copies to the destination, so it needs destination credentials. The config gives the exporter the service account and RBAC roles you set up in [Configure credentials on the destination](#schema-linking-credentials-destination). ```bash schema.registry.url= basic.auth.credentials.source=USER_INFO basic.auth.user.info=: ``` You can find the Destination Schema Registry URL (endpoint) either on the Cloud Console right panel of the DESTINATION environment under “Stream Governance API” or on the Confluent CLI by running the following command: ```bash confluent schema-registry cluster describe --environment ``` With the CLI command, your output should resemble the following: ```bash +-------------------------+--------------------------------------------------+ | Name | Always On Stream Governance | | | Package | | Cluster | lsrc-jzzx6w | | Endpoint URL | https://psrc-22y2ny.us-west2.gcp.confluent.cloud | | Used Schemas | 0 | | Available Schemas | 100 | | Free Schemas Limit | 100 | | Global Compatibility | BACKWARD | | Mode | READWRITE | | Cloud | GCP | | Region | us-west2 | | Package | ESSENTIALS | +-------------------------+--------------------------------------------------+ ``` Security considerations: : Exporters continue running even if the creator’s access is revoked, suspended, or deleted. The system does not perform continuous authorization checks after creation. Recommended action: : Monitor your centralized audit logs for Schema Registry management events to detect unauthorized exporter creation or changes. To learn more, see [audit logs](../monitoring/audit-logging/event-methods/sr-management.md#audit-log-schema-exporter-examples). ### Create schemas on the source Create at least three schemas in the source environment; at least one of which has a [qualified subject name](#schema-link-qualified-subject). To create each schema from the Cloud Console, follow these steps: 1. From the **Schema Registry** tab in the SOURCE environment, Click **Schemas** on the right panel for an environment, then **Add schema**. 2. Fill in the subject name. - To create a schema with an unqualified subject name, simply provide a name such as `coffee` or `donuts`. ![image](images/schema-link-unqualified-subject.png) - To create a schema with a qualified subject name in a specified context, use the syntax: `:.:`. For example: `:.snowcones:sales` or `:.burgers:locations`. ![image](images/schema-link-qualified-subject.png) On the Cloud Console, enter the following to create four subject names, two in the default context, and two qualified in custom contexts. Note that the qualified context names are prefixed by a dot (”**.**”): | Schema Context | Subject Name | |------------------|----------------| | default | coffee | | default | donuts | | .snowcones | sales | | .burgers | locations | 3. Use the following example for your Avro content for each of the schemas. The goal of this Quick Start is to learn how to organize schemas under subject names and contexts, not to design schemas. You can use the default schema, schemas you already have, or the example below. **Example Schema** ```json { "type": "record", "namespace": "com.mycorp.mynamespace", "name": "sampleRecord", "fields": [ { "name": "item", "type": "string" }, { "name": "location", "type": "string" }, { "name": "cost", "type": "double" }, { "name": "code", "type": "int" } ] } ``` 4. Click **Create**. 5. When you have created your subjects, review the list on the Schema Registry tab on the SOURCE, which should look similar to the following. ![image](images/schema-link-list-of-subjects.png) ### Create the exporter on the source 1. Switch to the SOURCE environment using `confluent environment list` and `confluent environment use `. 2. Create a new exporter using the [confluent schema-registry exporter create](https://docs.confluent.io/confluent-cli/current/command-reference/schema-registry/exporter/confluent_schema-registry_exporter_create.html) command. For this demo, you want the exporter to copy all schemas, including those in specific contexts (other than the default), so include the `--subjects` flag with the context wildcard to denote subjects under all contexts: `--subjects ":*:"` ```bash confluent schema-registry exporter create --subjects ":*:" --config /config.txt ``` For example, this command creates an exporter called “my-first-exporter” that will export all schemas, including those in specific contexts as well as those in the default context. (The `config.txt` for this example lives in the user’s home directory.): ```bash confluent schema-registry exporter create my-first-exporter --subjects ":*:" --config ~/config.txt ``` You should get output verifying that the exporter was created: `Created schema exporter "my-first-exporter".` You can list exporters with [confluent schema-registry exporter list](https://docs.confluent.io/confluent-cli/current/command-reference/schema-registry/exporter/confluent_schema-registry_exporter_list.html) and check the status of an exporter with confluent [confluent schema-registry exporter status describe ](https://docs.confluent.io/confluent-cli/current/command-reference/schema-registry/exporter/confluent_schema-registry_exporter_get-status.html). #### More options for exporters To create exporters that target specific subjects and contexts, customize the following parameters when you run `confluent schema-registry exporter create`: | Parameter | Default | Description | |--------------------|----------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | `--subjects` | `*` (default context only) | Comma-separated list of subjects to export (for example, `sales,coffee,donuts`). Use `:*:` to export from all contexts. Use `:.:*` to export only the default context. | | `--context-type` | `AUTO` | Context-handling mode. Set to `CUSTOM` to write subjects under a specific context name on the destination. | | `--context-name` | (none) | Destination context name. Required when `--context-type` is `CUSTOM`. | | `--subject-format` | (none) | Template for renaming subjects on the destination. Use `${subject}` as a placeholder for the source subject name (for example, `dc_${subject}` rewrites `orders` to `dc_orders`). Available with `confluent schema-registry exporter create` and `confluent schema-registry exporter update`. | Example with custom subjects and context: ```bash confluent schema-registry exporter create --subjects , \ --context-type CUSTOM --context-name \ --config ~/config.txt ``` You can create and run multiple exporters at once, so feel free to circle back at the end of the Quick Start to create and test more exporters with different parameters. For complete details, refer to [Configuration options](#exporter-configs). ### Verify the exporter is running and view information about it Still in the SOURCE environment, run the following commands. 1. List available exporters. ```bash confluent schema-registry exporter list ``` Your exporter will show in the list. 2. Describe the exporter. ```bash confluent schema-registry exporter describe ``` For example, fill in `my-first-exporter` for ``: ```bash confluent schema-registry exporter describe my-first-exporter ``` Your output should resemble: ```bash confluent schema-registry exporter describe my-first-exporter +----------------+------------------------------------------------------------------------+ | Name | my-first-exporter | | Subjects | * | | Subject Format | ${subject} | | Context Type | AUTO | | Context | .lsrc-x6612x | | Config | basic.auth.credentials.source="USER_INFO" | | | basic.auth.user.info="[hidden]" | | | schema.registry.url="https://psrc-22y2ny.us-west2.gcp.confluent.cloud" | +----------------+------------------------------------------------------------------------+ ``` 3. Get configurations for the exporter. ```bash confluent schema-registry exporter configuration describe my-first-exporter ``` 4. Get the status of exporter. ```bash confluent schema-registry exporter status describe my-first-exporter ``` Your output should resemble: ```bash confluent schema-registry exporter status describe my-first-exporter +-------------+-------------------+ | Name | my-first-exporter | | State | RUNNING | | Offset | 218567 | | Timestamp | 1711598116280 | +-------------+-------------------+ ``` 5. Finally, as a check, get a list of schemas on the source. Use the prefix wildcard to list all schemas: ```bash confluent schema-registry subject list --prefix ":*:" ``` With the wildcard, this is effectively the same as the command: `confluent schema-registry subject list` The command will return the list of subjects you’ve created on the source, for example: ```bash Subject ----------------------- :.burgers:locations :.snowcones:sales coffee donuts ``` ### Check that the schemas were exported Switch to the DESTINATION environment and verify that your source schemas appear there. 1. Switch to the DESTINATION. Use `confluent environment list` and `confluent environment use to navigate. 2. Run the following command to view all schemas. ```bash confluent schema-registry subject list --prefix ":*:" ``` Your output list of schemas on the DESTINATION should match those on the SOURCE. ```bash Subject ---------------------------------- :.lsrc-jzzx6w.burgers:locations :.lsrc-jzzx6w.snowcones:sales :.lsrc-jzzx6w:coffee :.lsrc-jzzx6w:donuts ``` 3. List only schemas in particular contexts. One you have a list of all subjects on the destination with the prefixes (as in the above example), you can pass only the context name to see a narrowed list of subjects in a particular context. - For example, to list schemas in the `burgers` context, where `lsrc-jzzx6w` is the destination Schema Registry cluster ID: ```bash confluent schema-registry subject list --prefix ":.lsrc-jzzx6w.burgers:" ``` The output will be: ```bash Subject ---------------------------------- :.lsrc-jzzx6w.burgers:locations ``` - To list schemas in the `snowcones` context: ```bash confluent schema-registry subject list --prefix ":.lsrc-jzzx6w.snowcones:" ``` The output will be: ```bash Subject -------------------------------- :.lsrc-jzzx6w.snowcones:sales ``` ### Pause the exporter and make changes 1. Pause the exporter. Switch back to the SOURCE, and run the following command to pause the exporter. ```bash confluent schema-registry exporter pause ``` You should get output verifying that the command was successful. For example: `Paused schema exporter "my-first-exporter".` Check the status, just to be sure. ```bash confluent schema-registry exporter status describe ``` Your output should resemble: ```bash confluent schema-registry exporter status describe my-first-exporter +--------------------+-------------------+ | Name | my-first-exporter | | Exporter State | PAUSED | | Exporter Offset | 10011386 | | Exporter Timestamp | 1631107710822 | +--------------------+-------------------+ ``` 2. Reset the schema exporter offset back to `schema.id=1`. - A reset will restart the incremental iteration through the source schemas. ```bash confluent schema-registry exporter reset ``` - After the reset, you can verify by getting the status of the exporter. ```bash confluent schema-registry exporter status describe ``` The status will show that the offset is reset. For example: ```bash confluent schema-registry exporter status describe my-first-exporter +--------------------+-------------------+ | Name | my-first-exporter | | Exporter State | PAUSED | | Exporter Offset | -1 | | Exporter Timestamp | 0 | +--------------------+-------------------+ ``` #### IMPORTANT In order to process all schemas with the new configurations, you must either reset the exporter offset (as described above), or delete and recreate the exporter. 3. Update exporter configurations or information. You can choose to update any of `subjects`, `context-type`, `context-name`, or `config`. For example: ```bash confluent schema-registry exporter update --context-name ``` 4. Resume schema exporter. ```bash confluent schema-registry exporter resume ``` ### Delete the exporter When you are ready to wrap up your testing, pause and then delete the exporter(s) as follows. 1. Pause the exporter. ```bash confluent schema-registry exporter pause ``` 2. Delete the exporter. ```bash confluent schema-registry exporter delete ``` This concludes the Quick Start. The remaining sections describe Schema Linking concepts, monitoring, RBAC, and OAuth configuration in detail. ## Contexts ### What is a context? A schema context, or simply context, is essentially a grouping of subject names and schema IDs. A single Schema Registry cluster can host any number of contexts. Each context can be thought of as a separate “sub-registry”. A context can also be copied to another Schema Registry cluster, using a [schema exporter](#schema-exporters). ### How contexts work Following are a few key aspects of contexts and how they help to organize schemas. #### Schemas and schema IDs are scoped by context Subject names and schema IDs are scoped by context so that two contexts in the same Schema Registry cluster can each have a schema with the same ID, such as `123`, or a subject with the same name, such as `mytopic-value`, without any problem. To put this another way, subject names and schema IDs are unique per context. You can have schema ID `123` in context `.mycontext` and schema ID `123` in context `.yourcontext` and these can be different from one another. #### Default context Any schema ID or subject name without an explicit context lives in the default context, which is represented as a single dot `.`. An explicit context starts with a dot and can contain any parts separated by additional dots, such as `.mycontext.subcontext`. You can think of context names as similar to absolute Unix paths, but with dots instead of forward slashes (in this analogy, the default schema context is like the root Unix path). However, there is no relationship between two contexts that share a prefix. #### Global context Schema Registry supports a global configuration context using the special context name `:.__GLOBAL:`. When looking up a configuration or mode with `defaultToGlobal`, the lookup proceeds from the subject to the context, and then to the global context (`:.__GLOBAL:`). The global context is only valid for configuration and mode settings; it cannot be used to store subjects or schemas. For contexts without global configuration, compatibility is handled per context. To learn more, see [Schema Evolution and Compatibility for Schema Registry on Confluent Cloud](fundamentals/schema-evolution.md#schema-evolution-and-compatibility). The global context provides a hierarchical configuration lookup mechanism: 1. **Subject level**: Configuration is first checked at the specific subject level. 2. **Context level**: If not found at the subject level, configuration is checked at the context level. 3. **Global level**: Finally, if not found at the context level, configuration falls back to the global context (`:.__GLOBAL:`). This hierarchy allows you to: - Set default configurations that apply across all contexts using the global context. - Override global defaults with context-specific configurations. - Further override with subject-specific configurations for fine-grained control. The global context is particularly useful for establishing organization-wide defaults for compatibility modes while maintaining the flexibility to customize behavior for specific contexts or subjects as needed. #### Qualified subjects A subject name can be qualified with a context, in which case it is called a qualified subject. When a context qualifies a subject, the context must be surrounded by colons. An example is `:.mycontext:mysubject`. A subject name that is unqualified is assumed to be in the default context, so that `mysubject` is the same as `:.:mysubject` (the dot representing the default context). The one exception to this rule is in the case of a subject name specified for a schema reference. In this case, if the subject name of the schema reference is not qualified with an explicit context, it inherits the context in which the root schema resides. For example, if the following set of schema references are associated with a schema in a subject in the `.prod` context: ```json [ { "name": "io.confluent.examples.avro.Customer", "subject": "customer", "version": 1 }, { "name": "io.confluent.examples.avro.Product", "subject": "product", "version": 1 }, { "name": "io.confluent.examples.avro.Order", "subject": "order", "version": 1 } ] ``` then, the subject names above, `customer`, `product`, and `order`, will refer to the subjects `:.prod:customer`, `:.prod:product`, and `:.prod:order`, respectively. There are two ways to pass a context to the REST APIs. - [Using a qualified subject](#qualified-subject) - [Using a base context path](#base-context-path) ### Using a qualified subject A qualified subject can be passed anywhere that a subject name is expected. Most REST APIs take a subject name, such as `POST /subjects/{subject}/versions`. There are a few REST APIs that don’t take a subject name as part of the URL path: - `/schemas/ids/{id}` - `/schemas/ids/{id}/subjects` - `/schemas/ids/{id}/versions` The three APIs above can now take a query parameter named “subject” (written as `?subject`), so you can pass a qualified subject name, such as `/schemas/ids/{id}?subject=:.mycontext:mysubject`, and the given context is then used to look up the schema ID. ### Using a base context path As mentioned, all APIs that specify an unqualified subject operate in the default context. Besides passing a qualified subject wherever a subject name is expected, a second way to pass the context is by using a base context path. A base context path takes the form `/contexts/{context}` and can be prepended to any existing Schema Registry path. Therefore, to look up a schema ID in a specific context, you could also use the URL `/contexts/.mycontext/schemas/ids/{id}`. A base context path can also be used to operate with the default context. In this case, the base context path takes the form “/contexts/:.:/”; for example, `/contexts/:.:/schemas/ids/{id}`. A single dot cannot be used because it is omitted by some URL parsers. ### Multi-Context APIs All the examples so far operate in a single context. There are three APIs that return results for multiple contexts. - `/contexts` - `/subjects` - `/schemas?subjectPrefix=:*:` The first two APIs, `/contexts` and `/subjects`, return a list of all contexts and subjects, respectively. The other API, `/schemas`, normally only operates in the default context. This API can be used to query all contexts by passing a `subjectPrefix` with the value `:*:`, called the *context wildcard*. The context wildcard matches all contexts. ### Specifying a context name for clients When using a client to talk to Schema Registry, you may want the client to use a particular context. An example of this scenario is when migrating a client from communicating with one Schema Registry to another. You can achieve this by using a base context path, as defined above. To do this, simply change the Schema Registry URL used by the client from `https://` to `https:///contexts/.mycontext`. Note that by using a base context path in the Schema Registry URL, the client will use the same schema context for every Schema Registry request. However, an advanced scenario might involve a client using different contexts for different topics. To achieve this, you can specify a context name strategy to the serializer or deserializer: - `context.name.strategy=com.acme.MyContextNameStrategy` The context name strategy is a class that must implement the following interface: ```none /** * A {@link ContextNameStrategy} is used by a serializer or deserializer to determine * the context name used with the schema registry. */ public interface ContextNameStrategy extends Configurable { /** * For a given topic, returns the context name to use. * * @param topic The Kafka topic name. * @return The context name to use */ String contextName(String topic); } ``` Again, the use of a context name strategy should not be common. Specifying the base context path in the Schema Registry URL should serve most needs. ## Exporters ### What is an Exporter? Previously, [Confluent Replicator](https://docs.confluent.io/platform/current/multi-dc-deployments/replicator/index.html) was the primary means of [migrating schemas](https://docs.confluent.io/platform/current/schema-registry/installation/migrate.html) from one Schema Registry cluster to another, as long as the source Schema Registry cluster was on-premise. To support schema migration using this method, the destination Schema Registry is placed in IMPORT mode, either globally or for a specific subject. The new schema exporter functionality replaces and extends the schema migration functionality of Replicator. Schema exporters reside within a Schema Registry cluster, and can be used to replicate schemas between two Schema Registry clusters in Confluent Cloud. ### Schema Linking You use schema exporters to accomplish Schema Linking, using contexts and/or qualified subject names to sync schemas across registries. Schema contexts provide the conceptual basis and namespace framework, while the exporter does the heavy-lift work of the linking. #### Schemas export from the source default context to a new context on the destination By default, a schema exporter exports schemas from the default context in the source Schema Registry to a new context in the destination Schema Registry. The destination context (or a subject within the destination context) is placed in IMPORT mode. This allows the destination Schema Registry to use its default context as usual, without affecting any clients of its default context. The new context created by default in the destination Schema Registry will have the form `.lsrc-xxxxxx`, taken from the logical name of the source. #### Schema Registry clusters can export schemas to each other Two Schema Registry clusters can each have a schema exporter that exports schemas from the default context to the other Schema Registry. In this setup, each side can read from or write to the default context, and each side can read from (but not write to) the exported context. This allows you to match the setup of [Cluster Linking](https://docs.confluent.io/cloud/current/multi-cloud/cluster-linking/index.html), where you might have a source topic and a read-only mirror topic on each side. #### Customizing exporters There are various ways to customize which contexts are exported from the source Schema Registry, and which contexts are used in the destination Schema Registry. The full list of configuration properties is shown below. #### How many exporters are allowed per Schema Registry? The limit on the number of exporters allowed at any one time per Schema Registry is 10. ### Configuration options An exporter has these main configuration properties: `name` : A unique name for the exporter. `subjects` : This can take several forms:
- A list of subject names, for example, `[ "subject1", "subject2" ]` - A singleton list containing a subject name prefix that ends in a wildcard, such as `["mytopic*"]` - A singleton list containing a lone wildcard, `["*"]`, that indicates all subjects in the default context. This is the default. - A singleton list containing the context wildcard, `[":*:"]`, that indicates all contexts. - A singleton list containing a context prefix that ends in a wildcard, such as `[":.mycontext:*"]`, to export subjects from multiple contexts that match the pattern (for example, `.mycontext1` and `.mycontext2`). - A singleton list containing a subject prefix within a specific context, such as `[":.mycontext:mysubjects-*"]`, to export subjects that match the prefix within that context. `subject-format` : This is an optional parameter you can use to specify a format for the subject name in the destination cluster. You can specify `${subject}` as a placeholder, which will be replaced with the default subject name. For example, `dc_${subject}` for the subject `orders` will map to the destination subject name `dc_orders`. `context-type` : One of:
- **AUTO** - Prepends the source context with an automatically generated context, which is `.lsrc-xxxxxx` for Confluent Cloud. This is the default. - CUSTOM - Prepends the source context with a custom context name, specified in `context` below. - NONE - Copies the source context as-is, without prepending anything. This is useful to make an exact copy of the source Schema Registry in the destination. - DEFAULT - Replaces the source context with the default context. This is useful for copying schemas to the default context in the destination. (**Note:** DEFAULT is available on Confluent Cloud as of July 2023, and on Confluent Platform starting with version 7.4.2.) - REPLACE - Replaces the source context with the context name specified in `context` below. REPLACE drops the source context. All exported subjects land in the destination under the replacement context, regardless of which context they came from in the source.
For example, if the source has subjects `:.:orders-value` and `:.nondefault:products-value` and you set `context-type: REPLACE` with `context-name: .archive`, they land in the destination as `:.archive:orders-value` and `:.archive:products-value`.
Using the same source subjects, the other context types map them as follows:
- With `context-type: AUTO`, they land as `:.lsrc-xxxxxx:orders-value` and `:.lsrc-xxxxxx.nondefault:products-value`. AUTO prepends the generated context and keeps the source context for non-default subjects. - With `context-type: CUSTOM` and `context-name: .archive`, they land as `:.archive:orders-value` and `:.archive.nondefault:products-value`. CUSTOM prepends the custom context and keeps the source context for non-default subjects. - With `context-type: NONE`, they land unchanged as `:.:orders-value` and `:.nondefault:products-value`. `context-name` : A context name to be used with the CUSTOM or REPLACE `contextType` above. `config` : A set of configurations for creating a client to talk to the destination Schema Registry, which can be passed in a config file (for example, `--config-file ~/.txt`). Typically, this includes:
- `schema.registry.url` - The URL of the destination Schema Registry. The `schema.registry.url` must contain a valid, fully-qualified domain name and not just an IP address. For example, `https://my-registry.example.com` would meet this requirement but `https://192.168.1.1` would not. - `basic.auth.credentials.source` - Typically “USER_INFO” - `basic.auth.user.info` - Typically of the form `:` ### System topics and security configurations The following configurations for system topics are available: - `exporter.config.topic` - Stores configurations for the exporters. The default name for this topic is `_exporter_configs`, and its default/required configuration is: `numPartitions=1`, `replicationFactor=3`, and `cleanup.policy=compact`. - `exporter.state.topic` - Stores the status of the exporters. The default name for this topic is `_exporter_states`, and its default/required configuration is: `numPartitions=1`, `replicationFactor=3`, and `cleanup.policy=compact`. If you are using role-based access control (RBAC), `exporter.config.topic` and `exporter.state.topic` require `ResourceOwner` on these topics, as does the `_schemas` internal topic. See also, [Use Role-Based Access Control (RBAC) in Confluent Cloud](https://docs.confluent.io/cloud/current/access-management/access-control/cloud-rbac.html#) and [Configuring Role-Based Access Control for Schema Registry on Confluent Platform](https://docs.confluent.io/platform/current/schema-registry/security/rbac-schema-registry.html). If you are configuring Schema Registry on Confluent Platform using the [Schema Registry Security Plugin](https://docs.confluent.io/platform/current/confluent-security-plugins/schema-registry/install.html), you must activate both the exporter and the [Schema Registry security plugin](https://docs.confluent.io/platform/current/confluent-security-plugins/schema-registry/install.html#activate-the-plugins) by specifying both extension classes in the `$CONFLUENT_HOME/etc/schema-registry/schema-registry.properties` files: ```bash resource.extension.class=io.confluent.kafka.schemaregistry.security.SchemaRegistrySecurityResourceExtension,io.confluent.schema.exporter.SchemaExporterResourceExtension ``` The configuration for the exporter resource extension class in the `schema-registry.properties` is described in [Set up source and destination environments](https://docs.confluent.io/platform/current/schema-registry/schema-linking-cp.html#set-up-source-and-destination-environments) in Schema Linking on Confluent Platform. ### Lifecycle and states Schema Registry stores schemas in a Kafka topic. A schema exporter uses the topic offset to determine its progress. When a schema exporter is created, it begins in the STARTING state. While in this state, it finds and exports all applicable schemas already written to the topic. After exporting previously registered schemas, the exporter then enters the RUNNING state, during which it will be notified of any new schemas, which it can export if applicable. As schemas are exported, the exporter will save its progress by recording the latest topic offset. If you want to make changes to the schema exporter, you must first “pause” it, which causes it to enter the PAUSED state. You can pause an exporter at any point in its lifecycle. The exporter can then be resumed after the proper changes are made. Upon resuming, the exporter will find and export any applicable schemas since the last offset that it recorded. While an exporter is paused, it can also be “reset”, which will cause it to clear its saved offset and re-export all applicable schemas when it resumes. To accomplish this, the exporter starts off again in STARTING state after a reset, and follows the same lifecycle. To update an exporter’s configuration, such as adding subjects to its subject list, pause the exporter, apply the update, and reset it before resuming. An exporter tracks its progress as an offset and exports only schemas registered after that offset. Schemas already registered for the newly added subjects fall before that offset. Without a reset, the exporter skips them and exports only schemas registered after it resumes. A reset clears the recorded offset. The exporter then re-scans from the start and exports all matching schemas, including the existing ones for the newly added subjects. The states of a schema exporter at various stages in its lifecycle are summarized below. | State | Description | |----------|----------------------------------------------------------------------------------------------------------------------------------------------------------------| | STARTING | The exporter finds and exports all applicable previously registered schemas for the topic. This is the starting state, or the state after a reset. | | RUNNING | The exporter is notified of new schemas, exports them if applicable, and tracks progress by recording last topic offset. | | PAUSED | An exporter can be paused; for example, to make configuration changes. When it resumes, the exporter finds and exports schemas since the last recorded offset. | ### REST APIs Schema Registry supports the following REST APIs, as fully detailed in [Exporters](https://docs.confluent.io/platform/current/schema-registry/develop/api.html#schemaregistry-api-exporters) in the [Schema Registry API documentation](https://docs.confluent.io/platform/current/schema-registry/develop/api.html): | Task | API | |-----------------------------------------|------------------------------| | Gets a list of exporters for a tenant | GET /exporters | | Creates a new exporter | POST /exporters | | Gets info about an exporter | GET /exporters/{name} | | Gets the config for an exporter | GET /exporters/{name}/config | | Gets the status of an exporter | GET /exporters/{name}/status | | Updates the information for an exporter | PUT /exporters/{name}/config | | Pauses an exporter | PUT /exporters/{name}/pause | | Resumes an exporter | PUT /exporters/{name}/resume | | Resets an exporter, clears offsets | PUT /exporters/{name}/reset | | Deletes an exporter | DELETE /exporters/{name} | ## Deployment strategies and Schema Linking Schema Linking can replicate schemas between Schema Registry clusters as follows: A schema link sends data from a “source cluster” to a “destination cluster”. The supported cluster types are shown in the table below. | Source Schema Registry Cluster Options | Destination Schema Registry Cluster Options | |------------------------------------------|--------------------------------------------------------------------------------| | Confluent Cloud with internet networking | Confluent Cloud with internet networking | | Confluent Cloud with internet networking | Confluent Platform 7.0+ with an IP address accessible over the public internet | | Confluent Platform 7.0+ | Confluent Platform 7.0+ | | Confluent Platform 7.0+ | Confluent Cloud with internet networking | Schema Linking can also be used in both directions between two clusters if each direction links to a different context, allowing each side to continue to receive both reads and writes for schemas. With regard to Confluent Cloud and Confluent Platform solutions, you would use Schema Linking with Cluster Linking to mirror from one instance to the other. Any use of Confluent Platform in these setups require Confluent Platform 7.0.+ or later. To learn more about Cluster Linking and mirror topics, see [Cluster Linking for Confluent Platform](/platform/current/multi-dc-deployments/cluster-linking/index.html) and [Geo-replication with Cluster Linking on Confluent Cloud](/cloud/current/multi-cloud/cluster-linking/index.html). ## Monitor a schema exporter This section details the key metrics available for monitoring Confluent Cloud schema exporters. Each metric listed includes a description, usage and a sample query for retrieving the metric. Note that the [Metrics API](../monitoring/metrics-api.md#metrics-api) retrieves data in one minute intervals. This means that you could see duplicate values, older metrics, or an empty response body depending on the frequency of your queries. ### Exporter tasks This metric tracks the number of exporters currently in each operational state: `RUNNING`, `PAUSED`, `FAILED`. This metric also indicates whether an exporter is in an error state. - Metric name: `io.confluent.kafka.schema_registry/exporter_tasks` - Type: Gauge - Usage: Primary metric for assessing overall health of schema exporters. Group by `exporter_state` and `has_error` to build a dashboard highlighting failed exporters and error conditions for rapid troubleshooting. This example queries for the count of exporter tasks by exporter name, state, and error status: ```json { "aggregations": [ { "metric": "io.confluent.kafka.schema_registry/exporter_tasks" } ], "filter": { "op": "AND", "filters": [ { "field": "resource.schema_registry.id", "op": "EQ", "value": "{{schema_registry_id}}" }, { "field": "metric.exporter_name", "op": "EQ", "value": "{{exporter_name}}" } ] }, "granularity": "PT1M", "intervals": [ "now-6m/now-5m" ], "group_by": [ "metric.exporter_name", "metric.exporter_state", "metric.has_error" ] } ``` ### Exporter starting progress This metric tracks the progress of an exporter’s initial batch export on a scale from zero to 100 (0-100). This metric is specifically for monitoring the `STARTING` task that covers the first-time transfer of existing schemas. - Metric name: `io.confluent.kafka.schema_registry/exporter_starting_progress` - Type: Gauge - Usage: Confirm successful initialization of a new schema exporter. A value of 100 indicates the initial transfer is complete. Prolonged values below 100 may signal setup or connectivity issues. The example queries for the starting progress for all exporters in a Schema Registry instance: ```json { "aggregations": [ { "metric": "io.confluent.kafka.schema_registry/exporter_starting_progress" } ], "filter": { "op": "AND", "filters": [ { "field": "resource.schema_registry.id", "op": "EQ", "value": "{{schema_registry_id}}" } ] }, "granularity": "PT1M", "intervals": [ "now-6m/now-5m" ], "group_by": [ "metric.exporter_name" ] } ``` ### Schema transfer success This metric counts schemas successfully transferred from the source to the destination registry. This metric is intended for monitoring transfers that occur after the initial `STARTING` state. - Metric name: `io.confluent.kafka.schema_registry/schema_transfer_success_total` - Type: Counter - Usage: Tracks real-time replication activity. An increasing value indicates successful transfer of new or updated schemas. The metric resets to zero after a batch transfer completes, which is useful for observing the frequency and completion of synchronization cycles. This example queries for successful schema transfers for a specific exporter: ```json { "aggregations": [ { "metric": "io.confluent.kafka.schema_registry/schema_transfer_success_total" } ], "filter": { "op": "AND", "filters": [ { "field": "resource.schema_registry.id", "op": "EQ", "value": "{{schema_registry_id}}" }, { "field": "metric.exporter_name", "op": "EQ", "value": "{{exporter_name}}" } ] }, "granularity": "PT1M", "intervals": [ "now-6m/now-5m" ], "group_by": [ "metric.exporter_name" ] } ``` ## Schema Linking on private networks You can use the private endpoints for a Schema Registry when creating schema exporters on Confluent Platform or Confluent Cloud Schema Registry. You can use this Confluent CLI command to retrieve the endpoint, [confluent schema-registry cluster describe](https://docs.confluent.io/confluent-cli/current/command-reference/schema-registry/cluster/confluent_schema-registry_cluster_describe.html). To learn more, see [Use Schema Linking with private endpoints](/cloud/current/sr/fundamentals/sr-private-link.html#schema-linking-private-endpoints). ## Link schemas from Confluent Platform to Confluent Cloud You can link schemas from Confluent Platform to Confluent Cloud clusters by using a proxy. Define the proxy settings in the exporter configuration, setting both the proxy host and port. ### Example proxy setup in the exporter configuration file Typically, you’ll create an exporter configuration file at the time you create the exporter. The file is typically located at `$CONFLUENT_HOME/etc/schema-registry/exporter-config.json`, and contains the following fields: - The name of the exporter - The subjects to export - The context type - The context name - The proxy settings For a Confluent Platform to Confluent Cloud exporter, configure proxy settings as follows. ```json "config": { "schema.registry.url": "${CC_SR_ENDPOINT}", "basic.auth.credentials.source": "USER_INFO", "basic.auth.user.info": "${SR_API_KEY}:${SR_API_SECRET}", "proxy.host": "{HTTP_OR_HTTPS}://{PROXY_HOST_FQDN}", "proxy.port": "{PORT}" } ``` ### Test connectivity Verify that the Confluent Platform Schema Registry can communicate with the Confluent Cloud Schema Registry through the proxy. You can use `curl` commands to test connectivity to the proxy and the endpoint, as shown in this example: ```bash curl -u ${SR_API_KEY}:${SR_API_SECRET} -k 'https://{PSRC}.{REGION}.{PROVIDER}.confluent.cloud/schemas/ids/{SCHEMA_ID}' --proxy {HTTP_OR_HTTPS}://{PROXY_HOST_FQDN}:{PORT} ``` ### Error handling If the exporter state returns as PAUSED or fails with connection errors, check for the following. - DNS resolution issues for the proxy host. - Ensure the proxy allows bidirectional communication. - Review logs for specific error messages that can guide troubleshooting. ### Where to learn more To learn more about this use case, see the following topics: - [Configure a proxy](/cloud/current/networking/ccloud-console-access.html#configure-a-proxy) - [Deployment strategies and Schema Linking](#schema-linking-deployment-strategies) ## Manage Schema Linking in Disaster Recovery Failover Scenarios Confluent provides disaster recovery (DR) failover strategies with Cluster Linking and [mirror topics](../multi-cloud/cluster-linking/mirror-topics-cc.md#mirror-topics-concepts). Failover can be achieved by using commands to [Reverse a source and mirror topic](../multi-cloud/cluster-linking/mirror-topics-cc.md#reverse-source-and-mirror-topic), [Convert a mirror topic to a normal topic](../multi-cloud/cluster-linking/mirror-topics-cc.md#convert-mirror-topic-to-normal-topic), or a combination of both. DR failover is available on [Confluent Cloud](/cloud/current/multi-cloud/cluster-linking/index.html#use-cases), [Confluent Platform](/platform/current/multi-dc-deployments/cluster-linking/index.html#disaster-recovery), and [hybrid deployments](/platform/current/multi-dc-deployments/cluster-linking/hybrid-cp.html); a full tutorial is provided in the Confluent Cloud documentation at [Cluster Linking Disaster Recovery and Failover on Confluent Cloud](/cloud/current/multi-cloud/cluster-linking/dr-failover.html), as the concepts and workflows are applicable to both Confluent Cloud and Confluent Platform. Also, you can refer to the white paper [Best Practices for Multi-Region Apache Kafka Disaster Recovery in the Cloud (Active/Passive)](https://assets.confluent.io/m/5be60b2223a8801e/original/20240703-WP-Disaster_Recovery_in_the_Cloud.pdf), which that covers similar information. ### Schemas and mirror topics Cluster Linking preserves the schema IDs stored in messages, allowing Schema Registry and Schema Linking to work seamlessly with mirror topics. To consume from a mirror topic that is using schemas, the consumer clients must use a Schema Registry context with the same schema IDs as on the Schema Registry context used by the producers to the source topic. This setup is described in further detail in [How schemas work with mirror topics](../multi-cloud/cluster-linking/mirror-topics-cc.md#schemas-and-mirror-topics). ### Configure a Schema Registry failover strategy Given one Schema Registry per environment in a single region, you can set up Schema Registry for DR failover and recovery with Schema Linking, concurrent with your Cluster Linking strategies. To prepare for failover, you must replicate schemas to the DR region’s Schema Registry. This allows the Kafka applications, connectors, and stream processing applications to continue reading and writing schema-based data formats even during a disaster scenario. As a best practice, place each Schema Registry in the same region as the Kafka cluster it serves. In steady state, Schema Linking requires the primary Schema Registry to be in `READWRITE` mode and the secondary Schema Registry to be in `IMPORT` mode, which allows new schemas to be written only by Schema Linking. This creates an Active/Passive topology for Schema Linking: - an active Schema Registry on one side that can accept new schemas - a passive Schema Registry on the other side being synced by means of Schema Linking for DR. ![image](sr/images/sr-dr.png) ### Reverse the schema link as a failover response In the event of a failover, you must reverse the schema link. This will be necessary in any DR scenario to prevent data loss around schemas during the outage: - Fail forward (the secondary cluster becomes the primary cluster, the primary becomes the secondary). - Fail back to the original primary region/cluster. Note that the secondary cluster is sometimes referred to as the “DR”, “standby”, or “destination” cluster. In this section, it is referred to as “secondary”. For Cluster Linking, a “secondary” would typically hosts mirror topics based on primary topics (in some Scenarios both clusters mirror). For Schema Linking, the secondary cluster imports schemas and metadata from the primary as a backup in case of DR. The primary Schema Registry must be caught up with changes that occurred during the outage. Once the primary Schema Registry is back up, first do the following. 1. Pause the original schema link (which linked from primary to secondary). 2. Put the original primary Schema Registry into `IMPORT` mode. If there were new or modified schemas during the outage, copy over all schemas from the DR (secondary) to the primary. 3. Wait until all schemas registered during the outage are migrated back to the primary (by means of the IMPORT). At this point, you will follow one or the other set of steps below, depending on whether you are failing forward (original primary and secondary switch places) or failing forward, where the original primary is once again the primary and original secondary is once again the secondary. #### Failing forward If you are failing forward: 1. Create a schema link from the original secondary Schema Registry to the original primary Schema Registry. 2. Put the original secondary (now primary) into `READWRITE` mode. (The original primary is already appropriately in `IMPORT` mode, as it is now the secondary). #### Failing back If you are failing back, you must reactivate the original primary Schema Registry. To do this: 1. Stop the active schema link. 2. Set the original primary Schema Registry to `READWRITE` mode. 3. Set the original secondary Schema Registry to `IMPORT` mode. 4. Resume the original schema link from primary to secondary. ## Using Client Side Field Level Encryption Client-side field level encryption (CSFLE) uses a technique called [envelope encryption](../_glossary.md#term-envelope-encryption), where a [key encryption key (KEK)](../_glossary.md#term-key-encryption-key-KEK) is used to encrypt [data encryption keys (DEKs)](../_glossary.md#term-data-encryption-key-DEK). [Schema Registry](../_glossary.md#term-Schema-Registry) exposes a subcomponent called the DEK Registry, which provides APIs for managing KEKs and DEKs. The key from the KMS is registered as a KEK to the DEK Registry, and the DEK Registry will also hold the encrypted DEKs used for CSFLE. DEKs are scoped by schema subject. When CSFLE is configured, Schema Linking copies the DEKs, pointers to the KEKs, along with rulesets and metadata, from one Schema Registry cluster to another. Confluent never has access to the KEK itself. The DEK Registry holds only a reference to it, such as an Amazon Resource Name (ARN). The key vault uses the KEK to encrypt and decrypt the DEKs. Confluent supports several different key management services (KMS) and key vaults, including AWS KMS. To learn more about CSFLE on Confluent Cloud and Confluent Platform, see the following topics: - [CSFLE on Confluent Cloud](/cloud/current/security/encrypt/csfle/client-side.html) - [CSFLE on Confluent Platform](/platform/current/security/protect-data/csfle/client-side.html) ### CSFLE considerations for Schema Linking exporters Schema Linking exporters that copy CSFLE-protected subjects have additional RBAC and troubleshooting considerations beyond plain subject replication. #### RBAC for Schema Linking exporters with CSFLE Exporters that copy CSFLE-protected subjects require RBAC permissions on both the subjects and the KEKs that those subjects reference. Subject-level roles alone are not enough. When an exporter encounters a DEK whose KEK does not yet exist on the destination, it calls the destination DEK Registry to register the KEK. Without KEK-scoped permissions on the destination, that call fails. On the destination Schema Registry cluster, the exporter principal needs: - `DeveloperRead` and `DeveloperWrite` (or `ResourceOwner`) on the target subjects, scoped to `Subject:` or a prefix. - `DeveloperWrite` (or `ResourceOwner`) on each KEK referenced by those subjects, scoped to `Kek:`. The exporter principal is the owner of the destination Schema Registry API key configured on the exporter, or the identity pool or workload identity used when the exporter authenticates with OAuth. To find the owner of an existing Confluent Cloud Schema Registry API key, run: ```bash confluent api-key describe ${DEST_SR_API_KEY} --environment ${DEST_ENV_ID} ``` The `Owner` field in the response is the service account that needs the KEK role bindings. Create the KEK binding with: ```bash confluent iam rbac role-binding create \ --principal User:${EXPORTER_SA} \ --role DeveloperWrite \ --environment ${DEST_ENV_ID} \ --schema-registry-cluster ${DEST_SR_ID} \ --resource "Kek:${KEK_NAME}" ``` For the full set of KEK-scoped roles and how they apply to CSFLE operations, see [Access control (RBAC) for CSFLE](/cloud/current/security/encrypt/csfle/client-side.html#access-control-rbac-for-csfle). #### Troubleshooting CSFLE-related exporter failures An exporter that copies CSFLE-protected subjects can enter the `PAUSED` state with the following error: ```text User is denied operation Register on Kek: ; error code: 40301 ``` This error means the exporter principal lacks `DeveloperWrite` on the named KEK in the destination Schema Registry. Diagnose the failure by following these steps: 1. Identify the principal the exporter authenticates as on the destination (the Schema Registry API key owner, or the OAuth identity pool). 2. List the principal’s current role bindings on the destination Schema Registry cluster: ```bash confluent iam rbac role-binding list \ --principal User:${EXPORTER_SA} \ --environment ${DEST_ENV_ID} \ --schema-registry-cluster ${DEST_SR_ID} ``` 3. Confirm the output includes a binding with `Resource Type: Kek`, `Name` matching the failing KEK (or `*`), and `Role` set to `DeveloperWrite` or `ResourceOwner`. To fix the failure, grant `DeveloperWrite` on the affected KEK as shown above, then resume the exporter: ```bash confluent schema-registry exporter resume ${EXPORTER_NAME} \ --schema-registry-endpoint ${SOURCE_SR_ENDPOINT} ``` ## [Role-Based Access Control (RBAC)](../security/access-control/rbac/overview.md#cloud-rbac) for Schema Linking [Role-Based Access Control (RBAC)](../security/access-control/rbac/overview.md#cloud-rbac) enables administrators to set up and manage user access to Schema Linking. This allows for multiple users to collaborate on with different access levels to various resources. The following table shows how [RBAC roles](../security/access-control/rbac/predefined-rbac-roles.md#cloud-rbac-roles) map to Schema Linking resources. For details on how to manage RBAC for these resources, see [List the role bindings for a principal](../security/access-control/rbac/manage-role-bindings.md#manage-rbac-using-console), [Predefined RBAC Roles in Confluent Cloud](../security/access-control/rbac/predefined-rbac-roles.md#cloud-rbac-roles), and [List the role bindings for a principal](../security/access-control/rbac/manage-role-bindings.md#cloud-rbac-cli). | Role | Scope | All Schema Linking resources | |------------------------------------------------------------------------------------------------------|------------------------------------|--------------------------------------------------------------------------------------------| | [OrganizationAdmin](../security/access-control/rbac/predefined-rbac-roles.md#organizationadmin-role) | Organization | ✔ | | [EnvironmentAdmin](../security/access-control/rbac/predefined-rbac-roles.md#environmentadmin-role) | Environment | ✔ | | [CloudClusterAdmin](../security/access-control/rbac/predefined-rbac-roles.md#cloudclusteradmin-role) | Cluster | | | [Operator](../security/access-control/rbac/predefined-rbac-roles.md#operator-role) | Organization, Environment, Cluster | | | [MetricsViewer](../security/access-control/rbac/predefined-rbac-roles.md#metricsviewer-role) | Organization, Environment, Cluster | | | [ResourceOwner](../security/access-control/rbac/predefined-rbac-roles.md#resourceowner-role) | Schema Subject | ✔ On both the source and destination, for all subjects being exported. [1](#f1) | | [DeveloperManage](../security/access-control/rbac/predefined-rbac-roles.md#developermanage-role) | Schema Subject | | | [DeveloperRead](../security/access-control/rbac/predefined-rbac-roles.md#developerread-role) | Schema Subject | | | [DeveloperWrite](../security/access-control/rbac/predefined-rbac-roles.md#developerwrite-role) | Schema Subject | | | [DataDiscovery](../security/access-control/rbac/predefined-rbac-roles.md#datadiscovery-role) | Environment | | | [DataSteward](../security/access-control/rbac/predefined-rbac-roles.md#datasteward-role) | Environment | ✔ | Table Legend: - ✔ = Yes - Blank space = No ### Footnotes * **[1]** For the destination role bindings, if you are setting a context type of AUTO (default) or CUSTOM, then make sure you have prefixed “ResourceOwner” on the destination context. For more details, see [Permissions on schema contexts](#permissions-on-schema-contexts). ### How permission changes affect the destination Schema Registry Keep these permission behaviors in mind when running exporters: - If schema exporters are running, removing subject permissions from the source does not block the same account from accessing those subjects on the destination — provided the destination Schema Registry is a separate registry from the source. As a precaution, also remove those subject permissions on the destination Schema Registry. - The schema exporter stops running if permissions are removed from the destination Schema Registry for the account that created the exporter. ### Permissions on schema contexts If you want to grant permissions to specific schema contexts, you can do so using the Prefix rule and grant permissions with prefix as `:.schema_context:*`. The permission can be applied to any of the [RBAC roles that use scoping](/platform/current/schema-registry/security/rbac-schema-registry.html#role-mappings-to-operations-for-subject-based-authorization). For example, topic apply READ permission (`DeveloperRead`) for the context `.my_context`: 1. Use the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/command-reference/api-key/confluent_api-key_create.html) to [create a resource specific service account](../security/authenticate/workload-identities/service-accounts/overview.md#service-accounts) which will provide the specific permissions for the context. 2. On Confluent Cloud Console, select **Access and Accounts** under the dropdown menu on the sidebar. ![image](images/schema-link-cloud-ui-rbac.png) 3. Select the service account and add the access as shown below. Applying the prefix rule, as shown, will enable the DeveloperRead role for the service account only for context `.my_context`. ![image](images/schema-link-contexts-prefix-rule.png) ## Using OAuth for exporters Use [OAuth 2.0](https://oauth.net/2/) to authenticate Schema Linking exporters instead of long-lived API keys. OAuth provides token-based authentication and integrates with your organization’s identity management system (Okta, Microsoft Entra ID, Auth0, Google, and others). #### SEE ALSO For comprehensive OAuth setup guidance, see [Use OAuth/OIDC to Authenticate to Confluent Cloud](../security/authenticate/workload-identities/identity-providers/oauth/overview.md#oauth-overview). For Schema Registry client OAuth configuration, see [Configure Schema Registry Java clients](../security/authenticate/workload-identities/identity-providers/oauth/clients/java-clients.md#configure-sr-java-clients-for-oauth). ### When to use OAuth with exporters OAuth is recommended for exporters when: - You want to integrate with your organization’s identity provider (Microsoft Entra ID, Okta, Auth0, Google, etc.) - You need short-lived, automatically rotated credentials for enhanced security - You want to use existing enterprise identity and access management systems - You need fine-grained access control based on user attributes and group memberships - You prefer token-based authentication over static API keys ### Prerequisites Before configuring OAuth for exporters, ensure you have: 1. **Identity Provider Setup**: A configured OAuth/OIDC identity provider (Okta, Microsoft Entra ID, Auth0, Google, etc.) 2. **Confluent Cloud OAuth Configuration**: - Added your identity provider to Confluent Cloud ([Add an identity provider using Confluent Cloud Console](../security/authenticate/workload-identities/identity-providers/oauth/identity-providers.md#add-identity-provider)) - Created identity pools with appropriate policies ([Use OAuth Identity Pools with Your OAuth/OIDC Identity Provider on Confluent Cloud](../security/authenticate/workload-identities/identity-providers/oauth/identity-pools.md#add-oauth-identity-pools)) 3. **OAuth Client**: A client application registered with your identity provider with: - Client ID and client secret - Appropriate scopes for Schema Registry access - Client credentials grant type enabled ### Example OAuth exporter configuration Use this minimal OAuth exporter configuration as a starting point. The table that follows describes each property. ```json { "name": "test-exporter-cloud", "contextType": "CUSTOM", "context": "test-context", "subjects": ["bacon", "eggs"], "config": { "schema.registry.url": "https://psrc-abc123.us-west-2.aws.confluent.cloud", "bearer.auth.client.id": "0oa1abc123def456", "bearer.auth.credentials.source": "OAUTHBEARER", "bearer.auth.client.secret": "your-client-secret-here", "bearer.auth.issuer.endpoint.url": "https://dev-12345.okta.com/oauth2/aus12345/v1/token", "bearer.auth.credentials.source": "OAUTHBEARER", "bearer.auth.scope": "groups", "bearer.auth.identity.pool.id": "pool-wxyz123", "bearer.auth.logical.cluster": "lsrc-1234abc" } } ``` The following table describes each configuration property used in the example above: | Property | Required | Description | |-----------------------------------|-------------|-------------------------------------------------------------| | `name` | Yes | Name of the exporter | | `contextType` | No | Context type for the exporter (`CUSTOM`, `AUTO`, or `NONE`) | | `context` | Conditional | Context name when contextType is CUSTOM | | `subjects` | No | Array of subject names to export. Default: all subjects | | `schema.registry.url` | Yes | Destination Schema Registry URL endpoint | | `bearer.auth.client.id` | Yes | OAuth client ID from your identity provider | | `bearer.auth.credentials.source` | Yes | Set to `OAUTHBEARER` for OAuth authentication | | `bearer.auth.client.secret` | Yes | OAuth client secret from your identity provider | | `bearer.auth.issuer.endpoint.url` | Yes | OAuth token endpoint URL from your identity provider | | `bearer.auth.scope` | No | OAuth scopes for the token request (e.g., `groups`) | | `bearer.auth.identity.pool.id` | Conditional | Identity pool ID. Required if not using auto pool mapping | | `bearer.auth.logical.cluster` | Yes | Schema Registry cluster ID (e.g., `lsrc-1234abc`) | ### OAuth configuration parameters The following OAuth configuration parameters are supported for Schema Registry exporters: | Parameter | Required | Description | |-------------------------------------------|-------------|-----------------------------------------------------------------------------| | `bearer.auth.credentials.source` | Yes | Set to `OAUTHBEARER` for OAuth authentication | | `bearer.auth.issuer.endpoint.url` | Yes | OAuth token endpoint URL from your identity provider | | `bearer.auth.client.id` | Yes | OAuth client ID from your identity provider | | `bearer.auth.client.secret` | Yes | OAuth client secret from your identity provider | | `bearer.auth.logical.cluster` | Yes | Schema Registry cluster ID (e.g., `lsrc-1234abc`) | | `bearer.auth.identity.pool.id` | Conditional | Identity pool ID. Required if not using auto pool mapping | | `bearer.auth.scope` | No | OAuth scopes for the token request (e.g., `groups`, `schema:read`) | | `bearer.auth.sub.claim.name` | No | Name of the JWT claim containing the subject. Default: `sub` | | `bearer.auth.scope.claim.name` | No | Name of the JWT claim containing the scope. Default: `scope` | | `bearer.auth.cache.expiry.buffer.seconds` | No | Buffer time before token expiration for refresh. Default: `300` (5 minutes) | ### Configuration examples The following examples show OAuth configuration for common identity providers and pool-mapping scenarios. #### Example 1: OAuth with Okta Create a configuration file (`oauth-config.txt`) for your exporter: ```properties schema.registry.url=https://psrc-abc123.us-west-2.aws.confluent.cloud bearer.auth.credentials.source=OAUTHBEARER bearer.auth.issuer.endpoint.url=https://dev-12345.okta.com/oauth2/aus12345/v1/token bearer.auth.client.id=0oa1abc123def456 bearer.auth.client.secret=your-client-secret-here bearer.auth.scope=groups bearer.auth.logical.cluster=lsrc-1234abc bearer.auth.identity.pool.id=pool-wxyz123 ``` Create the exporter using the CLI: ```bash confluent schema-registry exporter create oauth-exporter \ --subjects ":*:" \ --config oauth-config.txt ``` #### Example 2: OAuth with Microsoft Entra ID (Azure AD) ```properties schema.registry.url=https://psrc-def456.us-east-1.aws.confluent.cloud bearer.auth.credentials.source=OAUTHBEARER bearer.auth.issuer.endpoint.url=https://login.microsoftonline.com/your-tenant-id/oauth2/v2.0/token bearer.auth.client.id=your-azure-app-id bearer.auth.client.secret=your-azure-client-secret bearer.auth.scope=https://graph.microsoft.com/.default bearer.auth.logical.cluster=lsrc-5678def bearer.auth.identity.pool.id=pool-azure789 ``` #### Example 3: OAuth with auto pool mapping When using auto pool mapping, you can omit the `bearer.auth.identity.pool.id` parameter: ```properties schema.registry.url=https://psrc-ghi789.us-central1.gcp.confluent.cloud bearer.auth.credentials.source=OAUTHBEARER bearer.auth.issuer.endpoint.url=https://auth.example.com/oauth2/token bearer.auth.client.id=kafka-exporter-client bearer.auth.client.secret=super-secret-value bearer.auth.scope=schema-registry bearer.auth.logical.cluster=lsrc-automap ``` #### Example 4: OAuth with custom claims ```properties schema.registry.url=https://psrc-jkl012.eu-west-1.aws.confluent.cloud bearer.auth.credentials.source=OAUTHBEARER bearer.auth.issuer.endpoint.url=https://custom-idp.example.com/oauth2/token bearer.auth.client.id=schema-exporter-001 bearer.auth.client.secret=custom-secret-123 bearer.auth.scope=custom-scope bearer.auth.logical.cluster=lsrc-custom789 bearer.auth.identity.pool.id=pool-custom456 bearer.auth.sub.claim.name=userId bearer.auth.scope.claim.name=permissions bearer.auth.cache.expiry.buffer.seconds=600 ``` ### CLI commands for OAuth exporters The following [commands](https://docs.confluent.io/confluent-cli/current/command-reference/overview.html) are available for use with Schema Registry and exporters: #### Create an OAuth-enabled exporter ```bash # Create exporter with OAuth configuration confluent schema-registry exporter create my-oauth-exporter \ --subjects ":*:" \ --config oauth-config.txt ``` #### Create exporter for specific contexts with OAuth ```bash # Export only schemas from a specific context confluent schema-registry exporter create context-oauth-exporter \ --subjects "orders,customers" \ --context-type CUSTOM \ --context-name production \ --config oauth-config.txt ``` #### Update exporter OAuth configuration ```bash # Update OAuth configuration confluent schema-registry exporter update my-oauth-exporter \ --config updated-oauth-config.txt ``` #### Check exporter status with OAuth ```bash # Verify OAuth exporter is running confluent schema-registry exporter status describe my-oauth-exporter ``` ### JSON API configuration format For programmatic creation using the Schema Registry REST API, use the following JSON format: ```json { "name": "api-oauth-exporter", "contextType": "CUSTOM", "context": "production", "subjects": ["orders", "customers", "products"], "config": { "schema.registry.url": "https://psrc-api123.us-west-2.aws.confluent.cloud", "bearer.auth.credentials.source": "OAUTHBEARER", "bearer.auth.issuer.endpoint.url": "https://auth.company.com/oauth2/token", "bearer.auth.client.id": "schema-exporter-client", "bearer.auth.client.secret": "secure-client-secret", "bearer.auth.scope": "schema:read schema:write", "bearer.auth.logical.cluster": "lsrc-api456", "bearer.auth.identity.pool.id": "pool-api789" } } ``` ### Security considerations When using OAuth with exporters, follow these security best practices: #### Client Secret Management - Store client secrets securely using environment variables or secret management systems - Never commit secrets to version control - Rotate client secrets regularly according to your organization’s security policies - Use different client credentials for different environments (dev, staging, production) #### Scope and Permissions - Use the principle of least privilege when defining OAuth scopes - Grant only the minimum Schema Registry permissions needed for the exporter’s function - Regularly audit and review identity pool policies and access permissions #### Token Security - Configure appropriate token expiration times (typically 1-24 hours) - Use the `bearer.auth.cache.expiry.buffer.seconds` parameter to ensure tokens are refreshed before expiration - Monitor token usage and authentication failures #### Example secure configuration ```bash # Use environment variables for sensitive values export OAUTH_CLIENT_ID="your-client-id" export OAUTH_CLIENT_SECRET="your-client-secret" export SR_CLUSTER_ID="lsrc-your-cluster" export IDENTITY_POOL_ID="pool-your-pool" ``` Create configuration file with environment variable references: ```properties schema.registry.url=https://psrc-secure.region.provider.confluent.cloud bearer.auth.credentials.source=OAUTHBEARER bearer.auth.issuer.endpoint.url=https://secure-idp.company.com/oauth2/token bearer.auth.client.id=${OAUTH_CLIENT_ID} bearer.auth.client.secret=${OAUTH_CLIENT_SECRET} bearer.auth.logical.cluster=${SR_CLUSTER_ID} bearer.auth.identity.pool.id=${IDENTITY_POOL_ID} bearer.auth.cache.expiry.buffer.seconds=300 ``` ### Validate and debug OAuth configuration Use these utilities to verify your OAuth setup and inspect exporter state. #### Configuration validation To test your OAuth configuration before creating an exporter: ```bash # Test OAuth configuration with a simple API call curl -X POST "https://your-idp.com/oauth2/token" \ -H "Content-Type: application/x-www-form-urlencoded" \ -d "grant_type=client_credentials&client_id=YOUR_CLIENT_ID&client_secret=YOUR_CLIENT_SECRET" ``` #### Debug exporter status ```bash # Get detailed exporter information confluent schema-registry exporter describe my-oauth-exporter # Check exporter logs for errors confluent schema-registry exporter status describe my-oauth-exporter ``` ### Troubleshooting OAuth exporters #### Authentication failures ```bash Error: Unable to authenticate with OAuth provider ``` - Verify client ID and secret are correct - Check that the OAuth issuer endpoint URL is accessible - Ensure the client has the required grants and scopes - Verify the identity provider is properly configured in Confluent Cloud #### Authorization failures ```bash Error: Access denied to Schema Registry resources ``` - Check identity pool configuration and policies - Verify the logical cluster ID matches your Schema Registry cluster - Ensure the service account has appropriate RBAC permissions - Review JWT token claims and pool filter expressions #### Token expiration issues ```bash Error: Token expired ``` - Increase `bearer.auth.cache.expiry.buffer.seconds` value - Check token lifetime settings in your identity provider - Verify network connectivity for token refresh requests ### OAuth-specific resources - [Use OAuth/OIDC to Authenticate to Confluent Cloud](../security/authenticate/workload-identities/identity-providers/oauth/overview.md#oauth-overview) - Complete OAuth setup guide for Confluent Cloud - [Configure Schema Registry Java clients](../security/authenticate/workload-identities/identity-providers/oauth/clients/java-clients.md#configure-sr-java-clients-for-oauth) - Schema Registry client OAuth configuration - [Add an identity provider using Confluent Cloud Console](../security/authenticate/workload-identities/identity-providers/oauth/identity-providers.md#add-identity-provider) - Adding identity providers to Confluent Cloud - [Use OAuth Identity Pools with Your OAuth/OIDC Identity Provider on Confluent Cloud](../security/authenticate/workload-identities/identity-providers/oauth/identity-pools.md#add-oauth-identity-pools) - Creating and managing identity pools - [Schema Registry client configuration](sr-client-configs.md#sr-client-configs) - All Schema Registry client configuration options ## Related content - Blog post: [Easily Copy or Migrate Schemas Anywhere with Schema Linking](https://www.confluent.io/blog/easy-data-migration-with-schema-linking/) - [Exporters](/platform/current/schema-registry/develop/api.html#schemaregistry-api-exporters) in the [Schema Registry API documentation](/platform/current/schema-registry/develop/api.html) - [Schema Linking on Confluent Platform](/platform/current/schema-registry/schema-linking-cp.html)