JSON Schema Serializer and Deserializer for Schema Registry on Confluent Platform

Use JSON Schema with the Apache Kafka® Java client and console tools to serialize and deserialize structured Kafka data.

What is JSON Schema?

JSON Schema is a declarative language that defines the structure and constraints of JSON data. Applications use it to validate and enforce the format of JSON data before processing it. Web applications and APIs widely use JSON Schema to keep data consistent and correct. JSON Schema has these key features:

JSON Schema can evolve over time while maintaining backward and forward compatibility. For more information, see Schema Evolution and Compatibility for Schema Registry on Confluent Platform.
You can validate JSON data against a JSON Schema to ensure the data meets the criteria before processing.
JSON Schema works with many programming languages, including Java, Python, and C++.
You can read and process schemas at runtime without code generation.

The JSON Schema serializer for Confluent Schema Registry doesn’t include the message schema. It includes a magic byte, the schema ID, and then the data itself as JSON. You can choose whether to embed a schema inline, for cases where you want to communicate the schema offline, with headers, or some other way. Other systems, such as Hadoop, always include the schema with the message data. For more information, see Wire format.

To make the JSON Schema serializer and deserializer fail when the payload isn’t valid for the given schema, set json.fail.invalid.schema=true. The default is false.

The following examples use the default hostname and port for the Kafka bootstrap server (localhost:9092) and Schema Registry (localhost:8081).

Supported JSON Schema specifications

Confluent Platform versions earlier than 7.6.0 fully support these JSON Schema specifications:

Draft 04
Draft 06
Draft 07

Confluent Platform 7.6.0 and later fully support these JSON Schema specifications:

Draft 04
Draft 06
Draft 07
Draft 2020-12

To set the specification version for JSON schemas derived from objects, use the json.schema.spec.version configuration option, described in JSON Schema serializer. For notes on the drafts, see the JSON Schema Specification.

JSON Schema serializer

Plug KafkaJsonSchemaSerializer into KafkaProducer to send messages of JSON Schema type to Kafka.

The following example uses a Java class decorated with Jackson annotations:

public static class User {
  @JsonProperty
  public String firstName;

  @JsonProperty
  public String lastName;

  @JsonProperty
  public short age;

  public User() {}

  public User(String firstName, String lastName, short age) {
    this(firstName, lastName, age, null);
  }
}

The following code serializes User objects:

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
  "org.apache.kafka.common.serialization.StringSerializer");
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
  "io.confluent.kafka.serializers.json.KafkaJsonSchemaSerializer");
props.put("schema.registry.url", "http://127.0.0.1:8081");

Producer<String, User> producer = new KafkaProducer<String, User>(props);

String topic = "testjsonschema";
String key = "testkey";
User user = new User("John", "Doe", 33);

ProducerRecord<String, User> record
      = new ProducerRecord<String, User>(topic, key, user);
producer.send(record).get();
producer.close();

The following additional configurations are available for JSON schemas derived from Java objects:

json.type.allowed.packages: controls which Java classes the serializer can resolve from the javaType property in JSON schemas. Provide a comma-separated list of package prefixes. The default * allows any class. An empty value turns off javaType resolution. For availability and best practices, see Configure allowed packages.
json.schema.spec.version: specifies the specification version to use for JSON schemas derived from objects. Schema Registry supports schema draft versions 4 and later. Valid values are draft_4, draft_6, draft_7, draft_2019_09, and draft_2020_12. The default is draft_7. For the versions each Confluent Platform release supports, see Supported JSON Schema specifications.
json.oneof.for.nullables: whether JSON schemas derived from objects use oneOf for nullable fields. The default is true.
json.default.property.inclusion: controls the inclusion of properties during serialization. Set it to one of the values in Jackson’s JsonInclude.Include enumeration.
json.fail.unknown.properties: whether JSON schemas derived from objects fail if they encounter unknown properties. The default is true.
json.write.dates.iso8601: whether dates are written as ISO-8601 strings. The default is false.

Use the @Schema annotation on a Java object

To pass a schema directly to the producer without deriving it from the Java object, annotate the Java class, as in the following example:

@io.confluent.kafka.schemaregistry.annotations.Schema(value="{"
        + "\"$id\": \"https://acme.com/referrer.json\","
        + "\"$schema\": \"http://json-schema.org/draft-07/schema#\","
        + "\"type\":\"object\","
        + "\"properties\":{\"Ref\":"
        + "{\"$ref\":\"ref.json#/definitions/ExternalType\"}},\"additionalProperties\":false}",
        refs={@io.confluent.kafka.schemaregistry.annotations.SchemaReference(
                name="ref.json", subject="reference")})
public class MyObject {
    ...
}

Send a JsonNode payload

KafkaJsonSchemaSerializer also accepts a JsonNode in envelope format. An envelope is an ObjectNode with two fields: schema, which holds a JSON Schema, and payload, which holds a JsonNode. The following code serializes an envelope:

byte[] serializedRecord1 = serializer.serialize(TOPIC,
     JsonSchemaUtils.envelope(rawSchemaJson1, objectNode1));

JSON Schema deserializer

