Schema Registry & Data Contracts

Schema-on-write vs schema-on-read

Schema-on-write pushes failures left: the producer cannot publish a breaking change if compatibility checks fail. Schema-on-read is flexible but shifts burden to every consumer—each must handle missing fields, type changes, and renames independently.

The silent breakage problem

Without a registry, this sequence is common:

1. Team A ships v2 of OrderEvent — renames customerId → customer_id
2. Kafka accepts JSON bytes — no validation at broker
3. Team B’s Flink job still reads customerId → null for all new records
4. Revenue dashboard under-reports for three days until someone notices

Data contract

A data contract is the agreed structure of a topic’s key and value: field names, types, semantics, defaults, and evolution rules. It is owned by the producing team, reviewed by consumers, and enforced at registration time—not documented in a wiki and hoped for.

• Producer obligation: register schema before writing; only publish compatible versions
• Consumer obligation: deserialize with reader schema that resolves against writer schema
• Platform obligation: run Schema Registry HA, set global compatibility, audit subjects

Schema Registry role

Central repository for Avro, Protobuf, and JSON Schema definitions. Each registration gets a monotonically increasing schema ID and version per subject. Serializers fetch schema by ID on read; on write they register (or lookup) and prepend ID to payload. Incompatible schemas are rejected before a single byte hits the topic.

flowchart LR
  P[Producer\nAvro Serializer]
  SR[Schema Registry\nREST :8081]
  S[(_schemas topic\ncompacted)]
  K[(Kafka topic\norders)]
  C[Consumer\nAvro Deserializer]

  P -->|register / lookup ID| SR
  SR -->|persist| S
  P -->|0x00 + ID + bytes| K
  K --> C
  C -->|GET /schemas/ids/id| SR
  SR --> S

Storage: _schemas topic

All schema metadata lives in the internal compacted topic _schemas (configurable via kafkastore.topic). Keys are subject + version; values hold schema string, schema type, references. Compaction retains latest record per key— the registry rebuilds in-memory caches on startup by consuming this topic.

REST API

Primary operations (Confluent REST v1):

# Register new schema version
curl -X POST http://schema-registry:8081/subjects/orders-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "{\"type\":\"record\",\"name\":\"Order\",\"fields\":[{\"name\":\"orderId\",\"type\":\"long\"}]}"}'

# Check compatibility against latest (without registering)
curl -X POST "http://schema-registry:8081/compatibility/subjects/orders-value/versions/latest" \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "..."}'

# Fetch schema used by on-wire ID 42
curl http://schema-registry:8081/schemas/ids/42

Schema ID and wire format

Confluent serializers prepend a 5-byte header to each message (key and value independently):

Byte layout (Confluent Platform wire format)
┌────────┬──────────────────────────────┬─────────────────────────┐
│ 0x00   │ Schema ID (4 bytes, big-endian) │ Serialized payload      │
│ magic  │ e.g. 0x00 0x00 0x00 0x2A (=42)  │ Avro / Protobuf / JSON  │
└────────┴──────────────────────────────┴─────────────────────────┘

• Magic byte 0x00: Confluent framing (distinct from raw Avro or Kafka’s own formats)
• Schema ID: global integer assigned at registration—same ID forever for that exact schema text
• Payload: binary Avro (default), Protobuf bytes, or JSON Schema serialized form

Subject naming strategies

A subject is the registry namespace for a schema lineage. Strategy determines how subject names are derived from topic + record:

value.subject.name.strategy=io.confluent.kafka.serializers.subject.TopicNameStrategy
# alternatives:
# io.confluent.kafka.serializers.subject.RecordNameStrategy
# io.confluent.kafka.serializers.subject.TopicRecordNameStrategy

High availability

Run multiple Schema Registry instances behind a load balancer. All instances read/write the same _schemas topic; Kafka acts as coordination log. Leader election (older versions) or all-active mode depends on version— platform teams monitor registry lag and kafkastore connectivity like any critical dependency.

Why Avro for Kafka

• Schema required for read and write — no ambiguous bytes
• Compact binary — smaller than JSON; no field names on wire (uses schema)
• Evolution built-in — resolution rules between writer and reader schemas
• Ecosystem — Kafka, Connect, Flink, Spark, Hive native support

