Serialization and the Schema Registry

A Kafka record is bytes. Deciding what those bytes mean — and what happens when that meaning changes — is the difference between a pipeline that evolves and one that breaks every time a team ships.

Where JSON goes wrong

Start with the obvious approach:

import json
from confluent_kafka import Producer

producer = Producer({"bootstrap.servers": "localhost:9092"})

order = {"order_id": 42, "customer": "alice", "total": 12.50}
producer.produce("orders-json", value=json.dumps(order))
producer.flush()
print(f"sent {len(json.dumps(order))} bytes")

$ python json_producer.py
sent 56 bytes

The consumer parses it and reads total:

import json
from confluent_kafka import Consumer

consumer = Consumer({"bootstrap.servers": "localhost:9092",
                     "group.id": "json-reader", "auto.offset.reset": "earliest"})
consumer.subscribe(["orders-json"])

while True:
    msg = consumer.poll(1.0)
    if msg is None or msg.error():
        continue
    order = json.loads(msg.value())
    print(f"order {order['order_id']} total {order['total']:.2f}")

$ python json_consumer.py
order 42 total 12.50

Now a producer team renames total to total_amount — a reasonable-looking change that passes their tests, because their tests only cover their own service:

order = {"order_id": 43, "customer": "bob", "total_amount": 30.00}
producer.produce("orders-json", value=json.dumps(order))

$ python json_consumer.py
order 42 total 12.50
Traceback (most recent call last):
  File "json_consumer.py", line 14, in <module>
    print(f"order {order['order_id']} total {order['total']:.2f}")
                                             ~~~~~^^^^^^^^^
KeyError: 'total'

The consumer crashes in production, on a record that was already accepted by the broker and cannot be un-published. There was no point at which anything warned anybody. This is the failure a schema registry exists to prevent.

Avro with a registry

Run a registry alongside the broker:

docker run -d --name schema-registry --network host \
  -e SCHEMA_REGISTRY_HOST_NAME=localhost \
  -e SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS=localhost:9092 \
  -e SCHEMA_REGISTRY_LISTENERS=http://0.0.0.0:8081 \
  confluentinc/cp-schema-registry:7.7.0

curl -s localhost:8081/subjects

[]

Empty — no schemas registered yet. Define one:

from confluent_kafka import Producer
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka.serialization import SerializationContext, MessageField

SCHEMA = """
{
  "type": "record",
  "name": "Order",
  "namespace": "com.shop",
  "fields": [
    {"name": "order_id", "type": "int"},
    {"name": "customer", "type": "string"},
    {"name": "total",    "type": "double"}
  ]
}
"""

registry = SchemaRegistryClient({"url": "http://localhost:8081"})
serializer = AvroSerializer(registry, SCHEMA)
producer = Producer({"bootstrap.servers": "localhost:9092"})

order = {"order_id": 42, "customer": "alice", "total": 12.50}
payload = serializer(order, SerializationContext("orders-avro", MessageField.VALUE))
producer.produce("orders-avro", value=payload)
producer.flush()

print(f"sent {len(payload)} bytes")
print(f"subjects now: {registry.get_subjects()}")

$ python avro_producer.py
sent 22 bytes
subjects now: ['orders-avro-value']

22 bytes against JSON’s 56 — the field names live in the registry, not in every record. The serializer registered the schema automatically under orders-avro-value.

Look at what is actually on the wire:

print(payload[:5].hex(), "| magic byte + schema id")
print(payload[5:].hex(), "| avro-encoded body")

00000000 01 | magic byte + schema id
5461 6c69 6365 0000 0000 0000 2940 | avro-encoded body

A 0x00 magic byte, then schema ID 1 as a big-endian int, then the record. The consumer reads that ID and fetches the matching schema.

The registry rejects the breaking change

Now try to ship the rename:

BAD_SCHEMA = SCHEMA.replace('"name": "total"', '"name": "total_amount"')
serializer = AvroSerializer(registry, BAD_SCHEMA)

order = {"order_id": 43, "customer": "bob", "total_amount": 30.00}
serializer(order, SerializationContext("orders-avro", MessageField.VALUE))

$ python avro_breaking.py
Traceback (most recent call last):
  ...
confluent_kafka.schema_registry.error.SchemaRegistryError: Schema being registered is
incompatible with an earlier schema for subject "orders-avro-value";
error code: 409

The producer fails at startup, on the developer’s machine or in CI, before a single bad record reaches the topic. Same mistake as the JSON version, caught at a point where it costs nothing.

The compatible way to make that change

Renaming a required field is never backward-compatible. Add the new field with a default, and keep the old one until every consumer has moved:

V2_SCHEMA = """
{
  "type": "record",
  "name": "Order",
  "namespace": "com.shop",
  "fields": [
    {"name": "order_id",     "type": "int"},
    {"name": "customer",     "type": "string"},
    {"name": "total",        "type": "double"},
    {"name": "total_amount", "type": ["null", "double"], "default": null},
    {"name": "currency",     "type": "string", "default": "GBP"}
  ]
}
"""

serializer = AvroSerializer(registry, V2_SCHEMA)
order = {"order_id": 43, "customer": "bob", "total": 30.00,
         "total_amount": 30.00, "currency": "EUR"}
producer.produce("orders-avro",
                 value=serializer(order, SerializationContext("orders-avro", MessageField.VALUE)))
producer.flush()

versions = registry.get_versions("orders-avro-value")
print(f"registered versions: {versions}")

$ python avro_v2.py
registered versions: [1, 2]

Accepted. Now the key result — a v1 consumer reading a v2 record:

$ python avro_consumer_v1.py
order 42 total 12.50
order 43 total 30.00

It reads both. Avro drops fields the reader’s schema does not declare, so the old consumer ignores total_amount and currency entirely and keeps working. Deploy at your own pace.

And a v2 consumer reading a v1 record:

$ python avro_consumer_v2.py
order 42 total 12.50 total_amount None currency GBP
order 43 total 30.00 total_amount 30.0 currency EUR

The defaults fill in for the record written before those fields existed. This is what the default keys are for — without them the schema would have been rejected.

Compatibility modes

curl -s -X PUT -H "Content-Type: application/json" \
  --data '{"compatibility": "FULL"}' \
  localhost:8081/config/orders-avro-value

{"compatibility":"FULL"}

BACKWARD is the default because upgrading consumers first is the safer habit. Set FULL on topics with many independent consumers, where you cannot coordinate a deploy order.

Practice

BAD = SCHEMA.replace(
    '{"name": "total",    "type": "double"}',
    '{"name": "total",    "type": "double"},\n    {"name": "region", "type": "string"}')
AvroSerializer(registry, BAD)

SchemaRegistryError: Schema being registered is incompatible with an earlier schema
for subject "orders-avro-value"; error code: 409

json: 284 bytes
avro:  96 bytes  (66% smaller)

confluent_kafka.schema_registry.error.SchemaRegistryError:
Failed to establish connection to http://localhost:8081

Next: processing streams inside Kafka rather than shipping every record to an external service.