Plug KafkaJsonSchemaDeserializer into KafkaConsumer to receive messages of any JSON Schema type from Kafka. The following example configures a consumer with the deserializer:

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "group1");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringDeserializer");
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, "io.confluent.kafka.serializers.json.KafkaJsonSchemaDeserializer");
props.put("schema.registry.url", "http://localhost:8081");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
props.put(KafkaJsonSchemaDeserializerConfig.JSON_VALUE_TYPE, User.class.getName());

String topic = "testjsonschema";
final Consumer<String, User> consumer = new KafkaConsumer<String, User>(props);
consumer.subscribe(Arrays.asList(topic));

try {
  while (true) {
    ConsumerRecords<String, User> records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord<String, User> record : records) {
      User user = record.value();
      // Process the user record
    }
  }
} finally {
  consumer.close();
}

Similar to how the Avro deserializer can return an instance of a specific Avro record type or a GenericRecord, the JSON Schema deserializer can return an instance of a specific Java class or an instance of JsonNode. If the JSON Schema deserializer can’t determine a specific type, it returns a generic type.

One way to return a specific type is to use an explicit property. For the JSON Schema deserializer, configure KafkaJsonSchemaDeserializerConfig.JSON_VALUE_TYPE or KafkaJsonSchemaDeserializerConfig.JSON_KEY_TYPE.

When you deserialize a JSON payload, KafkaJsonSchemaDeserializer determines the target Java type as follows:

If you set json.key.type or json.value.type, the deserializer uses the specified type to perform deserialization.
The default for json.value.type is Object.class.getName(). If no type is provided or no type can be determined, the deserializer returns a LinkedHashMap for JSON objects or a LinkedList for JSON arrays.
If you set json.value.type=com.fasterxml.jackson.databind.JsonNode, the deserializer returns an instance of a Jackson JsonNode.
The previous configuration doesn’t work for RecordNameStrategy, where a topic can hold more than one type of JSON message. To handle this case, configure the deserializer with type.property and a value that names a top-level property on the JSON Schema that specifies the fully qualified Java type to use for deserialization. For example, if type.property=javaType, the JSON schema must have an additional top-level property named javaType that specifies the fully qualified Java type. When you use the mbknor-jackson-jsonSchema utility to generate a JSON Schema from a Java POJO, use the @JsonSchemaInject annotation to specify the javaType:
```
// Generate javaType property
@JsonSchemaInject(strings =
 {@JsonSchemaString(path="javaType", value="com.acme.User")})
public static class User {
 @JsonProperty
 public String firstName;
 @JsonProperty
 public String lastName;
 @JsonProperty
 @Min(0)
 public short age;
 @JsonProperty
 public Optional<String> nickName;

 public User() {}
   ...
}
```

Here is a summary of specific and generic return types for each schema format.

	Avro	Protobuf	JSON Schema
Specific type	Generated class that implements org.apache.avro.SpecificRecord	Generated class that extends com.google.protobuf.Message	Java class (that is compatible with Jackson serialization)
Generic type	org.apache.avro.GenericRecord	com.google.protobuf.DynamicMessage	com.fasterxml.jackson.databind.JsonNode

Configure allowed packages

The json.type.allowed.packages property controls which Java classes the deserializer can resolve from the javaType property in JSON schemas. The property is available starting with Confluent Platform patch releases 7.4.15, 7.5.14, 7.6.11, 7.7.9, 7.8.8, 7.9.7, 8.0.5, 8.1.3, and 8.2.1. Set the property as follows:

To allow specific packages, provide a comma-separated list of package prefixes.
To allow any class, specify an asterisk, *, which is the default. As a best practice, don’t use this setting.
To turn off javaType resolution entirely, set the property to an empty string, "". As a best practice, use an empty string unless your application relies on javaType resolution, because it prevents arbitrary class loading during deserialization.

Configure expiry time for client-side schema caches

The following format-agnostic configuration options for cache expiry time are available on both the serializer and deserializer:

latest.cache.size - The maximum size for caches holding latest schemas
latest.cache.ttl.sec - The time to live (TTL) in seconds for caches holding latest schemas, or -1 for no TTL

Test drive JSON Schema

Use the command line producer and consumer to try JSON Schema. They show how the built-in JSON Schema support works on Confluent Platform and Confluent Cloud.

When you incorporate the serializer and deserializer into the code for your own producers and consumers, messages and associated schemas are processed the same way as they are on the console producers and consumers.

The suggested consumer commands include a flag to read --from-beginning to be sure you capture the messages even if you don’t run the consumer immediately after running the producer. If you leave off the --from-beginning flag, the consumer will read only the last message produced during its current session.

The following examples include a few minimal configurations. For full property references, see Producer and consumer configuration reference.

Prerequisites

Confluent Platform

Prerequisites to run these examples are generally the same as those described for the Schema Registry Tutorial with the exception of Maven, which is not needed here. Also, Confluent Platform version 5.5.0 or later is required here.
The following examples use the default Schema Registry URL value (localhost:8081). The examples show how to configure this inline by supplying the URL as an argument to the --property flag in the command line arguments of the producer and consumer (--property schema.registry.url=<address of your schema registry>). Alternatively, you could set this property in $CONFLUENT_HOME/etc/kafka/server.properties, and not have to include it in the producer and consumer commands. For example: confluent.schema.registry.url=http://localhost:8081
These examples make use of the kafka-avro-console-producer and kafka-avro-console-consumer, which are located in $CONFLUENT_HOME/bin.

Confluent Cloud

Confluent Cloud prerequisites are:

A Confluent Cloud account
Permission to create a topic and schema in a cluster in Confluent Cloud
Stream Governance Package enabled
API key and secret for Confluent Cloud cluster ($APIKEY, $APISECRET)
API key and secret for Schema Registry ($SR_APIKEY, $SR_APISECRET)
Schema Registry endpoint URL ($SCHEMA_REGISTRY_URL)
Cluster ID ($CLUSTER_ID)
Schema registry cluster ID ($SR_CLUSTER_ID)

The examples assume that API keys, secrets, cluster IDs, and API endpoints are stored in persistent environment variables wherever possible, and refer to them as such. You can store these in shell variables if your setup is temporary. If you want to return to this environment and cluster for future work, consider storing them in a profile (such as .zsh, .bashrc, or powershell.exe profiles).

The following steps provide guidelines on these prerequisites specific to these examples. To learn more general information, see Manage Clusters.

Log in to Confluent Cloud:
```
confluent login
```
Create a Kafka cluster in Confluent Cloud
```
confluent kafka cluster create <name> [flags]
```
For example:
```
confluent kafka cluster create quickstart_cluster --cloud "aws" --region "us-west-2"
```
Your output will include a cluster ID (in the form of lkc-xxxxxx), show the cluster name and cluster type, a Basic cluster in this case, and endpoints. Take note of the cluster ID, and store it in an environment variable such as $CLUSTER_ID.
Get an API key and secret for the cluster:
```
confluent api-key create --resource $CLUSTER_ID
```
Store the API key and secret for your cluster in a safe place, such as shell environment variables: $APIKEY, $APISECRET
View Stream Governance packages and Schema Registry endpoint URL.

A Stream Governance package was enabled as a part of creating the environment.
- To view governance packages, use the Confluent CLI command confluent environment list:
```
confluent environment list
```
  Your output will show the environment ID, name, and associated Stream Governance packages.
- To view the Stream Governance API endpoint URL, use the command confluent schema-registry cluster describe:
```
confluent schema-registry cluster describe
```
  Your output will show the Schema Registry cluster ID in the form of lsrc-xxxxxx) and endpoint URL, which is also available to you in Cloud Console on the right side panel under “Stream Governance API” in the environment. Store these in environment variables: $SR_CLUSTER_ID and $SCHEMA_REGISTRY_URL.
Create a Schema Registry API key, using the Schema Registry cluster ID ($SR_CLUSTER_ID) from the previous step as the resource ID.
```
confluent api-key create --resource $SR_CLUSTER_ID
```
Store the API key and secret for your Schema Registry in a safe place, such as shell environment variables: $SR_APIKEY and $SR_APISECRET

Create and use schemas

Create a JSON Schema, send messages that the schema doesn’t declare, and view the schemas and messages.

Confluent Platform

Start Confluent Platform with the following command:
```
confluent local services start
```
Tip
- Alternatively, you can simply run confluent local services schema-registry start which also starts kafka and KRaft as dependencies. This demo does not directly reference the other services, such as Connect and Control Center (Legacy). 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 (Legacy). 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.
Verify registered schema types.

Schema Registry supports arbitrary schema types. You should verify which schema types are currently registered with Schema Registry.

