# Manage Schemas in Confluent Platform and Control Center Use the Schema Registry feature in Control Center to manage Confluent Platform topic schemas. You can: - create, edit, and view schemas - compare schema versions - download schemas The Schema Registry performs validations and compatibility checks on schemas. The Schema Registry feature in Control Center is [enabled](#ff-c3-edit-schema) by default. Disabling the feature disables both viewing and editing of schemas. ## Prerequisites If you do not already have an instance of Confluent Platform running, you can follow these workflows using the simple single-node developer test version of Confluent Platform along with the separately installed version of Control Center. 1. Make sure you have [Confluent Platform 8 or later](https://www.confluent.io/download/) installed on your local machine. To do this, see [Install Confluent Platform using ZIP and TAR Archives](../installation/installing_cp/zip-tar.md#prod-kafka-cli-install), and complete the first two steps, [Get the software](../installation/installing_cp/zip-tar.md#get-software) and [Configure CONFLUENT_HOME and PATH](../installation/installing_cp/zip-tar.md#configure-confluent-home). 2. Verify that the [Confluent CLI](https://docs.confluent.io/confluent-cli/current/overview.html) is installed and up-to-date by running the following commands. ```bash confluent --version ``` ```bash confluent update ``` 3. Follow the steps to download, extract, and configure Control Center to run a [Single-node manual installation](/control-center/current/installation/overview.html#single-node-manual-installation) using the [archives](/control-center/current/installation/overview.html#archive) (not Docker). Note that the Control Center installation will live and run in its own directory, $CONTROL_CENTER_HOME, at the same level as $CONFLUENT_HOME. This is different from the setup in previous versions of Confluent Platform, where Control Center was installed as a part of Confluent Platform and resided in a subdirectory under $CONFLUENT_HOME. 4. Start the Confluent Platform cluster by running the following command, preferably in its own terminal window. ```bash confluent local services start ``` 5. Configure the local environment variable $CONTROL_CENTER_HOME for Control Center as follows. - Change directories into your local Control Center install. - Type the following command to set the $CONTROL_CENTER_HOME environment variable. ```bash export CONTROL_CENTER_HOME=`pwd` ``` 6. Start Control Center by running the following command, preferably in its own terminal window. ```bash confluent local services control-center start ``` 7. Verify that Control Center is running by navigating to the following URL in your browser: [http://localhost:9021/](http://localhost:9021/) ## Create a schema for a topic in Control Center Create key and value schemas. Value schemas are typically created more frequently than a key schema. These best practices are recommended: - Provide default values for fields to facilitate [backward-compatibility](fundamentals/schema-evolution.md#avro-backward-compatibility) if pertinent to your schema. - Document at least the more obscure fields for human-readability of a schema. ### Create a topic If you don’t already have a topic, you can create one with either of the following methods. - Create a topic using the Confluent CLI: ```bash kafka-topics --create --topic my-transactions --bootstrap-server localhost:9092 ``` - Or, create a topic using the Control Center web interface: 1. Navigate to the **Topics** page in the Control Center web interface at [http://localhost:9021/](http://localhost:9021/). 2. Click **Add a topic**, and rename it from the default name to `my-transactions`. 3. Click **Create with defaults**. ### Create a topic value schema 1. Select a cluster. 2. Click **Topics** on the menu. 3. Select a topic.  The topic overview page is displayed.  4. Click the **Schema** tab. You are prompted to set a message value schema.  5. Click **Set a schema**. The Schema editor appears pre-populated with the basic structure of an [Avro schema](https://avro.apache.org/docs/1.8.1/spec.html#schemas) to use as a starting point, if desired.  6. For this example, take the default schema context. You can accept the default context, create a new context, or select an existing context other than the default. To learn more, see [Work with schema contexts](#work-with-schema-contexts-in-c3).  7. Select a schema format type: - Avro - JSON - Protobuf  Choose **Avro** if you want to try out the code examples provided in the next steps. 8. Enter the schema in the editor: - `name`: Enter a name for the schema if you do not want to accept the default, which is determined by the subject name strategy. The default is `schema_type_topic_name`. Required. - `type`: Either `record`, `enum`, `union`, `array`, `map`, or `fixed`. (The type `record` is specified at the schema’s top level and can include multiple fields of different data types.) Required. - `namespace`: Fully-qualified name to prevent schema naming conflicts. String that qualifies the schema `name`. Optional but recommended. - `fields`: JSON array listing one or more fields for a record. Required. Each field can have the following attributes: - `name`: Name of the field. Required. - `type`: Data type for the field. Required. - `doc`: Field metadata. Optional but recommended. - `default`: Default value for a field. Optional but recommended. - `order`: Sorting order for a field. Valid values are ascending, descending, or ignore. Default: Ascending. Optional. - `aliases`: Alternative names for a field. Optional. Copy and paste the following example schema. ```JSON { "type": "record", "namespace": "my.examples", "name": "Payment", "fields": [ { "name": "id", "type": "string" }, { "name": "amount", "type": "double" } ] } ```  9. In edit mode, you have options to: - Validate the schema for syntax and structure before you create it. - [Add schema references](#cp-sr-schema-references) with a guided wizard. - Choose a context for the schema. 10. Click **Create**. - If the entered schema is valid, the **Schema updated** message is briefly displayed in the banner area. - If the entered schema is not valid, an **Input schema is an invalid Avro schema** error is displayed in the banner area. If applicable, repeat the procedure as appropriate for the topic key schema. ### Work with schema contexts A schema context is a grouping of subject names and schema IDs. Contexts provide more flexibility with regard to subject naming, schema IDs, and how clients can reference schemas. #### Specify schema contexts You can either accept the default context or specify a custom context when you create a schema in Control Center. To create a new context under which to create a schema: 1. Create a topic or choose an existing topic. 2. Click the **Schema** tab. 3. With **Value** selected, click **Set a schema**. 4. Add the schema with the appropriate schema format selected. This example shows an Avro schema. 5. Next to the **Schema context** field, select the drop-down menu and choose **Define a new context**.  6. Enter a name for the new context in the dialog, then click **Create** on the confirmation dialog.  7. The schema is created under the new context.  #### Learn more about schema contexts Schema Registry provides the option to logically group schemas by specifying schema contexts. By default, schemas live in the default context. By providing qualified names for schemas, you group them into what are essentially sub-registries with context-specific paths. This gives you the ability to have multiple schemas with the same subject names and IDs existing as unique entities within their different contexts. There are several advantages to this, including the ability to provide specific contexts for different clients. Schema contexts are used extensively for [Schema Linking](schema-linking-cp.md#schema-linking-cp-overview), but can also be used independently of that feature as needed. There is no “global” compatibility across all contexts, only per context. To learn more, see [Schema Evolution and Compatibility for Schema Registry on Confluent Platform](fundamentals/schema-evolution.md#schema-evolution-and-compatibility). To learn more about schema contexts, see [Contexts](schema-linking-cp.md#schema-contexts) within Schema Linking. Also, the [Schema Linking Quick Start](schema-linking-cp.md#schema-linking-cp-quick-start) includes examples of working with both subjects in the default context (unqualified subjects) and named contexts (qualified subjects). ### Working with schema references You can add a reference to another schema, using the wizard to help locate available schemas and versions.  The **Reference name** you provide must match the target schema, based on guidelines for the schema format you are using: - In JSON Schema, the name is the value on the `$ref` field of the referenced schema - In Avro, the name is the value on the `type` field of the referenced schema - In Protobuf, the name is the value on the `Import` statement referenced schema First, locate the schema you want to reference, and get the reference name for it. **Add a schema reference to the current schema in the editor** 1. Click **Add reference**. 2. Provide a Reference name per the rules described above. 3. Select the schema fro the Subject list. 4. Select the Version of the schema you want to use. 5. Click **Validate** to check if the reference will pass. 6. Click **Save** to save the reference. For example, to reference the schema for the `my-transactions` topic (`my-transactions-value`) from the `widget` schema, you can configure a reference to type, `record` as shown.  To learn more, see [Schema references](fundamentals/serdes-develop/index.md#referenced-schemas) in the schema formats developer documentation. **View, edit, or delete schema references for a topic** Existing schema references show up on editable versions of the schema where they are configured. 1. Navigate to a topic; for example, the `widget-value` schema associated with the `widget` topic in the previous example. 2. Click into the editor as if to edit the schema. If there are references to other Schemas configured in this schema, they will display in the **Schema references** list below the editor. You can also add more references to this schema, modify existing, or delete references from this view. ### Create a topic key schema 1. Click the **Key** option. You are prompted to set a message key schema.  2. Click **Set a schema**. 3. Choose **Avro** format and/or delete the sample formatting and simply paste in a string UUID. 4. Enter the schema into the editor and click **Save**. Copy and paste the following example schema, and save it. ```JSON { "namespace": "io.confluent.examples.clients.basicavro", "name": "key_my_transactions", "type": "string" } ```  5. In edit mode, you have options to: - Validate the schema for syntax and structure before you create it. - [Add schema references](#cp-sr-schema-references) with a guided wizard. #### Best Practices and Pitfalls for Key Values Kafka messages are key-value pairs. Message keys and message values can be serialized independently. For example, the value may be using an [Avro](http://avro.apache.org/docs/current/gettingstartedjava.html#Defining+a+schema) `record`, while the key may be a primitive (`string`, `integer`, and so forth). Typically message keys, if used, are primitives. How you set the key is up to you and the requirements of your implementation. As a best practice, keep key value schema complexity to a minimum. Use either a simple, non-serialized data type such as a string UUID or long ID, or an Avro record that does not use maps or arrays as fields, as shown in the example below. Do not use Protobuf messages and JSON objects for key values. Avro does not guarantee deterministic serialization for maps or arrays, and Protobuf and JSON schema formats do not guarantee deterministic serialization for any object. Using these formats for key values will break topic partitioning. If you do decide to use a complex format for a key value schema, set `auto.register.schemas=false` to prevent registration of new valid and compatible schemas that, because of the complex key value format, will break your partitioning. To learn more, see [Auto Schema Registration](schema_registry_onprem_tutorial.md#auto-schema-registration) in the On-Premises Schema Registry Tutorial, and [Partitioning gotchas](https://forum.confluent.io/t/partitioning-gotchas-dont-use-avro-json-or-protobuf-for-keys-and-be-aware-of-client-hashing-differences/2718) in the Confluent Community Forum. For detailed examples of key and value schemas, see the discussion under [Formats, Serializers, and Deserializers](/platform/current/schema-registry/fundamentals/serdes-develop/index.html). ## Viewing a schema in Control Center View the schema details for a specific topic. 1. Select a cluster from the navigation bar. 2. Click the **Topics** menu. The [Manage Topics Using Control Center for Confluent Platform](https://docs.confluent.io/control-center/current/topics/overview.html#c3-all-topics) appears. 3. Select a topic.  The topic overview page appears. 4. Click the **Schema** tab. The Value schema is displayed by default.  5. Click the **Key** tab to view the key schema if present. ## Editing a schema in Control Center Edit an existing schema for a topic. 1. Select a cluster from the navigation bar. 2. Click the **Topics** menu. The [Manage Topics Using Control Center for Confluent Platform](https://docs.confluent.io/control-center/current/topics/overview.html#c3-all-topics) appears. 3. Select a topic. 4. Click the **Schema** tab. 5. Select the **Value** or **Key** tab for the schema. 6. Click anywhere in the schema to enable edit mode and make changes in the schema editor. For example, if you are following along with the example: - Select the topic `my-transactions`, click **Schema**, and select the **Value** tab. - Edit the schema by copy-pasting the following definition for a new `region` field, after the `id` and `amount` fields. Precede your new definition with a comma, per the syntax. ```JSON { "name": "region", "type": "string", "default": "" } ``` Note that the new `region` field includes a default value, which makes it backward compatible. By plugging in the default value, consumers can use the new schema to read data submitted by producers that use the older schema (without the `region` field). 7. Click **Save**. - If the schema update is valid and compatible with its prior versions (assuming a backward-compatible mode), the schema is updated and the version count is incremented. You can [compare the different versions](#c3-schema-compare-versions) of a schema.  - If the schema update is invalid or incompatible with an earlier schema version, an error is displayed. The example below shows the addition of another new field, `country`, with no default provided for backward compatibility.  ## Comparing schema versions in Control Center Compare versions of a schema to view its evolutionary differences. 1. Select a cluster from the navigation bar. 2. Click the **Topics** menu. The [Manage Topics Using Control Center for Confluent Platform](https://docs.confluent.io/control-center/current/topics/overview.html#c3-all-topics) appears. 3. Select a topic. 4. Click the **Schema** tab. 5. Select the **Key** or **Value** tab for the schema. 6. Select **Version history** from the inline menu.  The current version number of the schema is indicated on the version menu.  7. Select the **Turn on version diff** check box. 8. Select the versions to compare from each version menu. The differences are highlighted for comparison.  ## Changing the compatibility mode of a schema in Control Center The default compatibility mode is **Backward**. The mode can be changed for the schema of any topic if necessary. 1. Select a cluster from the navigation bar. 2. Click the **Topics** menu. The [Manage Topics Using Control Center for Confluent Platform](https://docs.confluent.io/control-center/current/topics/overview.html#c3-all-topics) appears. 3. Select a topic. 4. Click the **Schema** tab. 5. Select the **Key** or **Value** tab for the schema. 6. Select **Compatibility setting** from the inline menu.  The Compatibility settings are displayed.  7. Select a mode option: - [Backward](fundamentals/schema-evolution.md#avro-backward-compatibility) (Confluent Schema Registry default) - [Transitive backward](fundamentals/schema-evolution.md#avro-backward-compatibility) - [Forward](fundamentals/schema-evolution.md#avro-forward-compatibility) - [Transitive forward](fundamentals/schema-evolution.md#avro-forward-compatibility) - [Full](fundamentals/schema-evolution.md#avro-full-compatibility) - [Transitive full](fundamentals/schema-evolution.md#avro-full-compatibility) - [None](fundamentals/schema-evolution.md#avro-none-compatibility) (not recommended) Descriptions indicate the compatibility behavior for each option. For more information, including the changes allowed for each option, see [Schema Evolution and Compatibility for Schema Registry on Confluent Platform](fundamentals/schema-evolution.md#schema-evolution-and-compatibility). 8. Click **Save**. ## Downloading a schema from Control Center 1. Select a cluster from the navigation bar. 2. Click the **Topics** menu. The [Manage Topics Using Control Center for Confluent Platform](https://docs.confluent.io/control-center/current/topics/overview.html#c3-all-topics) appears. 3. Select a topic. 4. Click the **Schema** tab. 5. Select the **Key** or **Value** tab for the schema. 6. Click **Download**. A schema JSON file for the topic is downloaded into your Downloads directory. Example filename: `schema-transactions-v1-Ry_XaOGvTxiZVZ5hbBhWRA.json` Example contents: ```bash {"subject":"transactions-value","version":1,"id":2,"schema":"{\"type\":\"record\",\"name\":\"Payment\", \"namespace\":\"io.confluent.examples.clients.basicavro\", \"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"}, {\"name\":\"region\",\"type\":\"string\"}]}"} ``` This is the first version of the schema, and it has an `id` of 2. The schema is escaped JSON. A backslash precedes double-quotes. ## Troubleshoot error “Schema Registry is not set up” If you get an error message on Control Center when you try to access a topic schema (”Schema Registry is not set up”), first make sure that Schema Registry is running. Then verify that the Schema Registry `listeners` configuration matches the Control Center `confluent.controlcenter.schema.registry.url` configuration. Also check the HTTPS configuration parameters.  For more information, see [A schema for message values has not been set for this topic](https://docs.confluent.io/control-center/current/installation/troubleshooting.html#c3-schema-registry-not-set-up), and start-up procedures for [Quick Start for Confluent Platform](../get-started/platform-quickstart.md#quickstart), or [Install Confluent Platform On-Premises](../installation/overview.md#installation), depending on which one of these you are using to run Confluent Platform. ## Enabling and disabling Schema Registry in Control Center The feature that allows working with schemas in Control Center is enabled by default. The feature can be disabled if an organization does not want any users to access the feature. After disabling the feature, the Topics Schema menu and the Schema tab are no longer visible in the Control Center UI. The ability to view and edit schemas is disabled. To disable the edit schema feature in Control Center: 1. Set the `confluent.controlcenter.schema.registry.enable` option in your `control-center.properties` file to `false`. ```bash confluent.controlcenter.schema.registry.enable=false ``` #### NOTE Make the change in the appropriate Control Center properties file or files configured for your environments, including `control-center-dev.properties` or `control-center-production.properties`. The properties files are located in `/path-to-confluent/etc/confluent-control-center/`. 2. Restart Control Center and pass in the properties file for the configuration to take effect: ```bash ./bin/control-center-stop ./bin/control-center-start ../etc/confluent-control-center/control-center.properties ``` To enable the feature again, set the option back to `true` and restart Control Center with the updated properties file. ## Enabling Multi-Cluster Schema Registry Confluent Platform supports the ability to run multiple schema registries and associate a unique Schema Registry to each Kafka cluster in multi-cluster environments. The ability to scale up schema registries in conjunction with Kafka clusters is useful for evolving businesses; and particularly supports data governance, organization, and management across departments in large enterprises. When multi-cluster Schema Registry is configured and running, you can create and manage schemas per topics in Control Center as usual. ### Configuration Properties and Files Multiple Schema Registry clusters may be specified with `confluent.controlcenter.schema.registry.{name}.url` in the appropriate Control Center properties file. To use a Schema Registry cluster identified in this way, add or verify the following broker and Control Center configurations. A new endpoint `/v1/metadata/schemaRegistryUrls` has been exposed by Kafka to return the `confluent.schema.registry.url` field from the Kafka broker configurations. Control Center uses this field to look up the registries from Kafka broker configurations. To use this, you must configure unique listener endpoints for each cluster: - In the broker `server.properties` files (unique for each Kafka cluster), specify the REST endpoint with the `confluent.http.server.listeners` field, which defaults to `http://0.0.0.0:8090`. - In the appropriate Control Center properties file, use `confluent.controlcenter.streams.cprest.url` to define the REST endpoint for `controlcenter.cluster`. - For additional clusters, define REST endpoints using `confluent.controlcenter.kafka.{name}.cprest.url`. This should be consistent with the Kafka cluster name used for other Kafka Control Center configurations; for example, `confluent.controlcenter.kafka.{name}.bootstrap.servers`. A minimal viable configuration touches the following files, and includes settings for these properties (example names and ports are given): #### Control Center properties file The [Control Center Configuration Examples for Confluent Platform](https://docs.confluent.io/control-center/current/installation/properties.html#c3-properties-files) file includes: - `confluent.controlcenter.schema.registry.url=http://localhost:8081` - `confluent.controlcenter.schema.registry.sr-1.url=http://localhost:8082` - `confluent.controlcenter.streams.cprest.url=http://localhost:8090` - `confluent.controlcenter.kafka.AK1.cprest.url=http://localhost:8091` See [Control Center configuration reference](https://docs.confluent.io/control-center/current/installation/configuration.html#c3-schema-registry-url) for a full description of `confluent.controlcenter.schema.registry.url`. #### Broker configuration file for the Control Center cluster The Kafka broker configuration file for `controlcenter.cluster`, such as `server0.properties`, includes: - `confluent.http.server.listeners=http://localhost:8090` - `confluent.schema.registry.url=http://localhost:8081` #### Broker configuration file for the Kafka cluster The Kafka broker configuration file for `AK1`, `server1.properties`) includes: - `confluent.http.server.listeners=http://localhost:8091` - `confluent.schema.registry.url=http://localhost:8082` With these configurations, editing the schema through the Control Center UI will connect to `http://localhost:8081` for `controlcenter.cluster` and `http://localhost:8082` for `AK1`. #### Defaults and Fallback If `confluent.schema.registry.url` fields are not specified for any brokers, the `confluent.controlcenter.schema.registry.url` Schema Registry URL is applied. For example, if the Schema Registry URL was not provided for `AK1`, `AK1`’s associated Schema Registry cluster would also be specified at `http://localhost:8081`. If `confluent.controlcenter.schema.registry.url` is not explicitly specified in the Control Center properties file, it defaults to `http://localhost:8081`. ### Example Following is a detailed example of a functional multi-cluster Schema Registry setup with two Kafka clusters connected to Control Center, one the `controlcenter.cluster`, and the other named `AK1`, each with one broker. Example instructions refer to the location of your Confluent Platform installation as `$CONFLUENT_HOME`. To run the example, copy default configuration files to new files per the example filenames below, add/modify properties as shown, and run the components as described in [Run the Example](#run-example-multi-cluster-sr). #### IMPORTANT As of Confluent Platform 7.5, ZooKeeper is deprecated for new deployments. Confluent recommends KRaft mode for new deployments. To learn more about running Kafka in KRaft mode, see [KRaft Overview for Confluent Platform](../kafka-metadata/kraft.md#kraft-overview), [KRaft Configuration for Confluent Platform](../kafka-metadata/config-kraft.md#configure-kraft), and the [Platform Quick Start](../get-started/platform-quickstart.md#cp-quickstart-step-1), and [Settings for other Kafka and Confluent Platform components](../kafka-metadata/config-kraft.md#config-cp-components-kraft). The following example provides both KRaft (*combined mode*) configurations. Another example of running multi-cluster Schema Registry in KRaft mode is shown in the [Schema Linking Quick Start for Confluent Platform](schema-linking-cp.md#schema-linking-cp-overview). Note that KRaft combined mode is for local experimentation only and is not supported by Confluent. #### Key Configurations ### KRaft mode The example properties files are based on the defaults. You can copy the default properties files to use as a basis for the specialized versions of them shown here. The example assumes the new files are in the same directories as the originals. - KRaft server file is in `$CONFLUENT_HOME/etc/kafka/server.properties` (KRaft combined mode). Note that KRaft combined mode is for local experimentation only and is not supported by Confluent. Copy this to create `server0.properties` and `server1.properties`. - Schema Registry properties file is `$CONFLUENT_HOME/etc/schema-registry/schema-registry.properties`. Copy this to create `schema-registry0.properties` and `schema-registry1.properties`. - Control Center properties file is `$CONFLUENT_HOME/etc/confluent-control-center/control-center-dev.properties`. Copy this to create `$CONFLUENT_HOME/etc/confluent-control-center/control-center-multi-sr.properties`. In addition to the configs, shown below for this file, it is recommended to comment out the `zookeeper.connect` line, as it doesn’t apply in this mode.