Schema definition

Avro schemas are JSON documents: types, fields, defaults, namespaces, doc strings.

{
  "type": "record",
  "name": "Order",
  "namespace": "com.acme.events",
  "doc": "Order placed event",
  "fields": [
    { "name": "orderId", "type": "long" },
    { "name": "customerId", "type": "string" },
    { "name": "amount", "type": "double", "default": 0.0 },
    { "name": "status", "type": {
        "type": "enum",
        "name": "OrderStatus",
        "symbols": ["PENDING", "SHIPPED", "CANCELLED"]
      }, "default": "PENDING"
    },
    { "name": "tags", "type": { "type": "array", "items": "string" }, "default": [] }
  ]
}

Primitive types

Complex types

Writer schema + reader schema resolution

On deserialize, Avro uses the writer schema (embedded via registry ID in the message) and the reader schema (your application’s current model). Avro resolves differences field-by-field:

• Field in writer, not in reader: skipped (if reader doesn’t need it)
• Field in reader, not in writer: filled from default in reader schema
• Same name, compatible type promotion: e.g. int → long
• Field renamed: use aliases on reader schema to map old name

flowchart LR
  W[Writer schema v1\norderId, amount]
  B[Binary payload\non Kafka]
  R[Reader schema v2\norderId, amount, currency]
  O[Resolved Order object\ncurrency = default USD]

  W --> B
  B --> R
  R --> O

GenericRecord vs SpecificRecord

Order order = Order.newBuilder()
    .setOrderId(9001L)
    .setCustomerId("cust-42")
    .setAmount(149.99)
    .setStatus(OrderStatus.PENDING)
    .build();

ProducerRecord<String, Order> record =
    new ProducerRecord<>("orders", order.getOrderId().toString(), order);
producer.send(record);

Transitive variants

Non-transitive modes compare new schema only against the latest version. TRANSITIVE modes compare against all historical versions in the subject—stricter, catches multi-hop breaks.

• BACKWARD_TRANSITIVE — new schema must read data from every older version
• FORWARD_TRANSITIVE — every older schema must read new data
• FULL_TRANSITIVE — both directions against full history

curl -X PUT http://schema-registry:8081/config \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"compatibility": "FULL_TRANSITIVE"}'

# Per-subject override
curl -X PUT http://schema-registry:8081/config/orders-value \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"compatibility": "BACKWARD"}'

Compatibility examples

BACKWARD-safe change — add optional field with default:

{ "name": "currency", "type": "string", "default": "USD" }

BACKWARD-breaking — add required field (no default): old writers don’t populate it; old readers may not expect it depending on direction.

FORWARD-breaking — remove field without default: new writers stop sending it; old readers still require it.

Safe changes

• Add field with default — BACKWARD and FULL friendly; old messages deserialize with default
• Remove field that had a default — FORWARD friendly; old consumers still see default when reading new data
• Add optional union — ["null","string"] with default null
• Widen numeric type — int → long (promotion rules)

Unsafe changes

• Rename field — Avro sees remove + add; use alias instead
• Change type — string → int breaks resolution; widen only
• Remove required field — breaks FORWARD unless consumers upgraded
• Reorder fields — safe on wire (order by schema, not JSON) but confuses human diffs

Alias trick for renames

Add new canonical name; attach aliases with old name so reader schema resolves both:

{
  "name": "customer_id",
  "type": "string",
  "aliases": ["customerId"]
}

Deploy consumer with aliased reader schema first, then producer writing customer_id. Registry still sees a structural change—plan compatibility mode and version bump accordingly.

Reserved fields pattern

Before removing a field consumers might still read:

1. Stop writing the field in producer (or write null/default)
2. Mark field doc: "DEPRECATED — remove in v4"
3. Confirm all consumers on reader schema without the field
4. Remove in new major schema version under FULL_TRANSITIVE only when safe

{
  "name": "legacyStatusCode",
  "type": ["null", "int"],
  "default": null,
  "doc": "RESERVED: do not use. Will be removed when all consumers >= v2.3."
}

Versioning workflow

