05 - Schema Registry: Avro, Protobuf, Compatibility and Schema Evolution

The day someone renames user_id to userId

Twenty teams write to fifty topics. One day a team changes the field user_id to userId because "it is more consistent with the rest of the code", and the consumers that read that topic do not break with an error: they keep running and produce wrong data, because the field they look for is always empty. Kafka is not to blame: it carries bytes, it does not know what they mean. You need a contract on the shape of the messages, checked by a machine at the moment someone tries to change it.

Confluent's Schema Registry does three things. It keeps the schemas, each with a numeric identifier. It has that identifier written in the first bytes of every message. And, when you register a new version of a schema, it compares it with the previous ones according to a compatibility rule that you choose. In this article you see the three things with code that was run: real Avro (library avro-1.12.2.jar, SHA-1 verified) for serialization and for the compatibility check, and Protobuf (protobuf-java 3.25.5) for the field numbers. The Registry itself is not there: in the listings an in-memory map replaces it, and I say so where needed. The rules and settings come from the Confluent documentation "Schema Evolution and Compatibility" and "Formats, Serializers, and Deserializers", consulted on 2026-11-12.

How the right schema reaches the consumer: the message format

The serializer does not write the schema in every message (that would be a waste) but a reference. The documentation describes the wire format: byte 0 is the format version (it is 0), bytes 1 to 4 are the schema identifier as a 4-byte integer in big-endian order, and from byte 5 onwards comes the serialized data (for Protobuf, after the ID an array of message-type indexes is added). The consumer reads the ID, asks the Registry for the schema (and caches it), and decodes with it. The Registry uses a Kafka topic, by default _schemas, with a single partition, as the log on which it writes schemas, metadata and compatibility settings.

import org.apache.avro.Schema;
import org.apache.avro.generic.*;
import org.apache.avro.io.*;
import java.io.*;
import java.nio.ByteBuffer;
import java.util.*;

public class WireFormat {
    static final Map<Integer, Schema> registry = new HashMap<>();   // Registry model: id -> schema

    static byte[] serialize(int schemaId, Schema schema, GenericRecord r) throws IOException {
        ByteArrayOutputStream out = new ByteArrayOutputStream();
        out.write(0);                                               // byte 0: format version
        out.write(ByteBuffer.allocate(4).putInt(schemaId).array()); // byte 1-4: schema id, big-endian
        BinaryEncoder enc = EncoderFactory.get().binaryEncoder(out, null);
        new GenericDatumWriter<GenericRecord>(schema).write(r, enc);
        enc.flush();
        return out.toByteArray();
    }

    static GenericRecord deserialize(byte[] msg, Schema reader) throws IOException {
        ByteBuffer b = ByteBuffer.wrap(msg);
        b.get();                                                    // skip the version
        Schema writer = registry.get(b.getInt());                    // the schema it was written with
        BinaryDecoder dec = DecoderFactory.get().binaryDecoder(msg, 5, msg.length - 5, null);
        return new GenericDatumReader<GenericRecord>(writer, reader).read(null, dec);
    }

    static String hex(byte[] m, int n) {
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < n; i++) sb.append(String.format("%02x ", m[i]));
        return sb.toString().trim();
    }

    public static void main(String[] args) throws IOException {
        String base = "{\"name\":\"order_id\",\"type\":\"string\"},{\"name\":\"amount\",\"type\":\"double\"}";
        Schema v1 = new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Order\",\"fields\":[" + base + "]}");
        Schema v2 = new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Order\",\"fields\":[" + base
            + ",{\"name\":\"coupon\",\"type\":[\"null\",\"string\"],\"default\":null}]}");
        registry.put(1, v1);
        registry.put(2, v2);

        GenericRecord old = new GenericData.Record(v1);
        old.put("order_id", "A-1");
        old.put("amount", 19.9);
        byte[] m1 = serialize(1, v1, old);
        System.out.println("message v1: " + m1.length + " bytes, prefix " + hex(m1, 5));
        System.out.println("read with schema v2: " + deserialize(m1, v2));

        GenericRecord fresh = new GenericData.Record(v2);
        fresh.put("order_id", "A-2");
        fresh.put("amount", 5.0);
        fresh.put("coupon", "SAVE5");
        byte[] m2 = serialize(2, v2, fresh);
        System.out.println("message v2: " + m2.length + " bytes, prefix " + hex(m2, 5));
        System.out.println("read with schema v1: " + deserialize(m2, v1));
    }
}

