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

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

What is Avro?

Avro is a data serialization framework with rich data structures, a compact binary data format, and schema evolution capabilities. Developed within the Apache Hadoop project, Avro uses JSON-based schemas to define data structures and supports binary and JSON serialization formats. Avro has these key features:

Avro schemas can evolve over time while maintaining backward and forward compatibility. For more information, see Schema Evolution and Compatibility for Schema Registry on Confluent Platform.
Binary encoding provides compact, efficient storage and transmission.
Avro supports many programming languages, including Java, Python, and C++.
Data files include embedded schemas, which makes them self-contained.
You can read and process schemas at runtime without code generation.

The Avro serializer for Confluent Schema Registry doesn’t include the message schema. It includes a magic byte, the schema ID, and then the normal binary encoding of the data itself. 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.

Avro serializer

Plug KafkaAvroSerializer into KafkaProducer to send messages of Avro type to Kafka.

KafkaAvroSerializer supports the following types:

Primitive types: null, Boolean, Integer, Long, Float, Double, String, and byte[]
Complex type: IndexedRecord

Sending data of any other type to KafkaAvroSerializer causes a SerializationException. Typically, IndexedRecord is the type for the value of the Kafka message.

The key of the Kafka message, if used, is often one of the primitive types that KafkaAvroSerializer supports. When you send a message to a topic t, the serializer registers the Avro schemas for the key and the value in Schema Registry under the subjects t-key and t-value, respectively, if the compatibility test passes. The only exception is that Schema Registry never registers the null type.

The following example sends a message to Kafka with a key of type string and a value of type Avro record. The send call can throw a SerializationException if the data isn’t well formed. The example uses the default hostname and port for the Kafka bootstrap server (localhost:9092) and Schema Registry (localhost:8081):

import org.apache.avro.Schema;
import org.apache.avro.generic.GenericData;
import org.apache.avro.generic.GenericRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import java.util.Properties;

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.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
          io.confluent.kafka.serializers.KafkaAvroSerializer.class);
props.put("schema.registry.url", "http://localhost:8081");
KafkaProducer producer = new KafkaProducer(props);

String key = "key1";
String userSchema = "{\"type\":\"record\"," +
                    "\"name\":\"myrecord\"," +
                    "\"fields\":[{\"name\":\"f1\",\"type\":\"string\"}]}";
Schema.Parser parser = new Schema.Parser();
Schema schema = parser.parse(userSchema);
GenericRecord avroRecord = new GenericData.Record(schema);
avroRecord.put("f1", "value1");

ProducerRecord<Object, Object> record = new ProducerRecord<>("topic1", key, avroRecord);
try {
  producer.send(record);
} catch(SerializationException e) {
  // may need to do something with it
}
// When you're finished producing records, you can flush the producer to ensure it has all been written to Kafka and
// then close the producer to free its resources.
finally {
  producer.flush();
  producer.close();
}

Code that the avro-maven-plugin generates adds Java-specific properties, such as "avro.java.string":"String", that can prevent schema evolution. To override this behavior, set avro.remove.java.properties=true in the Avro serializer configuration.

Avro deserializer

Plug KafkaAvroDeserializer into KafkaConsumer to receive messages of any Avro type from Kafka.

The following example receives messages from Kafka with a key of type string and a value of type Avro record. Getting the message key or value can throw a SerializationException if the data isn’t well formed. The example uses the default hostname and port for the Kafka bootstrap server (localhost:9092) and Schema Registry (localhost:8081):

import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;

import org.apache.avro.generic.GenericRecord;

import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.Properties;
import java.util.Random;

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.KafkaAvroDeserializer");
props.put("schema.registry.url", "http://localhost:8081");

props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");

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

try {
  while (true) {
    ConsumerRecords<String, GenericRecord> records = consumer.poll(100);
    for (ConsumerRecord<String, GenericRecord> record : records) {
      System.out.printf("offset = %d, key = %s, value = %s \n", record.offset(), record.key(), record.value());
    }
  }
} finally {
  consumer.close();
}

Avro doesn’t need a property to specify a specific type, because the deserializer derives the type from the Avro schema, using the namespace and name of the Avro type. The Avro deserializer works by default with topics that have records of heterogeneous Avro types, such as when you use RecordNameStrategy or TopicRecordNameStrategy to store multiple types in the same topic. For more information, see Martin Kleppmann’s blog post Should You Put Several Event Types in the Same Kafka Topic?.

An alternative is to use schema references. For more information, see Multiple event types in the same topic and the blog post Putting Several Event Types in the Same Topic – Revisited.

Unlike Avro, the Protobuf and JSON Schema deserializers require a specific property to return a specific type rather than a generic type.

