Validate Broker-side Schema IDs in Confluent Platform

Schema ID Validation enables the broker to verify that data produced to a Kafka topic uses a valid schema ID. The schema ID must exist in Schema Registry under a subject that follows the subject naming strategy.

Schema ID Validation doesn’t inspect the data. It checks that Schema Registry lists the schema ID from the wire format under a valid subject.

You must use a serializer and deserializer (SerDes) that follow the Wire format or use a Confluent-supported SerDes, as described in Formats, Serializers, and Deserializers.

Schema ID Validation works when the topic name matches a subject and when it matches a schema alias, such as a named context.

Tip

Schema ID Validation is also available on Confluent Cloud clusters that use the hosted, per-environment Schema Registry. To learn more, see Using Broker-Side Schema ID Validation on Confluent Cloud.

Limitations

• Schema ID Validation supports only the default context. Other contexts risk subject name collisions in some cases. To learn more about schema contexts, see Use Schema Contexts in Confluent Platform and Schema Linking on Confluent Platform.

• Schema ID Validation doesn’t reject tombstone records, which have a null value, even when the record has no schema ID. Messages with a null value or a null key pass validation. This design supports data management and deletion in compacted topics.

Prerequisites and setting Schema Registry URLs on the brokers

The requirements for these examples match those for the Schema Registry Tutorial, except that you don’t need Maven.

To enable Schema ID Validation on the brokers, also set confluent.schema.registry.url in the Kafka server.properties file ($CONFLUENT_HOME/etc/kafka/server.properties) before you start Confluent Platform. This property tells the broker how to connect to Schema Registry.

For example:

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

This property accepts a comma-separated list of Schema Registry URLs. You need it to make Schema ID Validation available in the Confluent CLI and Control Center for Confluent Platform.

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 validation and key schema validation are independent. You can enable either or both. Both default to false.

For example, the following command creates a topic named my-topic-sv with schema validation enabled on the value schema:

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

The output is:

Created topic my-topic-sv.

With this configuration, the broker rejects any message sent to my-topic-sv without a valid value schema. The broker returns an error to the producer and discards the message.

If a producer sends a batch and at least one message is invalid, the broker discards the entire batch.

Tip

• To learn more about value and key schemas, see Schema Registry terms: topic, schema, and subject in the Schema Registry tutorial.

• To learn how to create a schema for a topic, see Manage Schemas in Confluent Platform and Control Center.

Add Schema ID Validation to an existing topic

Create a new topic called my-first-topic:

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

The output is:

Created topic my-first-topic.

To change the validation setting on an existing topic from false to true, use the --alter and --add-config flags:

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:

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

To turn off validation on this topic, rerun the previous command with --add-config confluent.value.schema.validation=false.

Change the subject naming strategy for a topic

By default, Confluent Server uses the TopicNameStrategy as the naming strategy to map topics with schemas in Schema Registry.

The naming strategy belongs to each topic. You can set a strategy other than the default for each topic. Use confluent.key.subject.name.strategy for key schema subjects and confluent.value.subject.name.strategy for value schema subjects.

From the Confluent CLI, use the --config option to create or modify a topic with a naming strategy.

To create a topic that uses RecordNameStrategy for the value:

./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 change a topic to use RecordNameStrategy for the key:

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.

Enabling Schema ID Validation on a topic in Confluent Control Center

You can view, enable, or turn off Schema ID Validation on a topic in Control Center, and configure per-topic subject naming strategies. For the steps, see Manage Schemas in Confluent Platform and Control Center in the Control Center guide.

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/):

1. Click the Configuration tab on an existing topic, and click Edit settings.

2. Click Switch to expert mode.

   

3. In expert mode, change the settings for confluent.value.schema.validation and confluent.key.schema.validation from false to true.

   You might need to scroll down to find confluent.key.schema.validation.

   

4. Click Save changes.

Change the subject naming strategy for a topic

To change the subject naming 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.

   

2. Change 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 perform the same tasks with the Confluent CLI and read more background on this feature, see Change the subject naming strategy for a topic.

Demo: Enabling Schema ID Validation on a topic at the command line

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

If you’re new to Confluent Platform and Schema Registry, first work through the Tutorial: Use Schema Registry on Confluent Platform to define schemas for a client application, and then return to this demo.

The examples use kafka-console-producer and kafka-console-consumer, which are in $CONFLUENT_HOME/bin.

1. On a local Confluent Platform instance, add the following Schema Registry URL configuration to $CONFLUENT_HOME/etc/kafka/server.properties:

   ############################## My Schema Validation Demo Settings ################
   # Schema Registry URL
   confluent.schema.registry.url=http://localhost:8081
   

   The two comment lines are optional. They help you keep track of the configurations in the file.

2. Start Confluent Platform:

   confluent local start
   

   Tip

   • Alternatively, you can simply run confluent local services schema-registry start which also starts kafka and zookeeper as dependencies. This demo does not directly reference the other services, such as Connect and Control Center. That said, you may want to run the full stack anyway to further explore, for example, how the topics and messages display on Control Center. To learn more about confluent local, see Quick Start for Confluent Platform and confluent local in the Confluent CLI command reference.

   • The confluent local commands run in the background so you can re-use this command window. Separate sessions are required for the producer and consumer.