// Output:
// message v1: 17 bytes, prefix 00 00 00 00 01
// read with schema v2: {"order_id": "A-1", "amount": 19.9, "coupon": null}
// message v2: 24 bytes, prefix 00 00 00 00 02
// read with schema v1: {"order_id": "A-2", "amount": 5.0}

The serialize method writes exactly the prefix of the documentation: a zero, then the ID in 4 bytes. The first message weighs 17 bytes with prefix 00 00 00 00 01; the second, which has the coupon field, weighs 24 with prefix 00 00 00 00 02. The interesting part is the reading: deserialize retrieves from the registry the schema with which the message was written and sets it beside the schema with which the consumer wants to read. With schema v2 the old message produces coupon: null, the default value; with schema v1 the new message loses coupon, which the reader does not know. This is Avro's schema resolution, and it is what the real consumer does for you. The real Registry adds the cache, the network, authentication and the compatibility check, which the model leaves out.

The compatibility rules, verified with Avro

The Confluent documentation defines the levels like this: BACKWARD (the default) means that consumers using the new schema can read data written with the old one; FORWARD that consumers with the old schema read data written with the new one; FULL both. The TRANSITIVE variants compare with all the previous versions and not only the last. The reason BACKWARD is the default, according to the same documentation, is that it allows you to rewind consumers to the beginning of the topic and reread old data with the new schema.

Definitions are fixed better by trying them. The listing uses the compatibility check of the Avro library, with the same direction criterion: BACKWARD means "reader = new schema, writer = old schema", FORWARD the opposite. Six typical changes to the Order schema:

import org.apache.avro.Schema;
import org.apache.avro.SchemaCompatibility;
import org.apache.avro.SchemaCompatibility.SchemaCompatibilityType;

public class AvroCompatibility {
    static Schema schema(String fields) {
        return new Schema.Parser().parse("{\"type\":\"record\",\"name\":\"Order\",\"fields\":[" + fields + "]}");
    }

    // BACKWARD: the new schema reads data written with the old one. FORWARD: the old one reads data written with the new one.
    static boolean reads(Schema reader, Schema writer) {
        return SchemaCompatibility.checkReaderWriterCompatibility(reader, writer).getType()
            == SchemaCompatibilityType.COMPATIBLE;
    }

    static void check(String change, Schema old, Schema updated) {
        boolean bw = reads(updated, old), fw = reads(old, updated);
        System.out.printf("%-42s BACKWARD=%-5s FORWARD=%-5s FULL=%s%n", change, bw, fw, bw && fw);
    }

    public static void main(String[] args) {
        String id = "{\"name\":\"order_id\",\"type\":\"string\"}";
        String amountField = "{\"name\":\"amount\",\"type\":\"double\"}";
        String currencyField = "{\"name\":\"currency\",\"type\":\"string\",\"default\":\"EUR\"}";
        Schema v1 = schema(id + "," + amountField + "," + currencyField);

        check("add field with default", v1,
            schema(id + "," + amountField + "," + currencyField + ",{\"name\":\"coupon\",\"type\":[\"null\",\"string\"],\"default\":null}"));
        check("add field WITHOUT default", v1,
            schema(id + "," + amountField + "," + currencyField + ",{\"name\":\"note\",\"type\":\"string\"}"));
        check("remove a field that had a default", v1, schema(id + "," + amountField));
        check("remove a field without default", v1, schema(id + "," + currencyField));
        check("amount from double to string", v1,
            schema(id + ",{\"name\":\"amount\",\"type\":\"string\"}," + currencyField));
        check("amount from int to long (promotion)",
            schema(id + ",{\"name\":\"amount\",\"type\":\"int\"}," + currencyField),
            schema(id + ",{\"name\":\"amount\",\"type\":\"long\"}," + currencyField));
    }
}