To return a specific type in Avro, add the following configuration:

props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);

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 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 Avro schema

Test drive the Avro schema format by using the producer and consumer commands in a shell to send and receive Avro data in JSON format. The producer and consumer use AvroMessageFormatter and AvroMessageReader to convert between Avro and JSON.

Avro defines a binary serialization format and a JSON serialization format. Use JSON when you want human-readable data, and use the more efficient binary format to store data in topics.

The command line producer and consumer show how the built-in Avro 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 an Avro schema, add a second version with a new field, 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. 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.
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 Avro records in JSON as the message value.

The following command starts a producer, creates the transactions-avro topic if it doesn’t already exist, and creates a schema for the topic with two fields, id and amount:
```
kafka-avro-console-producer --bootstrap-server localhost:9092 --property schema.registry.url=http://localhost:8081 --topic transactions-avro \
--property value.schema='{"type":"record","name":"Transaction","fields":[{"name":"id","type":"string"},{"name": "amount", "type": "double"}]}'
```
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-avro topic, run the consumer in a new terminal window:

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

The console shows the following output:

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

Leave this consumer running.

Register a new schema version under the same subject by adding a new field, customer_id.

Because the default subject-level compatibility is BACKWARD, give the new field a default value so that the schema is compatible with the previous version.

In a new terminal window, run the following command:

kafka-avro-console-producer --bootstrap-server localhost:9092 --property schema.registry.url=http://localhost:8081 --topic transactions-avro \
--property value.schema='{"type": "record","name": "Transaction","fields": [{"name": "id", "type": "string"},{"name": "amount", "type": "double"},{"name": "customer_id", "type": "string", "default":"null"}]}'

Type the following message into your producer, and press Enter:
```
{ "id":"1001", "amount":500, "customer_id":"1221" }
```
Return to your running consumer to read the new message from the transactions-avro topic.

The new output appears after the original output:
```
{"id":"1000","amount":500.0}
{"id":"1001","amount":500.0,"customer_id":"1221"}
```
If you closed the original consumer, restart it with the same command as in step 5.

To examine the schemas registered with Schema Registry, use curl in another shell:

curl --silent http://localhost:8081/subjects/transactions-avro-value/versions/1/schema

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

{"type":"record","name":"Transaction","fields":[{"name":"id","type":"string"},{"name":"amount","type":"double"}]}

To view version 2, run the following command:

curl --silent http://localhost:8081/subjects/transactions-avro-value/versions/2/schema

The output for version 2 includes the customer_id field:

{"type":"record","name":"Transaction","fields":[{"name":"id","type":"string"},{"name":"amount","type":"double"},{"name":"customer_id","type":"string","default":"null"}]}

View the schema in more detail with the following command, which pipes the output through jq and suppresses the curl download messages for readable output:

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

Your output should resemble:

"subject": "transactions-avro-value",
"version": 2,
"id": 2,
"schema": "{\"type\":\"record\",\"name\":\"Transaction\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"},{\"name\":\"customer_id\",\"type\":\"string\",\"default\":\"null\"}]}"

Use Confluent Control Center to examine schemas and messages.

Messages that were successfully produced also show on Control Center (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-avro --cluster $CLUSTER_ID

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

{
   "type": "record",
   "name": "Transaction",
   "fields": [
       {"name": "id", "type": "string"},
       {"name": "amount", "type": "double"}
   ]
 }

Run the following command to create a producer that uses the schema from the previous step:
```
confluent kafka topic produce transactions-avro \
--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 "avro"
```
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-avro topic, run a consumer in another terminal:

confluent kafka topic consume transactions-avro \
--cluster $CLUSTER_ID \
--from-beginning \
--value-format "avro" \
--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}

Register a new schema version under the same subject by adding a new field, customer_id.

Because the default subject-level compatibility is BACKWARD, give the new field a default value so that the schema is compatible with the previous version.

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

{
 "type": "record",
 "name": "Transaction",
 "fields": [
     {"name": "id", "type": "string"},
     {"name": "amount", "type": "double"},
     {"name": "customer_id", "type": "string", "default":"null"}
 ]
}

In another terminal, run the following command:

confluent kafka topic produce transactions-avro \
--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 "avro"

Your output should resemble:

Successfully registered schema with ID 100002
Starting Kafka Producer. Use Ctrl-C or Ctrl-D to exit.

Type the following message into your producer, and press Enter:
```
{ "id":"1001", "amount":500, "customer_id":"1221" }
```
Switch to the terminal with your running consumer to read the new message from the transactions-avro topic.

The new output appears after the original output:
```
{"id":"1000","amount":500.0}
{"id":"1001","amount":500.0,"customer_id":"1221"}
```
If you closed the original consumer, restart it with the same command as in step 5.

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-avro-value --version 1

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

Schema ID: 100001
Schema: {"type":"record","name":"Transaction","fields":[{"name":"id","type":"string"},{"name":"amount","type":"double"}]}

To view version 2, run the following command:

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

The output for version 2 includes the customer_id field:

Schema ID: 100002
Schema: {"type":"record","name":"Transaction","fields":[{"name":"id","type":"string"},{"name":"amount","type":"double"},{"name":"customer_id","type":"string","default":"null"}]}

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-avro-console-producer --bootstrap-server localhost:9092 \
--property schema.registry.url=http://localhost:8081 --topic transactions-avro \
--property value.schema='{"type":"record","name":"Transaction","fields":[{"name":"id","type":"string"},{"name": "amount", "type": "double"}]}' \
--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 Avro

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.

For more information, see the following:

Multiple event types in the same topic for an example of schema references.
The blog post Putting Several Event Types in the Same Topic – Revisited for further detail.
POST /subjects/(string: subject)/versions for an API example that registers a new schema with a referenced schema.

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 Avro:

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 an Avro union to define the schema references as a list of schema names, such as:

[
  "io.confluent.examples.avro.Customer",
  "io.confluent.examples.avro.Product",
  "io.confluent.examples.avro.Order"
]

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

[
  {
    "name": "io.confluent.examples.avro.Customer",
    "subject": "customer",
    "version": 1
  },
  {
    "name": "io.confluent.examples.avro.Product",
    "subject": "product",
    "version": 1
  },
  {
    "name": "io.confluent.examples.avro.Order",
    "subject": "order",
    "version": 1
  }
]

Configure the Avro serializer to use your Avro union 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(AbstractKafkaSchemaSerDeConfig.AUTO_REGISTER_SCHEMAS, false);
props.put(AbstractKafkaSchemaSerDeConfig.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 union as the latest schema in the subject. Setting use.latest.version to true causes the Avro serializer to look up the latest schema version in the subject, which is the union, 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 Connect, see Configuration Options for Kafka Connect.

Limitations for librdkafka clients

librdkafka clients don’t support Avro unions in serialization or deserialization. As a workaround, reference the schema as a field for each type in the union. Fields can contain referenced types. For example, convert a union such as ["typeA", "typeB"] to a schema that has a field for each element of the union, represented as:

{ "type": "A", "typeA": <typeA instance> }, { "type": "B", "typeB": <typeB instance> }

Describe type A as follows:

{
   "type": "A",
   "typeA": <typeA instance>,
   "typeB": null
}

Describe type B as follows:

{
   "type": "B",
   "typeA": null,
   "typeB": <typeB instance>
}

For more information about clients, see:

Reflection-based Avro serializer and deserializer

Confluent Platform includes ReflectionAvroSerializer and ReflectionAvroDeserializer to write and read data in reflection Avro format.

The serializer writes data in the wire format, and the deserializer reads data in the same format.

The serde for the reflection-based Avro serializer and deserializer is ReflectionAvroSerde.

For more information, see Kafka Streams Data Types and Serialization.

Allow null values with Avro reflection

Set avro.reflection.allow.null to true to allow null values in fields when you use the Reflection-based Avro serializer and deserializer. The default is false.

Avro schema compatibility rules

The compatibility rules for Avro are in the Avro specification under Schema Resolution. For more information, see Schema Evolution and Compatibility for Schema Registry on Confluent Platform and Compatibility checks in the overview.

Per the specification, extending an enum value follows the same rules as adding a field, and default values are forward-compatible.

If you add a new value to the enum array, any existing JSON instance that was valid against the old schema, using one of the old enum values, is still valid against the new schema. Adding a value is a compatible change for validation purposes. New instances can use the new value and are still valid.
If you remove a value from the enum array, any existing JSON instance that used that removed value is no longer valid against the new schema. Removing a value is a breaking change for validation.

Differences between Avro specific and Avro generic records

In Kafka applications, you can handle Avro records as generic records or as specific records.

Generic and specific records overview

Avro generic records take a dynamic, runtime-based approach to working with data. You don’t need to generate Java classes from your schema ahead of time. You work with the GenericRecord interface and access fields using string-based keys at runtime.
Avro specific records use code generation to create strongly typed Java classes from your Avro schemas. The generated classes add compile-time type safety and IDE autocompletion.

Key differences between generic and specific records

The following table summarizes the main differences between Avro generic and Avro specific records:

Feature	Generic records	Specific records
Code generation	Not required	Required, using `avro-maven-plugin` or similar
Type safety	Runtime checking only	Compile-time type checking
Schema handling	Schema parsed at runtime	Schema embedded in generated class
Field access	String-based: `record.get("fieldName")`	Method-based: `record.getFieldName()`
Development speed	Faster for prototyping and dynamic schemas	Slower initial setup, faster long-term development
Refactoring	Error-prone because field names are strings	IDE-supported, safer refactoring
Schema evolution	Requires manual field checking	Generated classes handle defaults automatically
Performance	Slightly slower because of field lookup by name	Slightly faster because of direct method calls
Flexibility	Works with any schema at runtime	Tied to specific schema versions

Use Avro generic records

Generic records are ideal when:

You need to process data with schemas that aren’t known at compile time
You build schema-agnostic tools or frameworks, such as data transformation pipelines
You want rapid prototyping without code generation steps
You need to handle different schema versions dynamically

Avro generic record producer

To use generic records, define or load the schema manually and use the GenericRecord interface. The following producer defines the schema inline and sends a generic record:

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
props.put("schema.registry.url", "http://localhost:8081");

Schema schema = new Schema.Parser().parse(
      "{\"type\":\"record\",\"name\":\"User\",\"namespace\":\"com.example\","
      + "\"fields\":[{\"name\":\"name\",\"type\":\"string\"},{\"name\":\"age\",\"type\":\"int\"}]}");

try (KafkaProducer<String, GenericRecord> producer = new KafkaProducer<>(props)) {
      GenericRecord user = new GenericData.Record(schema);
      user.put("name", "Alice");
      user.put("age", 30);
      producer.send(new ProducerRecord<>("users", "key1", user));
}

Avro generic record consumer

When you consume generic records, access fields with string lookups, which requires manual casting. The following consumer reads generic records and prints two fields:

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
props.put("schema.registry.url", "http://localhost:8081");
// GenericRecord is the default because specific.avro.reader defaults to false

try (KafkaConsumer<String, GenericRecord> consumer = new KafkaConsumer<>(props)) {
      consumer.subscribe(List.of("users"));
      while (true) {
         for (ConsumerRecord<String, GenericRecord> record : consumer.poll(Duration.ofMillis(100))) {
            GenericRecord user = record.value();
            System.out.println(user.get("name") + " " + user.get("age"));
         }
      }
}

Use Avro specific records

Specific records are ideal when:

You need compile-time type safety and want to catch schema-related errors early
Your schemas are stable and well defined
You want IDE support for autocompletion and refactoring
You build production applications with strict type requirements
You need better documentation through generated code

Avro specific record class

The following schema generates a User class:

{"type": "record", "name": "User", "namespace": "com.example",
"fields": [{"name": "name", "type": "string"}, {"name": "age", "type": "int"}]}

Avro specific record producer

The following producer sends a specific record:

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
props.put("schema.registry.url", "http://localhost:8081");

try (KafkaProducer<String, User> producer = new KafkaProducer<>(props)) {
      User user = User.newBuilder().setName("Alice").setAge(30).build();
      producer.send(new ProducerRecord<>("users", "key1", user));
}

Avro specific record consumer

The following consumer reads specific records. Specific records require specific.avro.reader set to true. For more information, see Avro deserializer.

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG, "my-group");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
props.put("schema.registry.url", "http://localhost:8081");
props.put("specific.avro.reader", true);  // required for SpecificRecord

try (KafkaConsumer<String, User> consumer = new KafkaConsumer<>(props)) {
      consumer.subscribe(List.of("users"));
      while (true) {
         for (ConsumerRecord<String, User> record : consumer.poll(Duration.ofMillis(100))) {
            User user = record.value();
            System.out.println(user.getName() + " " + user.getAge());
         }
      }
}

Schema evolution considerations

Generic and specific records support schema evolution, with these differences:

With generic records, you must manually check for field existence and handle missing fields when consuming data with evolved schemas.
With specific records, generated classes include default values from the schema, which makes it easier to handle schema evolution. You must regenerate the classes when schemas change.

Best practices for generic and specific records

Follow these best practices when you work with Avro records:

Use specific records for production applications because compile-time type safety and better maintainability outweigh the initial code generation setup cost.
Use generic records for tooling and frameworks when you build tools that process arbitrary schemas, such as data migration tools, monitoring applications, or generic extract, transform, load (ETL) pipelines. Generic records offer the necessary flexibility.
Configure avro.remove.java.properties when you use specific records. Setting it to true in your serializer configuration removes Java-specific properties that can affect schema evolution. For more information, see Avro serializer.
Be consistent within an application. Mixing generic and specific records in the same application can lead to confusion. Choose one approach and use it consistently.
Plan for schema evolution. Regardless of which approach you choose, design your schemas with evolution in mind from the start. Add default values to fields and follow the compatibility rules.