File	Properties
server0.properties	node.id=1 listeners=PLAINTEXT://:9092, CONTROLLER://:9094 log.dirs=/tmp/kraft-combined-logs (fresh logging directory) confluent.metrics.reporter.bootstrap.servers=localhost:9092 The following configurations are specific to multi-cluster Schema Registry setup for this broker: confluent.http.server.listeners=http://0.0.0.0:8090 confluent.schema.registry.url=http://localhost:8081
server1.properties	node.id=1 listeners=PLAINTEXT://:9093, CONTROLLER://:9095 og.dirs=/tmp/kraft-combined-logs-1(fresh logging directory) confluent.metrics.reporter.bootstrap.servers=localhost:9093 The following configurations are specific to multi-cluster Schema Registry setup for this broker: confluent.http.server.listeners=http://0.0.0.0:8091 confluent.schema.registry.url=http://localhost:8082
schema-registry0.properties	listeners=http://0.0.0.0:8081 kafkastore.bootstrap.servers=localhost:9092
schema-registry1.properties	listeners=http://0.0.0.0:8082 kafkastore.bootstrap.servers=localhost:9093 kafkastore.topic=_schemas1 (with both Schema Registry clusters on localhost, this value must be different from the default _schemas in schema-registry0.properties so that the registries do not overwrite each other) schema.registry.group.id=schema-registry-dest (with both Schema Registry clusters on localhost, this value must be different from the group ID used by schema-registry0.properties, which takes the default ID schema-registry)
control-center-multi-sr.properties	bootstrap.servers=localhost:9092 confluent.controlcenter.kafka.AK1.bootstrap.servers=localhost:9093 confluent.controlcenter.streams.cprest.url=http://0.0.0.0:8090 confluent.controlcenter.schema.registry.url=http://localhost:8081 confluent.controlcenter.kafka.AK1.cprest.url=http://0.0.0.0:8091 confluent.controlcenter.schema.registry.SR-AK1.url=http://localhost:8082 The configurations for cpcrest.url, confluent.controlcenter.kafka.AK1.cprest.url, and confluent.controlcenter.schema.registry.SR-AK1.url are new properties, specific to multi-cluster Schema Registry.