// Output:
// add field with default                     BACKWARD=true  FORWARD=true  FULL=true
// add field WITHOUT default                  BACKWARD=false FORWARD=true  FULL=false
// remove a field that had a default          BACKWARD=true  FORWARD=true  FULL=true
// remove a field without default             BACKWARD=true  FORWARD=false FULL=false
// amount from double to string               BACKWARD=false FORWARD=false FULL=false
// amount from int to long (promotion)        BACKWARD=true  FORWARD=false FULL=false

Read them line by line. Adding a field with a default is compatible in all directions; without a default it breaks BACKWARD, because the new reader would not know what value to give the field missing from the old data. Removing a field that had a default is compatible in all directions; removing a field without a default passes BACKWARD (the new reader ignores the extra data) but breaks FORWARD (the old reader looks for a field that is not there and has no fallback value). Changing amount from double to string breaks everywhere. The promotion from int to long is the only type change that passes, and only in BACKWARD, because a 64-bit reader understands a 32-bit integer but not the other way round. The table in the Confluent documentation for Avro is consistent with these results; the Avro library, however, is not the Registry, and I did not run the comparison with the real Registry's behaviour on a cluster.

From these rules follows the update order. With BACKWARD you update the consumers first and then the producers: the new consumer reads both old and new data. With FORWARD the producers are updated first, and the documentation adds a condition: the data produced with the old schemas must no longer be readable by consumers. With FULL you do not have to decide, but you are limited to optional fields, that is, with a default. For Kafka Streams the documentation admits only the FULL, TRANSITIVE and BACKWARD levels. The NONE level switches the check off and must be kept out of production.

Protobuf: compatibility lives in the numbers

In Protobuf a field is identified on the wire by its number, not its name. Adding a field with a new number is safe; renaming it changes nothing for whoever reads the bytes; reusing a number for something else is the trouble. The model builds three descriptors of the same Order message with the official Protobuf library and passes the bytes between different versions.

import com.google.protobuf.*;
import com.google.protobuf.DescriptorProtos.*;
import com.google.protobuf.DescriptorProtos.FieldDescriptorProto.Type;
import com.google.protobuf.Descriptors.Descriptor;

public class ProtobufNumbers {
    // Builds an Order message on the fly with the given fields (name, number, type).
    static Descriptor order(Object[][] fields) throws Exception {
        DescriptorProto.Builder msg = DescriptorProto.newBuilder().setName("Order");
        for (Object[] c : fields)
            msg.addField(FieldDescriptorProto.newBuilder().setName((String) c[0]).setNumber((Integer) c[1])
                .setType((Type) c[2]).setLabel(FieldDescriptorProto.Label.LABEL_OPTIONAL));
        FileDescriptorProto file = FileDescriptorProto.newBuilder().setName("order.proto").setSyntax("proto3")
            .addMessageType(msg).build();
        return Descriptors.FileDescriptor.buildFrom(file, new Descriptors.FileDescriptor[0]).findMessageTypeByName("Order");
    }

    public static void main(String[] args) throws Exception {
        Descriptor v1 = order(new Object[][] {{"order_id", 1, Type.TYPE_STRING}, {"amount", 3, Type.TYPE_DOUBLE}});
        Descriptor v2 = order(new Object[][] {{"order_id", 1, Type.TYPE_STRING}, {"amount", 3, Type.TYPE_DOUBLE},
                                               {"currency", 4, Type.TYPE_STRING}});
        Descriptor v3 = order(new Object[][] {{"order_id", 1, Type.TYPE_STRING}, {"amount", 3, Type.TYPE_STRING}});

        byte[] writtenV1 = DynamicMessage.newBuilder(v1)
            .setField(v1.findFieldByName("order_id"), "A-1").setField(v1.findFieldByName("amount"), 19.9).build().toByteArray();
        byte[] writtenV2 = DynamicMessage.newBuilder(v2)
            .setField(v2.findFieldByName("order_id"), "A-2").setField(v2.findFieldByName("amount"), 5.0)
            .setField(v2.findFieldByName("currency"), "EUR").build().toByteArray();

        DynamicMessage a = DynamicMessage.parseFrom(v2, writtenV1);
        System.out.println("v1 data read with v2: currency=\"" + a.getField(v2.findFieldByName("currency")) + "\" (default value)");

        DynamicMessage b = DynamicMessage.parseFrom(v1, writtenV2);
        System.out.println("v2 data read with v1: unknown fields preserved = " + b.getUnknownFields().asMap().keySet());

        DynamicMessage c = DynamicMessage.parseFrom(v3, writtenV1);
        System.out.println("number 3 reused with another type: amount=\"" + c.getField(v3.findFieldByName("amount"))
            + "\", unknown fields = " + c.getUnknownFields().asMap().keySet());
    }
}

