Kafka Schema Registry & Avro Serialization

Kafka messages are opaque bytes by default: producers and consumers must agree on the format out-of-band. The Confluent Schema Registry solves this by storing versioned Avro (or JSON Schema / Protobuf) schemas centrally. Producers embed only a schema ID in each message; consumers retrieve the schema by ID and deserialize correctly. Schema evolution rules enforce backward/forward compatibility automatically.

Share

Contents

How Schema Registry Works

The registry runs as a separate service (default port 8081). When a producer serializes a message it looks up or registers the schema, receiving a numeric ID. That ID is prefixed to the payload bytes. The consumer reads the ID, fetches the schema, and deserializes.

Java
/*
  Wire format of an Avro-serialized Kafka message value:

  Byte 0     : Magic byte  (always 0x00)
  Bytes 1–4  : Schema ID   (4-byte big-endian int)
  Bytes 5+   : Avro-encoded payload

  The KafkaAvroSerializer / KafkaAvroDeserializer handle
  this framing automatically when schema.registry.url is set.
*/

Defining an Avro Schema

Avro schemas are JSON documents. Place them in src/main/avro/ and use the Avro Maven plugin to generate Java classes.

JSON
{
  "type": "record",
  "name": "Order",
  "namespace": "com.example.avro",
  "fields": [
    { "name": "id",       "type": "string" },
    { "name": "product",  "type": "string" },
    { "name": "quantity", "type": "int" },
    { "name": "status",   "type": { "type": "enum", "name": "OrderStatus",
                                    "symbols": ["PENDING","PROCESSING","COMPLETED","CANCELLED"] } },
    { "name": "createdAt","type": "long", "logicalType": "timestamp-millis" }
  ]
}
XML
<!-- Maven plugin to generate Java classes from .avsc files -->
<plugin>
    <groupId>org.apache.avro</groupId>
    <artifactId>avro-maven-plugin</artifactId>
    <version>1.11.3</version>
    <executions>
        <execution>
            <phase>generate-sources</phase>
            <goals><goal>schema</goal></goals>
            <configuration>
                <sourceDirectory>${project.basedir}/src/main/avro/</sourceDirectory>
                <outputDirectory>${project.build.directory}/generated-sources/avro/</outputDirectory>
            </configuration>
        </execution>
    </executions>
</plugin>

Maven Dependencies

Add Kafka clients plus the Confluent Avro serializer. The Confluent artifacts are in the Confluent Maven repository.

XML
<repositories>
    <repository>
        <id>confluent</id>
        <url>https://packages.confluent.io/maven/</url>
    </repository>
</repositories>

<dependencies>
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>3.7.0</version>
    </dependency>
    <dependency>
        <groupId>io.confluent</groupId>
        <artifactId>kafka-avro-serializer</artifactId>
        <version>7.6.0</version>
    </dependency>
    <dependency>
        <groupId>org.apache.avro</groupId>
        <artifactId>avro</artifactId>
        <version>1.11.3</version>
    </dependency>
</dependencies>

Avro Producer

Configure the producer to use KafkaAvroSerializer for both key and value, pointing at the Schema Registry URL.

Java
import io.confluent.kafka.serializers.KafkaAvroSerializer;
import io.confluent.kafka.serializers.AbstractKafkaSchemaSerDeConfig;
import com.example.avro.Order;
import com.example.avro.OrderStatus;

Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,  "localhost:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,   KafkaAvroSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, KafkaAvroSerializer.class);
props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");

try (KafkaProducer<String, Order> producer = new KafkaProducer<>(props)) {

    Order order = Order.newBuilder()
        .setId("ORD-001")
        .setProduct("Laptop")
        .setQuantity(1)
        .setStatus(OrderStatus.PENDING)
        .setCreatedAt(System.currentTimeMillis())
        .build();

    ProducerRecord<String, Order> record = new ProducerRecord<>("orders", order.getId().toString(), order);
    producer.send(record, (metadata, ex) -> {
        if (ex != null) ex.printStackTrace();
        else System.out.printf("Sent to partition=%d offset=%d%n",
                               metadata.partition(), metadata.offset());
    });
}