### ZooKeeper mode The example properties files are based on the defaults. You can copy the default properties files to use as a basis for the specialized versions of them shown here. The example assumes the new files are in the same directories as the originals. - ZooKeeper `zookeeper.properties` file is in `$CONFLUENT_HOME/etc/kafka/zookeeper.properties`. Copy this to create `zookeeper0.properties` and `zookeeper1.properties`. - KRaft `server.properties` file is in `$CONFLUENT_HOME/etc/kafka/`. Copy this to create `server0.properties` and `server1.properties`. - Schema Registry properties file is `$CONFLUENT_HOME/etc/schema-registry/schema-registry.properties`. Copy this to create `schema-registry0.properties` and `schema-registry1.properties`. - Control Center properties file is `$CONFLUENT_HOME/etc/confluent-control-center/control-center-dev.properties`. Copy this to create `$CONFLUENT_HOME/etc/confluent-control-center/control-center-multi-sr.properties`.

File	Properties
zookeeper0.properties	dataDir=/tmp/zookeeper/zk-0 (fresh logging directory) clientPort=2181 (same as zookeeper.properties)
zookeeper1.properties	dataDir=/tmp/zookeeper/zk-1(fresh logging directory) clientPort=2182 (since zookeeper0 is already using 2181)
server0.properties	broker.id=0 listeners=PLAINTEXT://:9092 log.dirs=/tmp/kafka-logs-bk-0 (fresh logging directory) zookeeper.connect=localhost:2181 confluent.metrics.reporter.bootstrap.servers=localhost:9092 The following configurations are specific to multi-cluster Schema Registry setup for this broker: confluent.http.server.listeners=http://0.0.0.0:8090 confluent.schema.registry.url=http://localhost:8081
server1.properties	broker.id=1 listeners=PLAINTEXT://:9093 log.dirs=/tmp/kafka-logs-bk-1(fresh logging directory) zookeeper.connect=localhost:2182 confluent.metrics.reporter.bootstrap.servers=localhost:9093 The following configurations are specific to multi-cluster Schema Registry setup for this broker: confluent.http.server.listeners=http://0.0.0.0:8091 confluent.schema.registry.url=http://localhost:8082
schema-registry0.properties	listeners=http://0.0.0.0:8081 kafkastore.bootstrap.servers=localhost:9092
schema-registry1.properties	listeners=http://0.0.0.0:8082 kafkastore.bootstrap.servers=localhost:9093 kafkastore.topic=_schemas1 (with both Schema Registry clusters on localhost, this value must be different from the default _schemas in schema-registry0.properties so that the registries do not overwrite each other) schema.registry.group.id=schema-registry-dest (with both Schema Registry clusters on localhost, this value must be different from the group ID used by schema-registry0.properties, which takes the default ID schema-registry)
control-center-multi-sr.properties	bootstrap.servers=localhost:9092 zookeeper.connect=localhost:2181 confluent.controlcenter.kafka.AK1.bootstrap.servers=localhost:9093 confluent.controlcenter.kafka.AK1.zookeeper.connect=localhost:2182 confluent.controlcenter.streams.cprest.url=http://0.0.0.0:8090 confluent.controlcenter.schema.registry.url=http://localhost:8081 confluent.controlcenter.kafka.AK1.cprest.url=http://0.0.0.0:8091 confluent.controlcenter.schema.registry.SR-AK1.url=http://localhost:8082 The configurations for cpcrest.url, confluent.controlcenter.kafka.AK1.cprest.url, and confluent.controlcenter.schema.registry.SR-AK1.url are new properties, specific to multi-cluster Schema Registry.