// Output:
// v1 data read with v2: currency="" (default value)
// v2 data read with v1: unknown fields preserved = [4]
// number 3 reused with another type: amount="", unknown fields = [3]

In the first line the data written with v1 read by v2 gives currency="", the default value of a string. In the second the v2 data read by v1 does no harm: field 4, which v1 does not know, is preserved among the "unknown fields". In the third, number 3, which was a double, was reused for a string: the data is not rejected, amount turns out empty and the value ends up among the unknown fields. No exception, no warning, while the language guide says clearly that "field numbers should never be reused". It is the silent error that the Registry can block only if compatibility is on. For Protobuf the Confluent documentation recommends using BACKWARD_TRANSITIVE, because adding new message types is not forward compatible.

Avro, Protobuf or JSON Schema: how to choose

The table does not say "the best": it says where the contract lives in each format and where you go wrong without noticing. The most solid choice follows a simple criterion: use the format your teams already know and your toolchain already reads, and put compatibility into practice before the format. I give no figures on the byte savings among the three formats: it depends on your data, and the only way to know is to serialize a real sample with all three.

Governance: names, registration and checking in CI

The default subject naming strategy is TopicNameStrategy: the subject is <topic>-value for values and <topic>-key for keys (value.subject.name.strategy, key.subject.name.strategy). With this choice a schema lives with its topic, which is fine as long as a topic holds a single type of event. To mix several types in the same topic the documentation says to switch off automatic registration and to use references between schemas.

On automatic registration the documentation is explicit: in production it pays to switch off auto.register.schemas and switch on use.latest.version, to avoid "Schema not found" and accidental registrations; the schema is registered by the release pipeline, not by the first producer that starts. The real check then lives in CI: before the merge you ask the Registry whether the new schema is compatible with the latest version of the subject. The template in the script that follows is the tool: the endpoint's reply contains the boolean field is_compatible and the level applied is the one configured for the subject (otherwise the global one).

# Template to adapt (not run: it needs a reachable Registry).
# Checks whether a schema is compatible with the latest version of the subject, using the level configured for that subject.
SCHEMA_JSON=$(jq -Rs '{schema: .}' src/main/avro/Order.avsc)

curl -s -X POST "http://schema-registry-staging:8081/compatibility/subjects/orders-value/versions/latest" \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d "$SCHEMA_JSON"
# expected reply according to the manual: an object with the boolean field "is_compatible"

The command was not run because there is no Registry in this environment; the shape of the request and the is_compatible field come from the REST API documentation. If the call returns false, the pipeline must fail: it is exactly the moment when a renamed user_id is stopped, before reaching the topic. Remember too that the Registry keeps its data in the _schemas topic: replication and retention of that topic are part of your backup strategy.

What you can do now

You know what is in the first five bytes of a message with a schema and why the consumer can read data written with a version different from its own. You can predict whether a change to an Avro schema passes in BACKWARD, FORWARD or FULL, and in which order to update producers and consumers. You know that in Protobuf the risk lies in the field numbers and that reusing them produces no errors. You know that in production the schema is registered by the pipeline and checked in CI. For the part of the flow that carries these messages towards databases continue with Kafka Connect; for the settings of the producer that sends them, go back to the advanced producers.