Avro Consumer

The consumer uses KafkaAvroDeserializer. Setting specific.avro.reader=true returns the generated Java class instead of the generic GenericRecord.

Java
import io.confluent.kafka.serializers.KafkaAvroDeserializer;
import io.confluent.kafka.serializers.KafkaAvroDeserializerConfig;

Properties props = new Properties();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,     "localhost:9092");
props.put(ConsumerConfig.GROUP_ID_CONFIG,              "order-consumer-group");
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,     "earliest");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,   KafkaAvroDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, KafkaAvroDeserializer.class);
props.put(AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, "http://localhost:8081");
// return Order, not GenericRecord
props.put(KafkaAvroDeserializerConfig.SPECIFIC_AVRO_READER_CONFIG, true);

try (KafkaConsumer<String, Order> consumer = new KafkaConsumer<>(props)) {
    consumer.subscribe(List.of("orders"));
    while (true) {
        ConsumerRecords<String, Order> records = consumer.poll(Duration.ofMillis(100));
        for (ConsumerRecord<String, Order> record : records) {
            Order order = record.value();
            System.out.printf("Received order %s: %s x%d%n",
                              order.getId(), order.getProduct(), order.getQuantity());
        }
    }
}

Avro with Spring Kafka

In a Spring Boot project, configure the Avro serializers in application.properties and use @KafkaListener with typed parameters as usual.

Properties
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
spring.kafka.producer.value-serializer=io.confluent.kafka.serializers.KafkaAvroSerializer
spring.kafka.consumer.key-deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
spring.kafka.consumer.value-deserializer=io.confluent.kafka.serializers.KafkaAvroDeserializer
spring.kafka.consumer.group-id=order-group
spring.kafka.consumer.properties.schema.registry.url=http://localhost:8081
spring.kafka.consumer.properties.specific.avro.reader=true
spring.kafka.producer.properties.schema.registry.url=http://localhost:8081
Java
@Service
public class OrderConsumer {

    @KafkaListener(topics = "orders")
    public void consume(Order order) {  // Spring Kafka deserializes directly to Order
        System.out.println("Processing: " + order.getId() + " — " + order.getProduct());
    }
}

Schema Evolution & Compatibility

The Schema Registry enforces compatibility rules before registering a new schema version. The default mode is BACKWARD: new consumers can read old messages.

Java
/*
  Compatibility modes:
  ─────────────────────────────────────────────────────────────────────────
  BACKWARD  (default) — new schema reads old messages
    • Safe to: add optional fields with defaults, remove fields
    • Unsafe: add required fields (no default), change field types

  FORWARD   — old schema reads new messages
    • Safe to: remove optional fields, add fields
    • Unsafe: remove required fields

  FULL      — both BACKWARD and FORWARD simultaneously

  NONE      — no compatibility check; use during development only
  ─────────────────────────────────────────────────────────────────────────
*/

// Adding a new optional field — BACKWARD compatible
{
  "name": "discount",
  "type": ["null", "double"],   // union with null = optional
  "default": null
}

// Removing a field — NOT backward compatible if consumers use that field
// Always add new fields with defaults; never remove required fields.
Changing a field's type (e.g. int → string) is never compatible under any mode. If you need to change a type, add a new field with a different name alongside the old one and migrate consumers before removing the old field.

Related articles

All Kafka articles →
Schema Evolution: Avro Compatibility Modes with Schema RegistryKafka Schema Evolution: Avro compatibility modes (BACKWARD, FORWARD, FULL, TRANSITIVE) with Schema Registry, Java integration, and REST API examples.Kafka Message Serialization: String, JSON & AvroKafka message serialization and deserialization: String, JSON (Jackson), and Avro serializers with Java examples.Kafka Producer & Consumer ConfigurationEssential Kafka producer and consumer configuration: acks, retries, batch.size, linger.ms, compression, and tuning for throughput vs reliability.Kafka - ProducersKafka Producer: API, acks, retries, idempotent producer configuration with Java code examples.