3. Create a test topic called test-schemas without specifying the Schema ID Validation setting so that it defaults to false.

   kafka-topics --bootstrap-server localhost:9092 --create --partitions 1 --replication-factor 1 --topic test-schemas
   

   This creates a topic with no broker validation, as the first part of the demo requires. To verify that the topic exists, run 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.

   kafka-console-producer --bootstrap-server localhost:9092 --topic test-schemas --property parse.key=true --property key.separator=,
   

   The command succeeds because Schema ID Validation is disabled for this topic. With broker Schema ID Validation enabled, producing to this topic with the same command would fail.

   The output is a producer prompt of >, where you type the messages to produce.

   At the > prompt, type your first message:

   1,my first record
   

   Keep the producer session running.

5. Open a new command window for the consumer, and enter this command to read the messages:

   kafka-console-consumer --bootstrap-server localhost:9092 --from-beginning --topic test-schemas --property print.key=true
   

   The output is my first record.

   Keep the consumer session running.

6. Set Schema ID Validation for the test-schemas topic to true.

   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.

   2,my second record
   

   The producer returns an error because Schema ID Validation is enabled and the message doesn’t contain a schema ID: This record has failed the validation on broker

   To turn off Schema ID Validation again, run the same command with false. Restart the producer, and send the same or a similar message, such as 3,my third record. The message goes through.

   Tip

   After you set schema validation back to false, you can skip restarting the producer. Paste 3,my third record on the blank line after the error, and press Enter. The consumer picks up the message, and the producer prompt returns. Don’t press Enter before you paste the message, because that shuts down the producer.

   To see the messages that were produced successfully, open Control Center at http://localhost:9021/ in your web browser, and go to Topics > test-schemas > messages. You might have to select a partition or jump to a timestamp to see earlier messages.

   

8. 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 ID Validation tracks the maximum schema ID in Schema Registry. If a schema ID is too far past that maximum, Schema ID Validation treats it as invalid. The maximum can be slightly out of date. You can configure an extra range past it. Schema ID Validation then checks whether the schema ID exists in Schema Registry.

The confluent.missing.id.query.range property sets how far past the maximum schema ID Schema ID Validation checks whether an ID exists in Schema Registry. Beyond that range, Schema ID Validation treats schema IDs as invalid. The default is 200. With the default, a schema ID greater than the maximum plus 200 is invalid and triggers an alert.

Messages with null keys or values will pass schema validation

A message with a null value passes validation when you enable confluent.value.schema.validation, although a null value doesn’t follow the Wire format.

A message with a null key passes validation when you enable confluent.key.schema.validation, although a null key doesn’t follow the Wire format.

Configuring security for Schema ID Validation

In general, Schema Registry connects to the brokers. Schema ID Validation is the exception, because the brokers connect to Schema Registry. They connect to retrieve schemas from the registry and verify that the messages they receive from producers match the schemas associated with particular topics. With Schema ID Validation enabled, the sequence looks 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.

To set up security on a cluster with broker-side Schema ID Validation enabled, configure each Kafka broker to support its connection to Schema Registry. For example, with mTLS, ideally use a different certificate for each broker.

The internal Kafka client that Schema Registry uses to connect to the brokers is separate from the connection between broker-side Schema ID Validation and the Schema Registry HTTP listeners. The security settings in this section don’t apply to that internal connection.

The following broker configurations include confluent.schema.registry.url, which tells the broker how to connect to Schema Registry. You might already have set it as a prerequisite for using Schema Validation. The remaining settings are security configurations.

Tip

The following sections cover security configurations on the brokers. To set up security on Schema Registry, see Secure Schema Registry for Confluent Platform and related sections. For example, to require clients, such as brokers, to authenticate with a username and password, see Schema Registry under Use HTTP Basic Authentication in Confluent Platform.

mTLS (mutual TLS) authentication

With mutual TLS (mTLS) authentication, also known as Mutual TLS (mTLS), the broker authenticates to Schema Registry with certificates and uses TLS for encryption.

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

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>

Basic authentication

In this setup, the brokers authenticate to Schema Registry with basic authentication.

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

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.source to USER_INFO, you must also specify confluent.basic.auth.user.info.

Basic authentication with SSL

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

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>

For the Schema Registry side of HTTP basic authentication, see Schema Registry in the security documentation.

Role-based access control (RBAC) and ACLs

You can configure Role-Based Access Control, ACLs, or both for authorization to Schema Registry, and use them with other types of security on the brokers.

The Schema Registry Security Plugin for Confluent Platform supports authorization for both RBAC and ACLs.

Configure the brokers

Define the settings for your security type on the brokers, as in the previous sections. For example, for basic authentication, the broker properties files contain:

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 authorize Schema Registry

At a minimum, Schema Registry needs the following role binding:

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

With broker-side schema ID validation on a topic, the broker connects to Schema Registry. Starting with Confluent Platform 7.7, you can configure the Schema Registry server for Open Authentication (OAuth) to secure Schema Registry services and clients. Broker-side schema ID validation requires specific configurations on the Kafka broker. To learn about the broker, Schema Registry server, and client configurations, see Configure OAuth for Schema Registry in Confluent Platform.

License

Broker-side Schema ID Validation requires a Confluent Enterprise license. Apply the license to the brokers, not to the Schema Registry node. To learn more, see License.

A Confluent Enterprise license is available for the Customer-Managed Confluent Platform for Confluent Cloud subscription. This license allows you to use self-managed Confluent Platform components with Confluent Cloud services only. If you use the Customer-Managed Confluent Platform for Confluent Cloud license with Schema Registry, you can’t use Schema ID validation with Kafka brokers on Confluent Platform.

Related content

• Blog post: Schema Validation with Confluent Platform 5.4

• Schema ID Validation and Replicator explains how Schema ID Validation interacts with Replicator configurations used for replicating data from one cluster to another