<a id="schema-validation"></a>

# Validate Broker-side Schemas IDs in Confluent Platform

Schema ID Validation enables the broker to verify that data produced to a Kafka topic is using a valid schema ID in Schema Registry that is registered according to the [subject naming strategy](fundamentals/serdes-develop/index.md#sr-schemas-subject-name-strategy).

Schema Validation does not perform data introspection, but rather checks that the schema ID in the Wire Format is registered in Schema Registry under a valid subject.

You must use a serializer and deserializer (serdes) that respect the [Wire format](fundamentals/serdes-develop/index.md#messages-wire-format), or use a Confluent supported serde, as described in [Formats, Serializers, and Deserializers](/platform/current/schema-registry/fundamentals/serdes-develop/index.html).

Schema validation works both in cases where the topic name matches a subject or the topic name matches a schema alias (for example, a named context).

## Limitations

- Schema validation is supported only in the default context. Otherwise, you run the risk of subject name collisions for some cases.
  To learn more about schema contexts, see [Use Schema Contexts in Confluent Platform](schema-contexts-cp.md#work-with-schema-contexts) and [Schema Linking on Confluent Platform](schema-linking-cp.md#schema-linking-cp-overview).
- Schema validation feature does not reject tombstone records, records with a null value, even if there is no schema ID associated with the record. Messages with a null value or a null key will pass validation.
  This is a design choice that supports effective data management and deletion in compacted topics.

<a id="sv-set-sr-url-on-brokers"></a>

## Prerequisites and Setting Schema Registry URLs on the Brokers

Basic requirements to run these examples are generally the same as those described for the [Schema Registry Tutorial](schema_registry_onprem_tutorial.md#sr-tutorial-prereqs)
with the exception of Maven, which is not needed here. Also, Confluent Platform version 5.4.0 or later is required here.

As an additional prerequisite to enable Schema ID Validation on the brokers, you must specify `confluent.schema.registry.url` in the Kafka
`server.properties` file (`$CONFLUENT_HOME/etc/kafka/server.properties`) before you start Confluent Platform.  This tells the broker
how to connect to Schema Registry.

For example:

```none
confluent.schema.registry.url=http://schema-registry:8081
```

This configuration accepts a comma-separated list of URLs for Schema Registry instances. This setting is required to make Schema ID Validation available both from the
[Confluent CLI](/ccloud-cli/current/command-reference/index.html) and on the [Control Center for Confluent Platform](https://docs.confluent.io/control-center/current/overview.html).

## Enabling Schema ID Validation on a topic from the Confluent CLI

You can enable Schema ID Validation on a topic when you create a topic or modify an existing topic.

### Create a Topic with Schema ID Validation

To set Schema ID Validation on a topic when you create the topic, set `confluent.value.schema.validation=true` and `confluent.key.schema.validation=true`.

Value schema and key schema validation are independent of each other; you can enable either or both. (By default, schema validation is not enabled; both key and value schema validation default to `false`.)

For example, this command creates a topic called `my-topic-sv` with schema validation enabled on the value schema:

```none
kafka-topics --create --bootstrap-server localhost:9092 --replication-factor 1 \
--partitions 1 --topic my-topic-sv \
--config confluent.value.schema.validation=true
```

The output of this command is:

```none
Created topic my-topic-sv.
```

With this configuration, if a message is produced to the topic `my-topic-sv`
that does not have a valid schema for the value of the message, an error is
returned to the producer, and the message is discarded.

If a batch of messages is sent, and at least one is invalid, then the entire batch
is discarded.

### Add Schema ID Validation to an Existing Topic

Create a new topic called `my-first-topic`:

```none
kafka-topics --create --bootstrap-server localhost:9092 --replication-factor 1 --partitions 1 --topic my-first-topic
```

The output of this command is:

```none
Created topic my-first-topic.
```

To change topic validation configuration on an existing topic (in this case from `false` to `true`), specify validation using the `alter` and `--add-config`
flags as shown in this example:

```none
kafka-configs --bootstrap-server localhost:9092 --alter --entity-type topics --entity-name my-first-topic --add-config confluent.value.schema.validation=true
```

You should get this confirmation:

```none
Completed updating config for topic 'my-first-topic'.
```

To disable topic validation on this topic (from `true` to `false`), rerun the above command with `--add-config confluent.value.schema.validation=false`.

<a id="sr-per-topic-subject-name-strategy"></a>

### Change the subject naming strategy for a topic

By default, Confluent Server uses the `TopicNameStrategy` as the [naming strategy](fundamentals/serdes-develop/index.md#sr-schemas-subject-name-strategy) to map topics with schemas in Schema Registry.

Before Confluent Platform 5.5.0, the subject name strategy was configured on the brokers in `server.properties`, which required that you use the same strategy for all topics on a broker.

The naming strategy is associated with the topics. Therefore, you now have the option to configure
a naming strategy to something other than the default on a per-topic basis for both the schema subject key and value with
`confluent.key.subject.name.strategy` and `confluent.value.subject.name.strategy`.

From the Confluent CLI, use the `--config` option to create or modify a topic with the specified naming strategy. For example:

To create a topic that uses `RecordNameStrategy` for the value:

```bash
./bin/kafka-topics --create --bootstrap-server localhost:9092 \
--replication-factor 1 --partitions 1 --topic my-other-cool-topic \
--config confluent.value.schema.validation=true --config confluent.value.subject.name.strategy=io.confluent.kafka.serializers.subject.RecordNameStrategy
```

To modify a topic to use `RecordNameStrategy` as the key:

```bash
kafka-configs --bootstrap-server localhost:9092 \
--alter --entity-type topics --entity-name my-other-cool-topic \
--add-config confluent.key.subject.name.strategy=io.confluent.kafka.serializers.subject.RecordNameStrategy
```

The possible configurations for both `confluent.value.subject.name.strategy` and `confluent.key.subject.name.strategy` are:

- `io.confluent.kafka.serializers.subject.TopicNameStrategy` (the default)
- `io.confluent.kafka.serializers.subject.RecordNameStrategy`
- `io.confluent.kafka.serializers.subject.TopicRecordNameStrategy`

#### SEE ALSO
[Schemas, subjects, and topics](fundamentals/index.md#sr-subjects-topics-primer).

## Enabling Schema ID Validation on a topic in Confluent Control Center

You can view, enable, or disable Schema ID Validation on a topic in Control Center (see [Manage Schemas in Confluent Platform and Control Center](schema.md#topicschema) in Control Center guide). Additionally, you can configure per-topic subject naming strategies.

### View or Change Schema ID Validation settings for a topic

To view current configurations or enable Schema ID Validation on a topic from the Control Center ([http://localhost:9021/](http://localhost:9021/)):

1. Click the **Configuration** tab on an existing topic, and click **Edit settings**.
2. Click **Switch to expert mode**.
   ![image](images/sv-c3-topic-expert-settings.png)
3. In Expert mode, change the settings for `confluent.value.schema.validation` and `confluent.key.schema.validation` from false to **true**.

   You may need to scroll down to find `confluent.key.schema.validation`.
   ![image](images/sv-c3-topic-enable.png)
4. Click **Save changes**.

### Change the subject naming strategy for a topic

To change the [subject naming strategy](fundamentals/serdes-develop/index.md#sr-schemas-subject-name-strategy) using Control Center:

1. On Control Center, select the topic to update, click **Configuration**, then click **Switch to expert mode**.

   Search for `confluent.value.subject.name.strategy` and `confluent.key.subject.name.strategy`.
   ![image](images/c3-schema-subject-name-strategy.png)
2. Modify the settings and click **Save changes**.

The possible configurations for both `confluent.value.subject.name.strategy` and `confluent.key.subject.name.strategy` are:

- `io.confluent.kafka.serializers.subject.TopicNameStrategy` (the default)
- `io.confluent.kafka.serializers.subject.RecordNameStrategy`
- `io.confluent.kafka.serializers.subject.TopicRecordNameStrategy`

To learn more, see [Change the subject naming strategy for a topic](#sr-per-topic-subject-name-strategy), which describes how to perform the same tasks on the Confluent CLI and gives more background information on the feature.

## Demo: Enabling Schema ID Validation on a Topic at the Command Line

This short demo shows the effect of enabling or disabling schema validation on a topic.

If you are just getting started with Confluent Platform and Schema Registry, you might want to first work through the [Tutorial: Use Schema Registry on Confluent Platform to Implement Schemas for a Client Application](schema_registry_onprem_tutorial.md#schema-registry-onprem-tutorial), then return to this demo.

The examples make use of the `kafka-console-producer` and `kafka-console-consumer`, which are located in `$CONFLUENT_HOME/bin`.

1. On a local install of Confluent Platform version 5.4.0 or later, modify `$CONFLUENT_HOME/etc/kafka/server.properties` to include the following configuration for the Schema Registry URL:
   ```bash
   ############################## My Schema Validation Demo Settings ################
   # Schema Registry URL
   confluent.schema.registry.url=http://localhost:8081
   ```

   The example above includes two lines of comments, which are optional, to keep track of the configurations in the file.
2. Start Confluent Platform using the following command:
   ```bash
   confluent local start
   ```

   <!-- start confluent for producer, consumer demos -->
3. Create a test topic called `test-schemas` without specifying the Schema ID Validation setting so that it defaults to `false`.
   ```bash
   kafka-topics --bootstrap-server localhost:9092 --create --partitions 1 --replication-factor 1 --topic test-schemas
   ```

   This creates a topic with no broker validation on records produced to the test topic, which is what you want for the first part of the demo.
   You can verify that the topic was created with `kafka-topics --bootstrap-server localhost:9092 --list`.
4. In a new command window for the producer, run this command to produce a serialized record (using the default string serializer) to the topic `test-schemas`.
   ```bash
   kafka-console-producer --bootstrap-server localhost:9092 --topic test-schemas --property parse.key=true --property key.separator=,
   ```

   The command is successful because you currently have Schema ID Validation disabled for this topic. If broker Schema ID Validation had been enabled for this topic, the above command to produce to it would not be permitted.

   The output of this command is a producer command prompt (`>`), where you can type the messages you want to produce.

   Type your first message at the `>` prompt as follows:
   ```bash
   1,my first record
   ```

   Keep this session of the producer running.
5. Open a new command window for the consumer, and enter this command to read the messages:
   ```bash
   kafka-console-consumer --bootstrap-server localhost:9092 --from-beginning --topic test-schemas --property print.key=true
   ```

   The output of this command is `my first record`.

   Keep this session of the consumer running.
6. Now, set Schema ID Validation for the topic `test-schemas` to `true`.
   ```bash
   kafka-configs --bootstrap-server localhost:9092 --alter --entity-type topics --entity-name test-schemas --add-config confluent.value.schema.validation=true
   ```

   You should get a confirmation: `Completed updating config for topic test-schemas.`
7. Return to the producer session, and type a second message at the `>` prompt.
   ```bash
   2,my second record
   ```

   You will get an error because Schema ID Validation is enabled and the messages we are sending do not contain schema IDs: `This record has failed the validation on broker`

   If you subsequently disable Schema ID Validation (use the same command to set it to `false`), restart the producer, then type and resend the same or another similarly formatted message,
   the message will go through. (For example, produce `3,my third record`.)

   The messages that were successfully produced also show on Control Center ([http://localhost:9021/](http://localhost:9021/) in your web browser) in
   **Topics > test-schemas > messages**. You may have to select a partition or jump to a timestamp to see messages sent earlier.
   ![image](images/sv-topics.png)
8. <!-- WARNING: THIS IS A SHARED FILE AND THE SOURCE IS LOCATED IN DOCS-COMMON. DO NOT ADD TO ANY OTHER REPO. -->
   <!-- How to stop producers, consumers, confluent server for SR serdes test drives -->

   Run shutdown and cleanup tasks.
   - You can stop the consumer and producer with Ctl-C in their respective command windows.
   - To stop Confluent Platform, type `confluent local services stop`.
   - If you would like to clear out existing data (topics, schemas, and messages) before starting again with another test, type `confluent local destroy`.

## Configuring the range for valid schema IDs

Schema Validation tracks the maximum schema ID seen in Schema Registry.  If a
schema ID is seen that is too far above the maximum schema ID, it is automatically
considered invalid. Since the maximum schema ID may be slightly out of date, you can
configure an additional range above the maximum schema ID to allow Schema
Validation to verify whether the schema ID exists in Schema Registry.

The property `confluent.missing.id.query.range` indicates the range above the
maximum schema ID beyond which schemas are automatically considered invalid.
The default value for `confluent.missing.id.query.range` is `200`. With the
default setting, if the schema ID is greater than the maximum schema ID plus 200,
it is automatically considered invalid and an alert is triggered.

<a id="null-key-or-value-messages-pass-validation"></a>

## Messages with null keys or values will pass schema validation

If you have `confluent.value.schema.validation` enabled and produce a Kafka message with a null value, it will pass schema validation even though the value does not follow
the protocol defined for the [Wire format](fundamentals/serdes-develop/index.md#messages-wire-format) (because it is null).

If you have `confluent.key.schema.validation` enabled, and produce a Kafka message with null key, it will pass schema validation, even though the key does not follow
the protocol defined for the [Wire format](fundamentals/serdes-develop/index.md#messages-wire-format) (because it is null).

<a id="sv-broker-sr-security-settings"></a>

## Configuring Security for Schema ID Validation

In general, Schema Registry initiates the connection to the brokers. Schema ID Validation is unique in that
the broker(s) initiate the connection to Schema Registry. They do so in order to retrieve
schemas from the registry, and verify that the messages they receive from
producers match schemas associated with particular topics. With Schema ID Validation enabled,
the sequence of tasks looks something like this:

1. A broker receives a message from a producer, and sees that it’s directed to a topic that has a schema associated.
2. The broker initiates a connection to Schema Registry.
3. The broker asks for the schema associated with the topic (by schema ID).
4. Schema Registry receives the request, finds the requested schema in its schema storage, and returns it to the broker.
5. The broker validates the schema ID.

Therefore, to set up security on a cluster that has broker-side Schema ID Validation enabled on
topics, you must configure settings on the Kafka broker to support this broker-initiated
connection to Schema Registry. For multiple brokers, each broker must be configured. For example,
for mTLS, ideally you would have a different certificate for each broker.

Note that Schema Registry’s internal Kafka client to Kafka brokers is not relevant
at all to the connection between broker-side Schema ID Validation and Schema Registry’s HTTP listeners.
The security settings below do not reflect anything about the Schema Registry internal client-to-broker connection.

The broker configurations below include `confluent.schema.registry.url`, which tells the broker how to connect to Schema Registry.
You may already have configured this on your brokers, as a [prerequisite for using Schema Validation](#sv-set-sr-url-on-brokers).
The rest of the settings shown are specific to security configurations.

### mTLS (mutual TLS) Authentication

To configure [mutual TLS (mTLS) authentication](../security/authentication/mutual-tls/overview.md#kafka-ssl-authentication) (also known as [Mutual TLS (mTLS)](../security/authentication/overview.md#mtls-authentication-option)),
the broker will authenticate to Schema Registry using TLS/SSL for authentication (certificates) and [encryption](../security/protect-data/encrypt-tls.md#kafka-ssl-encryption).

Define the following settings in the broker properties file for each broker (`$CONFLUENT_HOME/etc/kafka/server.properties`).

```bash
confluent.schema.registry.url=https://<host>:<port>
confluent.ssl.truststore.location=<path to truststore file>
confluent.ssl.truststore.password=<password for the truststore if encrypted>
confluent.ssl.keystore.location=<path to file with private key for the broker>
confluent.ssl.keystore.password=<password to keystore file>
confluent.ssl.key.password=<password>
```

<a id="schema-id-validation-basic-auth"></a>

### Basic Authentication

For this setup, the brokers are configured to authenticate to Schema Registry using [basic authentication](../security/authentication/http-basic-auth/overview.md#http-basic-auth).

Define the following settings on each broker (`$CONFLUENT_HOME/etc/kafka/server.properties`).

```bash
confluent.schema.registry.url=http://<host>:<port>
confluent.basic.auth.credentials.source=<USER_INFO, URL, or SASL_INHERIT>
confluent.basic.auth.user.info=<username>:<password> #required only if credentials source is set to USER_INFO
```

- The property `confluent.basic.auth.credentials.source` defines the type of credentials to use (user name and password). These are literals, not variables.
- If you set `confluent.basic.auth.credentials` to `USER_INFO`, you must also specify `confluent.basic.auth.user.info`.

<a id="schema-id-validation-basic-auth-with-ssl"></a>

### Basic Authentication with SSL

Define the following settings on each broker (`$CONFLUENT_HOME/etc/kafka/server.properties`).

```bash
confluent.schema.registry.url=https://<host>:<port>
confluent.basic.auth.credentials.source=USER_INFO
confluent.basic.auth.user.info=<username>:<password>
confluent.ssl.truststore.location=<path to the truststore file>
confluent.ssl.truststore.password=<password for the truststore if encrypted>
```

To learn more about Basic Authentication, see [Schema Registry](../security/authentication/http-basic-auth/overview.md#basic-auth-sr) in the Security documentation for authenticating using HTTP Basic Authentication.

### Role-based Access Control (RBAC) and ACLs

You can configure [Role-Based Access Control](security/rbac-schema-registry.md#schemaregistry-rbac) (and/or [ACLs](../confluent-security-plugins/schema-registry/authorization/sracl_authorizer.md#confluentsecurityplugins-sracl-authorizer))
for authorization to Schema Registry, and use it with other types of security on the brokers.

The [Schema Registry Security Plugin for Confluent Platform](../confluent-security-plugins/schema-registry/introduction.md#confluentsecurityplugins-schema-registry-security-plugin) supports authorization for both RBAC and ACLs.

#### Configure the Brokers

Define the appropriate settings on the brokers similar to those shown in previous sections, depending on the type of security.
For example, for basic authentication, the broker properties files would contain:

```bash
confluent.schema.registry.url=http://<host>:<port>
confluent.basic.auth.credentials.source=USER_INFO
confluent.basic.auth.user.info=<username>:<password> #required only if credentials source is set to USER_INFO
```

#### Set RBAC Role Bindings to Allow Access to Schema Registry

At a minimum, in terms of defining RBAC role bindings, Schema Registry needs the following IAM assignment:

```bash
confluent iam rbac role-binding list --principal User:<user-id> \
--role DeveloperRead --kafka-cluster <kafka-cluster-id> \
--resource Subject:* --schema-registry-cluster-id <schema-registry-group-id>
```

The Schema Registry cluster ID is the same as `schema-registry-group-id`, which defaults to **schema-registry**.

### OAuth on Schema Registry server

When a service such as broker-side schema ID validation is enabled on topics, the broker tries to connect to Schema Registry.

Starting with Confluent Platform version 7.7, the Schema Registry server can be configured for Open Authentication (OAuth) to
secure Schema Registry services and clients. For broker-side schema ID validation, specific configurations are
are required on the Kafka broker. To learn more about these broker, Schema Registry server, and client configurations, see [Configure OAuth for Schema Registry in Confluent Platform](security/oauth-schema-registry.md#schemaregistry-oauth).

## License

A Confluent Enterprise license is required for broker-side Schema ID Validation. In this case, the Confluent Enterprise license must be applied to the brokers, not the Schema Registry node.
To learn more, see [License](../installation/license.md#sr-license).

Starting October 1, 2025, a new Confluent Enterprise license is available for the
[Customer-Managed Confluent Platform for Confluent Cloud](../installation/license.md#customer-managed-cp-cc-license) subscription.
This license allows you to use self-managed Confluent Platform components exclusively with Confluent Cloud services. However, If you
use the Customer-Managed Confluent Platform for Confluent Cloud license with Schema Registry, you can’t use [Schema ID validation](#schema-validation) with Kafka brokers on Confluent Platform.

## Related Content

- Blog post: [Schema Validation with Confluent Platform 5.4](https://www.confluent.io/blog/data-governance-with-schema-validation/)
- [Schema ID Validation and Replicator](../multi-dc-deployments/replicator/index.md#replicator-and-schema-validation) explains how Schema ID Validation interacts with Replicator configurations used for replicating data from one cluster to another