sequenceDiagram
  participant Dev as Developer
  participant CI as CI pipeline
  participant SR as Schema Registry
  participant Cons as Consumers
  participant Prod as Producer

  Dev->>CI: PR updates .avsc
  CI->>SR: compatibility check
  SR-->>CI: compatible ✓
  CI->>Cons: deploy consumers (new reader)
  Cons-->>Prod: ready
  Prod->>SR: register + publish vN

Protobuf

Messages identified by field numbers—renaming a field does not break wire compatibility if number stays same. Never reuse field numbers. reserved blocks deprecated numbers. Schema Registry stores .proto text; serializer uses Confluent wire format (same magic byte + schema ID).

syntax = "proto3";
package com.acme.events;

message Order {
  int64 order_id = 1;
  string customer_id = 2;
  double amount = 3;
  OrderStatus status = 4;
  reserved 5;           // formerly legacy_code
  reserved "old_field";
}

enum OrderStatus {
  PENDING = 0;
  SHIPPED = 1;
  CANCELLED = 2;
}

Forward compatibility: old code ignores unknown fields (proto3). Adding fields is safe; removing without reservation is not. Excellent when same models serve gRPC and Kafka.

JSON Schema

Most flexible, least safe—schemas can be permissive (additionalProperties: true). Registry supports draft versions; compatibility checking compares JSON Schema structures. Useful when partners submit JSON and you want registry governance without Avro codegen.

{
  "$schema": "http://json-schema.org/draft-07/schema#",
  "title": "Order",
  "type": "object",
  "required": ["orderId", "customerId"],
  "properties": {
    "orderId": { "type": "integer" },
    "customerId": { "type": "string" },
    "amount": { "type": "number", "default": 0 }
  },
  "additionalProperties": false
}

Choosing a format

• Internal Kafka, JVM + analytics: Avro + SpecificRecord
• Cross-system with gRPC: Protobuf everywhere
• External REST / webhook ingress: JSON Schema or schemaless JSON with validation service
• Connect sink to BigQuery: Avro/Protobuf with registry often required for strongly typed tables

Dependencies

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
</dependency>
<dependency>
  <groupId>io.confluent</groupId>
  <artifactId>kafka-avro-serializer</artifactId>
  <version>7.6.0</version>
</dependency>

KafkaAvroSerializer / KafkaAvroDeserializer

spring:
  kafka:
    bootstrap-servers: kafka:9092
    properties:
      schema.registry.url: http://schema-registry:8081
      specific.avro.reader: true
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: io.confluent.kafka.serializers.KafkaAvroSerializer
      properties:
        auto.register.schemas: false
        use.latest.version: true
    consumer:
      group-id: order-service
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: io.confluent.kafka.serializers.KafkaAvroDeserializer

@KafkaListener with Avro

@KafkaListener(topics = "orders", groupId = "order-service")
public void onOrder(ConsumerRecord<String, Order> record) {
  Order order = record.value();
  log.info("orderId={} amount={}", order.getOrderId(), order.getAmount());
}

SchemaRegistryClient bean

For admin tasks—compatibility checks, explicit registration, subject listing—inject Confluent’s client. @EnableSchemaRegistryClient (Spring Cloud Stream schema registry starters) or manual bean:

@Configuration
public class SchemaRegistryConfig {

  @Bean
  public SchemaRegistryClient schemaRegistryClient(
      @Value("${spring.kafka.properties.schema.registry.url}") String url) {
    Map<String, Object> props = Map.of(
        AbstractKafkaSchemaSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, url
    );
    return new CachedSchemaRegistryClient(
        List.of(url),
        1000,
        List.of(new AvroSchemaProvider()),
        props
    );
  }
}

// Usage: preflight in admin endpoint or startup health
// schemaRegistryClient.testCompatibility("orders-value", new AvroSchema(schemaString));

@Service
public class OrderPublisher {
  private final KafkaTemplate<String, Order> kafkaTemplate;

  public OrderPublisher(KafkaTemplate<String, Order> kafkaTemplate) {
    this.kafkaTemplate = kafkaTemplate;
  }

  public void publish(Order order) {
    kafkaTemplate.send("orders", order.getOrderId().toString(), order);
  }
}