To do so, type the following command (assuming you use the default URL and port for Schema Registry, localhost:8081):
```
curl http://localhost:8081/schemas/types
```
The response will be one or more of the following. If additional schema format plugins are installed, these will also be available.
```
["JSON", "PROTOBUF", "AVRO"]
```
Alternatively, use the curl --silent flag, and pipe the command through jq (curl --silent http://localhost:8081/schemas/types | jq) to get nicely formatted output:
```
"JSON",
"PROTOBUF",
"AVRO"
```
Use the producer to send JSON records as the message value.

The following command starts a producer, creates the transactions-json topic if it doesn’t already exist, and creates a schema for the topic with two fields, id and amount:
```
kafka-json-schema-console-producer --bootstrap-server localhost:9092 \
--property schema.registry.url=http://localhost:8081 --topic transactions-json \
--property value.schema='{"type":"object", "properties":{"id":{"type":"string"},"amount":{"type":"number"} }}'
```
Tip

The producer doesn’t show a > prompt. Type producer messages at the blank line.
Type the following message in the shell, and press Enter:
```
{ "id":"1000", "amount":500 }
```

To read the message value in JSON from the transactions-json topic, run the consumer in a new terminal window:

kafka-json-schema-console-consumer --bootstrap-server localhost:9092  --from-beginning --topic transactions-json --property schema.registry.url=http://localhost:8081

The console shows the following output:

{"id":"1000","amount":500}

Leave this consumer running.

Use the producer to send another record that has a property the schema doesn’t declare.

JSON Schema has an open content model, which allows any number of additional properties in a JSON document without declaring them in the schema. This behavior comes from additionalProperties, which is true by default. Unless you set additionalProperties to false, records can include undeclared properties. The next steps demonstrate this behavior.

Return to the running producer session. The following message adds the property customer_id, which the schema doesn’t declare. Send it by pressing Enter:
```
{"id":"1000","amount":500,"customer_id":"1221"}
```
To read the new message from the transactions-json topic, return to your running consumer.

The new output appears after the original output:
```
{"id":"1000","amount":500}
{"id":"1000","amount":500,"customer_id":"1221"}
```
The producer sends the message with the new property, customer_id, and the consumer reads it. With Avro or Protobuf, the producer command fails because those formats require schemas to declare all properties.

Keep this consumer running.

Start a producer with a schema that sets additionalProperties to false.

In the producer command window, stop the producer with Ctrl+C. Then run the following command, which uses the same topic, transactions-json, and almost the same schema as before. This schema sets additionalProperties to false:

kafka-json-schema-console-producer --bootstrap-server localhost:9092  --property schema.registry.url=http://localhost:8081 --topic transactions-json \
--property value.schema='{"type":"object", "properties":{"id":{"type":"string"},"amount":{"type":"number"} }, "additionalProperties": false}'

To get the top-level compatibility configuration, use curl in another shell:
```
curl --silent -X GET http://localhost:8081/config
```
The default result is:
```
{"compatibilityLevel":"BACKWARD"}
```
Tip

If you don’t update the compatibility requirements, the next step fails with a different error than the one this example demonstrates, because of the BACKWARD compatibility setting. For more examples of using curl against the APIs to test and set configurations, see Schema Registry API Usage Examples.

To update the compatibility requirements globally, run the following command:

curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" \
--data '{"compatibility": "NONE"}' \
http://localhost:8081/config

The output is:

{"compatibilityLevel":"NONE"}

Start a new producer with a schema that sets additionalProperties to false.

You can shut down the previous producer and start this one in the same window:

kafka-json-schema-console-producer --bootstrap-server localhost:9092 \
--property schema.registry.url=http://localhost:8081 --topic transactions-json \
--property value.schema='{"type":"object", "properties":{"id":{"type":"string"}, "amount":{"type":"number"} }, "additionalProperties": false}'

Try to send a record that has a property the schema doesn’t declare:

{ "id":"1001","amount":500,"customer_id":"this-will-break"}

The command fails with the following error:

org.apache.kafka.common.errors.SerializationException: Error serializing JSON message
...
Caused by: org.apache.kafka.common.errors.SerializationException: JSON {"id":"1001","amount":500,"customer_id":"this-will-break"} does not match schema
{"type":"object","properties":{"id":{"type":"string"},"amount":{"type":"number"}},"additionalProperties":false}       at
io.confluent.kafka.serializers.json.AbstractKafkaJsonSchemaSerializer.serializeImpl(AbstractKafkaJsonSchemaSerializer.java:132)
... 5 more
Caused by: org.everit.json.schema.ValidationException: #: extraneous key [customer_id] is not permitted
...

The consumer keeps running but shows no new messages. Avro and Protobuf behave the same way by default in this scenario.

Tip

To get this behavior in the JSON Schema producer code of your applications, include "additionalProperties": false in the schemas. For examples, see the discussion about properties in Understanding JSON Schema.

Rerun the producer with the original schema.

In the producer command window, stop the producer with Ctrl+C. Then run the original producer command. You don’t need to declare additionalProperties as true, although you can, because true is the default:

kafka-json-schema-console-producer --bootstrap-server localhost:9092 \
--property schema.registry.url=http://localhost:8081 --topic transactions-json \
--property value.schema='{"type":"object", "properties":{"id":{"type":"string"},"amount":{"type":"number"} }}'

Send another record that has a property the schema doesn’t declare:
```
{ "id":"1001","amount":500,"customer_id":"this-will-work-again"}
```
To read the new message, return to the consumer session.

The consumer is still running and reading from the transactions-json topic. The console shows the following new message:
```
{"id":"1001","amount":500,"customer_id":"this-will-work-again"}
```
If you followed all steps in order and started the consumer with the --from-beginning flag, the consumer shows the history of all messages sent:
```
{"id":"1000","amount":500}
{"id":"1000","amount":500,"customer_id":"1221"}
{"id":"1001","amount":500,"customer_id":"this-will-work-again"}
```

To query the schemas registered with Schema Registry as versions 1 and 2, use curl with jq for readability in another shell.

To query version 1 of the schema, run the following command:

curl --silent -X GET http://localhost:8081/subjects/transactions-json-value/versions/1/schema | jq .

The output for version 1 is:

{
  "type": "object",
  "properties": {
    "id": {
      "type": "string"
    },
    "amount": {
      "type": "number"
    }
  }
}

To query version 2 of the schema, run the following command:

curl --silent -X GET http://localhost:8081/subjects/transactions-json-value/versions/2/schema | jq .

The output for version 2 is:

{
  "type": "object",
  "properties": {
    "id": {
      "type": "string"
    },
    "amount": {
      "type": "number"
    }
  },
  "additionalProperties": false
}

To view the latest version of the schema in more detail, run the following command:

curl --silent -X GET http://localhost:8081/subjects/transactions-json-value/versions/latest | jq .

The output is:

"subject": "transactions-json-value",
"version": 2,
"id": 2,
"schemaType": "JSON",
"schema": "{\"type\":\"object\",\"properties\":{\"id\":{\"type\":\"string\"},\"amount\":{\"type\":\"number\"}},\"additionalProperties\":false}"

Use Confluent Control Center (Legacy) to examine schemas and messages.

Messages that were successfully produced also show on Control Center (Legacy) (http://localhost:9021/) in Topics > <topicName> > Messages. You may have to select a partition or jump to a timestamp to see messages sent earlier. (For timestamp, type in a number, which will default to partition 1/Partition: 0, and press return. To get the message view shown here, select the cards icon on the upper right.)

Schemas you create are available on the Schemas tab for the selected topic.
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.

Confluent Cloud

Create a Kafka topic:

confluent kafka topic create transactions-json --cluster $CLUSTER_ID

Copy the following schema into a file named schema.txt:

{
    "type":"object",
    "properties":{
        "id":{"type":"string"},
        "amount":{"type":"number"}
    }
}

Run the following command to create a producer that uses the schema from the previous step:
```
confluent kafka topic produce transactions-json \
--cluster $CLUSTER_ID \
--schema "<full path to file>/schema.txt" \
--schema-registry-endpoint $SCHEMA_REGISTRY_URL \
--schema-registry-api-key $SR_APIKEY \
--schema-registry-api-secret $SR_APISECRET \
--api-key $APIKEY --api-secret $APISECRET  \
--value-format "jsonschema"
```
Your output should resemble:
```
Successfully registered schema with ID 100001
Starting Kafka Producer. Use Ctrl-C or Ctrl-D to exit.
```
Provide the full path to the schema file, even if the file is in the current directory.

Note

The examples assume that you use the latest version of the Confluent CLI. The new --schema-registry-endpoint, --schema-registry-api-key, and --schema-registry-api-secret flags replace the deprecated --sr-endpoint, --sr-api-key, and --sr-api-secret flags. If the CLI doesn’t recognize the new flags, either update the CLI or use the deprecated flags.
Type the following message in the shell, and press Enter:
```
{ "id":"1000", "amount":500 }
```

To read the message value in JSON from the transactions-json topic, run a consumer in another terminal:

confluent kafka topic consume transactions-json \
--cluster $CLUSTER_ID \
--from-beginning \
--value-format "jsonschema" \
--schema-registry-endpoint $SCHEMA_REGISTRY_URL \
--schema-registry-api-key $SR_APIKEY \
--schema-registry-api-secret $SR_APISECRET \
--api-key $APIKEY --api-secret $APISECRET

Your output should be:

{"id":"1000","amount":500}

Use the producer to send another record that has a property the schema doesn’t declare.

JSON Schema has an open content model, which allows any number of additional properties in a JSON document without declaring them in the schema. This behavior comes from additionalProperties, which is true by default. Unless you set additionalProperties to false, records can include undeclared properties. The next steps demonstrate this behavior.

Return to the running producer session. The following message adds the property customer_id, which the schema doesn’t declare. Send it by pressing Enter:
```
{"id":"1000","amount":500,"customer_id":"1221"}
```
To read the new message from the transactions-json topic, return to your running consumer.

The new output appears after the original output:
```
{"id":"1000","amount":500}
{"id":"1000","amount":500,"customer_id":"1221"}
```
The producer sends the message with the new property, customer_id, and the consumer reads it. With Avro or Protobuf, the producer command fails because those formats require schemas to declare all properties.

Keep this consumer running.

To update the compatibility requirement for the subject transactions-json-value, run the following command:

confluent schema-registry subject update transactions-json-value --compatibility "none"

The output is:

Successfully updated Subject Level compatibility to "none" for subject "transactions-json-value"

Copy the following schema into a file named schema2.txt:

{
    "type":"object",
    "properties":{
        "id":{"type":"string"},
        "amount":{"type":"number"}
    },
    "additionalProperties": false
}

This schema is almost the same as the original schema in schema.txt, except that it sets additionalProperties to false.

Run another producer to register the new schema.

Shut down the running producer with Ctrl+C, and then start a new one with the new schema:

confluent kafka topic produce transactions-json \
--cluster $CLUSTER_ID \
--schema "<full path to file>/schema2.txt" \
--schema-registry-endpoint $SCHEMA_REGISTRY_URL \
--schema-registry-api-key $SR_APIKEY \
--schema-registry-api-secret $SR_APISECRET \
--api-key $APIKEY --api-secret $APISECRET  \
--value-format "jsonschema"

Try to send a record that has a property the schema doesn’t declare:
```
{ "id":"1001","amount":500,"customer_id":"this-will-break"}
```
The command fails with the following error:
```
Error: the JSON document is invalid
```
The consumer keeps running but shows no new messages. Avro and Protobuf behave the same way by default in this scenario.

Tip

To get this behavior in the JSON Schema producer code of your applications, include "additionalProperties": false in the schemas. For examples, see the discussion about properties in Understanding JSON Schema.

Rerun the producer with the original schema, schema.txt.

In the producer command window, stop the producer with Ctrl+C. Then run the original producer command. You don’t need to declare additionalProperties as true in the schema, although you can, because true is the default:

confluent kafka topic produce transactions-json \
--cluster $CLUSTER_ID \
--schema "<full path to file>/schema.txt" \
--schema-registry-endpoint $SCHEMA_REGISTRY_URL \
--schema-registry-api-key $SR_APIKEY \
--schema-registry-api-secret $SR_APISECRET \
--api-key $APIKEY --api-secret $APISECRET  \
--value-format "jsonschema"

Send another record that has a property the schema doesn’t declare:
```
{"id":"1001","amount":500,"customer_id":"this-will-work-again"}
```
To read the new message, return to the consumer session.

The consumer is still running and reading from the transactions-json topic. The console shows the following new message:
```
{"id":"1001","amount":500,"customer_id":"this-will-work-again"}
```
If you followed all steps in order and started the consumer with the --from-beginning flag, the consumer shows the history of all messages sent:
```
{"id":"1000","amount":500}
{"id":"1000","amount":500,"customer_id":"1221"}
{"id":"1001","amount":500,"customer_id":"this-will-work-again"}
```

View versions 1 and 2 of the schema registered with Schema Registry. To view version 1, run the following command:

confluent schema-registry schema describe --subject transactions-json-value --version 1

The output for version 1 shows the id and amount fields:

Schema ID: 100001
Type: JSON
Schema: {"type":"object","properties":{"id":{"type":"string"},"amount":{"type":"number"}}}

To view version 2, run the following command:

confluent schema-registry schema describe --subject transactions-json-value --version 2

The output for version 2 includes the same fields and sets additionalProperties to false:

Schema ID: 100002
Type: JSON
Schema: {"type":"object","properties":{"id":{"type":"string"},"amount":{"type":"number"}},"additionalProperties":false}

Use the Confluent Cloud Console to examine schemas and messages.

Messages that were successfully produced also show on the Confluent Cloud Console (https://confluent.cloud/). in Topics > <topicName> > Messages. You may have to select a partition or jump to a timestamp to see messages sent earlier. (For timestamp, type in a number, which will default to partition 1/Partition: 0, and press return. To get the message view shown here, select the cards icon on the upper right.)

Schemas you create are available on the Schemas tab for the selected topic.
Run shutdown and cleanup tasks.
- You can stop the consumer and producer with Ctl-C in their respective command windows.
- If you were using shell environment variables and want to keep them for later, remember to store them in a safe, persistent location.
- You can remove topics, clusters, and environments from the command line or from the Confluent Cloud Console.

Producer and consumer configuration reference

The following configuration properties apply to any Kafka producers and consumers, regardless of schema format.

Add Schema Registry security credentials

Pass truststore and keystore credentials for Schema Registry as properties, in addition to the Schema Registry properties that the test drive examples pass on the command line and in configuration files. For more information, see Additional configurations for HTTPS. The following example shows a producer on Confluent Platform:

kafka-json-schema-console-producer --bootstrap-server localhost:9092 \
--property schema.registry.url=http://localhost:8081 --topic transactions-json \
--property value.schema='{"type":"object", "properties":{"id":{"type":"string"}, "amount":{"type":"number"} }, "additionalProperties": false}' \
--property schema.registry.ssl.truststore.location=/etc/kafka/security/schema.registry.client.truststore.jks \
--property schema.registry.ssl.truststore.password=myTrustStorePassword

Kafka producer configurations

For a complete reference of producer configuration properties, see Kafka Producer Configurations.

Kafka consumer configurations

For a complete reference of consumer configuration properties, see Kafka Consumer Configurations.

Schema Registry configuration options

For a complete reference of Schema Registry configuration options, see Schema Registry Configuration Options in the Confluent Platform documentation.

Use Schema Registry with Connect

If you use serializers and deserializers with Kafka Connect, you need to configure key and value converters. For details, see Configuring key and value converters in the Connect documentation.

Schema references in JSON Schema

Confluent Platform provides full support for the notion of schema references, the ability of a schema to refer to other schemas.

Tip

Schema references are also supported in Confluent Cloud on Avro, Protobuf, and JSON Schema formats. On the Confluent CLI, you can use the --refs <file> flag on confluent schema-registry schema create to reference another schema.

A schema can reference another schema with the $ref keyword. These rules apply to $ref in JSON Schemas:

If you refer to another schema within Schema Registry, the value of $ref must match the name of a schema reference registered with the schema.
If the value of $ref doesn’t match the name of any schema reference for the schema, it must be a relative URI reference that resolves against the schema’s base URI. For example, to reference a relative URI, use { "$ref": "<URI path to referenced schema>" }.
Otherwise, an error results if the $ref is a non-relative URI reference, such as an HTTP URL, or doesn’t match the name of any schema reference for the schema.

External references aren’t supported.

For examples of schema references, see Structuring a complex schema on the JSON Schema website, Multiple event types in the same topic, and the blog post Putting Several Event Types in the Same Topic – Revisited.

Multiple event types in the same topic

Use schema references to combine multiple event types in the same topic and still maintain subject-topic constraints. Schema references also let one schema call other schemas.

To do this in JSON Schema:

Use the default subject name strategy, TopicNameStrategy. It uses the topic name to determine the subject for schema lookups and helps enforce subject-topic constraints.

Use the JSON Schema construct oneOf to define a list of schema references, such as:

{
  "oneOf": [
     { "$ref": "Customer.schema.json" },
     { "$ref": "Product.schema.json" },
     { "$ref": "Order.schema.json" }
  ]
}

When you register the schema, send an array of reference versions:

[
  {
    "name": "Customer.schema.json",
    "subject": "customer",
    "version": 1
  },
  {
    "name": "Product.schema.json",
    "subject": "product",
    "version": 1
  },
  {
    "name": "Order.schema.json",
    "subject": "order",
    "version": 1
  }
]

Configure the JSON Schema serializer to use your oneOf for serialization by setting the following properties in your producer application:
```
auto.register.schemas=false
use.latest.version=true
```
To set the same properties in Java code:
```
props.put(KafkaJsonSchemaSerializerConfig.AUTO_REGISTER_SCHEMAS, "false");
props.put(KafkaJsonSchemaSerializerConfig.USE_LATEST_VERSION, "true");
```
Tip

Setting auto.register.schemas to false turns off auto-registration of the event type so that it doesn’t override the oneOf as the latest schema in the subject. Setting use.latest.version to true causes the JSON Schema serializer to look up the latest schema version in the subject, which is the oneOf, and use that version for serialization. If use.latest.version is false, the serializer looks for the event type in the subject and fails to find it.

For examples and more information about these settings, see Auto schema registration.

For more information about configuring Schema Registry with Kafka Connect, see Configuration Options.

JSON Schema compatibility rules

The JSON Schema compatibility rules are loosely based on similar rules for Avro, but the rules for backward compatibility are more complex. For more information, see the following sections and the blog post Understanding JSON Schema Compatibility.

Primitive type compatibility

JSON Schema has a more limited set of types than Avro. The following rule from Avro also applies to JSON Schema:

A writer’s schema of integer can be promoted to the reader’s schema of number.

Some changes make a JSON primitive type schema less restrictive, which lets a client with the new schema read a JSON document written with the old schema. Examples include:

For string types, the writer’s schema can have a minLength value that is greater than a minLength value in the reader’s schema or not present in the reader’s schema, or a maxLength value that is less than a maxLength value in the reader’s schema or not present in the reader’s schema.
For string types, the writer’s schema can have a pattern value that isn’t present in the reader’s schema.
For number types, the writer’s schema can have a minimum value that is greater than a minimum value in the reader’s schema or not present, or a maximum value that is less than a maximum value in the reader’s schema or not present in the reader’s schema.
For integer types, the writer’s schema can have a multipleOf value that is a multiple of the multipleOf value in the reader’s schema, or that isn’t present in the reader’s schema.

For example, a writer’s schema of integer is compatible with a reader’s schema of number because every integer value the writer produces is also a valid number:

Writer’s schema:

{"type": "integer"}

Reader’s schema:

{"type": "number"}

A string field can also become less restrictive by loosening or dropping a minLength or maxLength constraint. The following change is compatible because every string that satisfied the writer’s range of five to 20 characters also satisfies the reader’s wider range of one to 50 characters:

Writer’s schema:

{"type": "string", "minLength": 5, "maxLength": 20}

Reader’s schema:

{"type": "string", "minLength": 1, "maxLength": 50}

Object compatibility

For object schemas, JSON Schema supports open, closed, and partially open content models.

An open content model allows any number of additional properties to appear in a JSON document without being specified in the JSON schema. Specify additionalProperties as true, which is the default.
A closed content model signals an error if a property appears in the JSON document that isn’t specified in the JSON schema. Specify additionalProperties as false.
A partially open content model allows additional properties to appear in a JSON document without being named in the JSON schema, but restricts the additional properties to a particular type, name, or both. Specify a schema for additionalProperties, or a value for patternProperties that maps regular expressions to schemas.

The following schemas illustrate each content model for the same base object, which has one named property, id:

Open content model. By default, a schema accepts any additional property of any type:

{"type": "object", "properties": {"id": {"type": "string"}}}

Closed content model. No properties beyond id are accepted:

{"type": "object", "properties": {"id": {"type": "string"}}, "additionalProperties": false}

Partially open content model. Any additional property is accepted, but its value must be a string:

{"type": "object", "properties": {"id": {"type": "string"}}, "additionalProperties": {"type": "string"}}

For example, a reader’s schema can add a property, such as myProperty, to the properties of the writer’s schema. This change is backward compatible only if the writer’s schema has a closed content model. If the writer’s schema has an open content model, the writer might have produced JSON documents with myProperty in a different type than the type the reader’s schema expects for myProperty.

The following change is backward compatible. The writer’s closed content model guarantees that no document written with it contains a myProperty value of the wrong type:

Writer’s schema:

{"type": "object", "properties": {"id": {"type": "string"}}, "additionalProperties": false}

Reader’s schema:

{"type": "object", "properties": {"id": {"type": "string"}, "myProperty": {"type": "integer"}}, "additionalProperties": false}

The same change isn’t backward compatible if the writer’s schema has an open content model. The writer could already have produced a document like {"id": "abc", "myProperty": "not a number"}, which the reader’s schema fails to parse:

Writer’s schema:

{"type": "object", "properties": {"id": {"type": "string"}}}

Reader’s schema:

{"type": "object", "properties": {"id": {"type": "string"}, "myProperty": {"type": "integer"}}}

With the notion of content models, you can adapt the Avro rules as follows:

The ordering of fields can differ because fields are matched by name.
Schemas for fields with the same name in both records are resolved recursively.
If the writer’s schema contains a field with a name not present in the reader’s schema, the reader’s schema must have an open content model or a partially open content model that captures the missing field.
If the reader’s schema has a required field that contains a default value, and the writer’s schema has a closed content model and either doesn’t have a field with the same name or has an optional field with the same name, the reader should use the default value from its field.
If the reader’s schema has a required field with no default value, and the writer’s schema either doesn’t have a field with the same name or has an optional field with the same name, an error is signaled.
If the reader’s schema has an optional field, and the writer’s schema has a closed content model and doesn’t have a field with the same name, the reader should ignore the field.

JSON Schema has these additional compatibility rules for objects:

The writer’s schema can have a minProperties value that is greater than the minProperties value in the reader’s schema or that isn’t present in the reader’s schema, or a maxProperties value that is less than the maxProperties value in the reader’s schema or that isn’t present in the reader’s schema.
The writer’s schema can have a required value that is a superset of the required value in the reader’s schema or that isn’t present in the reader’s schema.
The writer’s schema can have a dependencies value that is a superset of the dependencies value in the reader’s schema or that isn’t present in the reader’s schema.
The writer’s schema can have an additionalProperties value of false, whereas it can be true or a schema in the reader’s schema.

Enum compatibility

The Avro rule for enums applies directly to JSON Schema:

If the writer’s symbol isn’t present in the reader’s enum, an error is signaled.

Array compatibility

JSON Schema supports two types of validation for arrays: list validation, where every array element has the same type, and tuple validation, where array elements can have different types.

The resolution algorithm applies recursively to the reader’s and writer’s array item schemas.

JSON Schema has these additional compatibility rules for arrays:

The writer’s schema can have a minItems value that is greater than the minItems value in the reader’s schema or that isn’t present in the reader’s schema, or a maxItems value that is less than the maxItems value in the reader’s schema or that isn’t present in the reader’s schema.
The writer’s schema can have a uniqueItems value of true, whereas it can be false or not present in the reader’s schema.

Union compatibility

JSON Schema implements unions with the oneOf keyword. You can adapt the rules from Avro as follows:

If the reader’s and writer’s schemas are both unions, the writer’s schema must be a subset of the reader’s schema.
If the reader’s schema is a union, but the writer’s isn’t, the first schema in the reader’s union that matches the writer’s schema is recursively resolved against it. If none match, an error is signaled.

Resolve the open content model compatibility error

When you use schema evolution in the Confluent Cloud Console with JSON Schema, you can get the following error when you add an optional field, even if the default compatibility is BACKWARD:

Schema being registered is incompatible with an earlier schema for subject
"test.v6-value", details: [{errorType:"PROPERTY_ADDED_TO_OPEN_CONTENT_MODEL",
description:"The new schema has an open content model and has a property or item at
path '#/properties/dname' which is missing in the old schema'}

The error occurs because a schema has an open content model by default. Any new properties you add to the schema must be compatible with the existing schema, which isn’t always the case.

For example, the following schema, schema 1, has one field named field1:

{
  "properties": {
    "field1": {
      "connect.index": 0,
      "oneOf": [
        {
          "type": "null"
        },
        {
          "type": "string"
        }
      ]
    }
  },
  "title": "Record",
  "type": "object"
}

Adding another field named field2, as in the following schema 2, fails with a compatibility error:

{
  "properties": {
    "field1": {
      "connect.index": 0,
      "oneOf": [
        {
          "type": "null"
        },
        {
          "type": "string"
        }
      ]
    },
    "field2": {
      "connect.index": 1,
      "oneOf": [
        {
          "type": "null"
        },
        {
          "type": "string"
        }
      ]
    }
  },
  "title": "Record",
  "type": "object"
}

Schemas use the open content model by default. As a result, the data {"field1": "100", "field2": 123} isn’t compatible with schema 2. If a schema change is backward compatible, data that is compatible with schema 1 is also compatible with schema 2. That isn’t the case in this example, because field2 is an int, but schema 2 expects field2 to be a string.

To resolve this error, set the additionalProperties: false attribute in the initial schema. This setting ensures that any new properties you add later are compatible. The following version of schema 1 is a closed content model because it adds "additionalProperties": false:

{
  "properties": {
    "field1": {
      "connect.index": 0,
      "oneOf": [
        {
          "type": "null"
        },
        {
          "type": "string"
        }
      ]
    }
  },
  "additionalProperties": false,
  "title": "Record",
  "type": "object"
}

After you add "additionalProperties": false to schema 1, you can add another field in schema 2:

{
  "properties": {
    "field1": {
      "connect.index": 0,
      "oneOf": [
        {
          "type": "null"
        },
        {
          "type": "string"
        }
      ]
    },
    "field2": {
      "connect.index": 1,
      "oneOf": [
        {
          "type": "null"
        },
        {
          "type": "string"
        }
      ]
    }
  },
  "additionalProperties": false,
  "title": "Record",
  "type": "object"
}

Note

Apache Flink® on Confluent Cloud doesn’t support setting additionalProperties: false with the CREATE TABLE statement through hints or the WITH() clause.

JSON Schema Serializer and Deserializer for Schema Registry on Confluent Platform

What is JSON Schema?

Supported JSON Schema specifications

JSON Schema serializer

Use the @Schema annotation on a Java object

Send a JsonNode payload

JSON Schema deserializer

Configure allowed packages

Configure expiry time for client-side schema caches

Test drive JSON Schema

Prerequisites

Create and use schemas

Producer and consumer configuration reference

Add Schema Registry security credentials

Kafka producer configurations

Kafka consumer configurations

Schema Registry configuration options

Use Schema Registry with Connect

Schema references in JSON Schema

Multiple event types in the same topic

JSON Schema compatibility rules

Primitive type compatibility

Object compatibility

Enum compatibility

Array compatibility

Union compatibility

Resolve the open content model compatibility error

Related content