#### Run the Example ### KRaft mode To run the example in KRaft mode: 1. Configure cluster IDs and format log directories for the Kafka clusters. 2. Start the Kafka brokers in dedicated command windows, one per broker. 3. Start the Schema Registry clusters in dedicated command windows, one per Schema Registry cluster. 4. Start Confluent Control Center in its own dedicated command window. **Configure KRaft specific settings for server0 and server1** The following configuration commands must be run from `$CONFLUENT_HOME` (the top level directory where you installed Confluent Platform). Assuming you configured your example KRaft servers in the same directory as the default `server.properties` file, this would be `$CONFLUENT_HOME/etc/kafka`. 1. In a new command window where you plan to run server0, generate a `random-uuid` for server0 using the kafka-storage tool. ```bash KAFKA_CLUSTER_ID="$(bin/kafka-storage random-uuid)" ``` 2. Format the log directories for server0: ```bash ./bin/kafka-storage format -t $KAFKA_CLUSTER_ID -c $CONFLUENT_HOME/etc/kafka/server0.properties --ignore-formatted ``` This is the dedicated window in which you will run server0. 3. In a new command window where you plan to run server1, a `random-uuid` for server1 using the kafka-storage tool. ```bash KAFKA_CLUSTER_ID="$(bin/kafka-storage random-uuid)" ``` 4. Format the log directories for server0: ```bash ./bin/kafka-storage format -t $KAFKA_CLUSTER_ID -c $CONFLUENT_HOME/etc/kafka/server1.properties --ignore-formatted ``` This is the dedicated window in which you will run server0. **Start the Kafka brokers** ```none kafka-server-start etc/kafka/server0.properties ``` ```none kafka-server-start etc/kafka/server1.properties ``` **Start Schema Registry clusters** ```none schema-registry-start etc/schema-registry/schema-registry0.properties ``` ```none schema-registry-start etc/schema-registry/schema-registry1.properties ``` **Start Control Center** ```none control-center-start etc/confluent-control-center/control-center-multi-sr.properties ``` ### ZooKeeper mode To run the example in ZooKeeper mode: 1. Start the ZooKeepers in dedicated command windows, one per ZooKeeper. 2. Start the Kafka brokers in dedicated command windows, one per broker. 3. Start the Schema Registry clusters in dedicated command windows, one per Schema Registry cluster. 4. Start Confluent Control Center in its own dedicated command window. **Start ZooKeepers** ```none zookeeper-server-start etc/kafka/zookeeper0.properties ``` ```none zookeeper-server-start etc/kafka/zookeeper1.properties ``` **Start the Kafka brokers** ```none kafka-server-start etc/kafka/server0.properties ``` ```none kafka-server-start etc/kafka/server1.properties ``` **Start Schema Registry clusters** ```none schema-registry-start etc/schema-registry/schema-registry0.properties ``` ```none schema-registry-start etc/schema-registry/schema-registry1.properties ``` **Start Control Center** ```none control-center-start etc/confluent-control-center/control-center-multi-sr.properties ``` #### Manage Schemas for Both Clusters on Control Center 1. When the example clusters are running and Control Center finishes initialization, open Control Center in your web browser. (Control Center runs at `http://localhost:9021/` by default, as described in [Install and access](https://docs.confluent.io/control-center/current/overview.html#install-and-access).) 2. Select a cluster from the navigation bar, click the **Topics** menu, and explore the schema management options for one or both clusters. ### Security Any other configurations used to set up a Schema Registry client with Control Center can be configured for an additional Schema Registry cluster by simply appending the Schema Registry cluster’s name to the `confluent.controlcenter.schema.registry` prefix. For example, for HTTP Basic authentication with multi-cluster Schema Registry, specify the following in the Confluent Control Center configuration file: - Use `confluent.controlcenter.schema.registry.basic.auth.credentials.source` and `confluent.controlcenter.schema.registry.basic.auth.user.info` to define authentication for the `confluent.controlcenter.schema.registry.url` cluster. - Use `confluent.controlcenter.schema.registry.{name}.basic.auth.credentials.source` and `confluent.controlcenter.schema.registry.{name}.basic.auth.user.info` for additional Schema Registry clusters (associated with the URL fields by `{name}`). Some Schema Registry client configurations also include a `schema.registry` prefix. For TLS/SSL security settings, specify the following in the Confluent Control Center configuration file: - Use `confluent.controlcenter.schema.registry.schema.registry.ssl.truststore.location` and `confluent.controlcenter.schema.registry.schema.registry.ssl.truststore.password` for the `confluent.controlcenter.schema.registry.url` cluster. - Use `confluent.controlcenter.schema.registry.{name}.schema.registry.ssl.truststore.location` and `confluent.controlcenter.schema.registry.{name}.schema.registry.ssl.truststore.password` for additional Schema Registry clusters (associated with the URL fields by `{name}`). To learn more, see Schema Registry authentication properties in the Control Center Configuration Reference under [Security for Confluent Platform components settings](https://docs.confluent.io/control-center/current/installation/configuration.html#controlcenter-configuration-encryption) and the section on [Schema Registry](../security/authentication/http-basic-auth/overview.md#basic-auth-sr) in HTTP Basic authentication. To learn more, see [How to configure clients to Schema Registry](security/index.md#sr-https-additional) in the Schema Registry Security Overview. ### Errors and Troubleshooting If the brokers for the cluster have matching Schema Registry URLs, but these URLs were not defined in the Control Center properties file, an error message is displayed on the cluster overview page.  ### Version Compatibility The `confluent.controlcenter.schema.registry.url` configuration in the Control Center properties file acts as a default if a cluster’s broker configurations do not contain `confluent.schema.registry.url fields`. Multiple Schema Registry clusters may be specified with `confluent.controlcenter.schema.registry.{name}.url` fields. Multi-cluster Schema Registry cannot be used with Kafka versions prior to Kafka 2.4.x, the version current with Confluent Platform 5.4.0. However, using a single cluster Schema Registry setup will work with earlier Kafka versions. To learn more, see [Confluent Platform and Apache Kafka compatibility](../installation/versions-interoperability.md#cp-ak-compatibility). ### Suggested Reading - See [Tutorial: Use Schema Registry on Confluent Platform to Implement Schemas for a Client Application](schema_registry_onprem_tutorial.md#schema-registry-onprem-tutorial) for hands-on examples of developing schemas, mapping to topics, and sending messages. - See [Schema Registry](/platform/current/schema-registry/index.html) for an overview of Schema Registry and schema management. - [Control Center Configuration Reference for Confluent Platform](https://docs.confluent.io/control-center/current/installation/configuration.html#controlcenter-configuration) - [Control Center Configuration Examples for Confluent Platform](https://docs.confluent.io/control-center/current/installation/properties.html#c3-properties-files) - [A schema for message values has not been set for this topic](https://docs.confluent.io/control-center/current/installation/troubleshooting.html#c3-schema-registry-not-set-up) in [Troubleshoot Control Center for Confluent Platform](https://docs.confluent.io/control-center/current/installation/troubleshooting.html) - [Validate Broker-side Schemas IDs in Confluent Platform](schema-validation.md#schema-validation) - [Change the subject naming strategy for a topic](schema-validation.md#sr-per-topic-subject-name-strategy)