S
AlgoSimplified

Schema Registry and Data Governance

Master Kafka Schema Registry: Avro, Protobuf, JSON Schema, schema evolution strategies, and managing schema versions in production systems.

Course Navigation

← Back to Kafka Course
Previous Lesson
Kafka Streams and Real-Time Processing
Current Lesson
8 of 10
Next Lesson
Security, Authentication, and Authorization
Progress 80%

Why Schema Registry?

In a distributed system, data schemas evolve. Without governance, you get:

🚨 The Schema Hell

  • Producer sends new field → Consumer crashes
  • Producer removes field → Consumer breaks
  • Producer changes type → Data corruption
  • Multiple teams → Incompatible schemas
  • No versioning → Can't rollback safely

Schema Registry solves this by centralizing schema management and enforcing compatibility rules.

How Schema Registry Works

Think of it as a "Git for data schemas":

1. Producer registers schema → Schema Registry stores it
2. Producer sends message with schema ID → Kafka stores message + ID
3. Consumer reads message + ID → Fetches schema from Registry
4. Consumer deserializes using correct schema version

Schema Formats Comparison

Three main formats, each with trade-offs:

Avro (Recommended)

✅ Pros

  • Compact binary format - 50% smaller than JSON
  • Rich schema evolution - backward/forward compatibility
  • Schema Registry native - built for Kafka
  • Type safety - compile-time validation

❌ Cons

  • Learning curve - not as familiar as JSON
  • Schema Registry dependency - single point of failure
  • Tooling complexity - more setup required

Protobuf

✅ Pros

  • Google's standard - battle-tested
  • Excellent performance - very fast
  • Language agnostic - works everywhere
  • Rich type system - complex data structures

❌ Cons

  • Schema evolution complexity - more rules to follow
  • Binary format - harder to debug
  • Less Kafka integration - newer support

JSON Schema

✅ Pros

  • Human readable - easy to understand
  • No binary format - easy debugging
  • Web standard - familiar to developers
  • No Schema Registry needed - self-contained

❌ Cons

  • Larger messages - verbose format
  • No built-in evolution - manual versioning
  • Performance overhead - parsing JSON
  • Type safety issues - runtime validation only

Setting Up Schema Registry

Let's get Schema Registry running:

1. Download and Start

# Download Confluent Platform
wget https://packages.confluent.io/archive/7.0/confluent-community-7.0.1.tar.gz
tar -xzf confluent-community-7.0.1.tar.gz
cd confluent-7.0.1

# Start Schema Registry
bin/schema-registry-start config/schema-registry.properties

2. Configuration

# config/schema-registry.properties
listeners=http://0.0.0.0:8081
kafkastore.bootstrap.servers=PLAINTEXT://localhost:9092
kafkastore.topic=_schemas
kafkastore.group.id=schema-registry
schema.registry.url=http://localhost:8081

Schema Evolution Strategies

The key to successful schema evolution is understanding compatibility:

Backward Compatibility

Definition: New schema can read data written with old schema

Use case: Consumer upgrades before producer

# Old schema
{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "id", "type": "int"},
    {"name": "name", "type": "string"}
  ]
}

# New schema (backward compatible)
{
  "type": "record", 
  "name": "User",
  "fields": [
    {"name": "id", "type": "int"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": ["null", "string"], "default": null}  # Optional field
  ]
}}

Forward Compatibility

Definition: Old schema can read data written with new schema

Use case: Producer upgrades before consumer

# Old schema
{
  "type": "record",
  "name": "User", 
  "fields": [
    {"name": "id", "type": "int"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": ["null", "string"], "default": null}
  ]
}

# New schema (forward compatible)
{
  "type": "record",
  "name": "User",
  "fields": [
    {"name": "id", "type": "int"},
    {"name": "name", "type": "string"}
    # Removed email field - old consumers ignore it
  ]
}}

Full Compatibility

Definition: Both backward and forward compatible

Use case: Safe for any upgrade order

Schema Evolution Rules

Follow these rules to maintain compatibility:

✅ Safe Changes (Backward Compatible)

  • Add optional fields with default values
  • Remove optional fields (if not used)
  • Change field order (Avro only)
  • Add union types with null

⚠️ Risky Changes (Requires Careful Planning)

  • Change field types (int → long is safe, int → string is not)
  • Remove required fields (breaks backward compatibility)
  • Add required fields (breaks forward compatibility)
  • Change field names (breaks both)

❌ Breaking Changes (Never Do This)

  • Change record name
  • Change namespace
  • Change field types incompatibly
  • Remove fields without default

Producer with Schema Registry

Let's build a producer that uses Schema Registry:

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

# 1. Define schema
schema_str = """
{
  "type": "record",
  "name": "User",
  "namespace": "com.example",
  "fields": [
    {"name": "id", "type": "int"},
    {"name": "name", "type": "string"},
    {"name": "email", "type": ["null", "string"], "default": null}
  ]
}
"""

# 2. Configure Schema Registry client
schema_registry_conf = {'url': 'http://localhost:8081'}
schema_registry_client = SchemaRegistryClient(schema_registry_conf)

# 3. Create Avro serializer
avro_serializer = AvroSerializer(
    schema_registry_client,
    schema_str,
    to_dict=lambda obj, ctx: obj.__dict__
)

# 4. Configure producer
producer_conf = {
    'bootstrap.servers': 'localhost:9092',
    'key.serializer': avro_serializer,
    'value.serializer': avro_serializer
}

producer = Producer(producer_conf)

# 5. Send message
user_data = {"id": 1, "name": "John Doe", "email": "[email protected]"}
producer.produce(
    topic='users',
    value=user_data,
    on_delivery=lambda err, msg: print(f'Message delivered: {msg.value()}')
)

producer.flush()

Consumer with Schema Registry

Now let's build a consumer that deserializes using Schema Registry:

from confluent_kafka import Consumer
from confluent_kafka.serialization import SerializationContext, MessageField
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroDeserializer

# 1. Configure Schema Registry client
schema_registry_conf = {'url': 'http://localhost:8081'}
schema_registry_client = SchemaRegistryClient(schema_registry_conf)

# 2. Create Avro deserializer
avro_deserializer = AvroDeserializer(
    schema_registry_client,
    from_dict=lambda obj, ctx: obj
)

# 3. Configure consumer
consumer_conf = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'user-consumer-group',
    'auto.offset.reset': 'earliest',
    'key.deserializer': avro_deserializer,
    'value.deserializer': avro_deserializer
}

consumer = Consumer(consumer_conf)
consumer.subscribe(['users'])

# 4. Consume messages
while True:
    msg = consumer.poll(1.0)
    if msg is None:
        continue
    if msg.error():
        print(f"Consumer error: {msg.error()}")
        continue
    
    user = msg.value()
    print(f"Received user: {user}")

consumer.close()

Schema Registry API

Manage schemas via REST API:

Register Schema

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "{\"type\":\"record\",\"name\":\"User\",\"fields\":{\"id\":\"int\",\"name\":\"string\"}}"}' \
  http://localhost:8081/subjects/users-value/versions

Get Schema

curl http://localhost:8081/subjects/users-value/versions/1

Check Compatibility

curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  --data '{"schema": "{\"type\":\"record\",\"name\":\"User\",\"fields\":{\"id\":\"int\",\"name\":\"string\",\"email\":[\"null\",\"string\"],\"default\":null}}"}' \
  http://localhost:8081/compatibility/subjects/users-value/versions/latest

Schema Registry Best Practices

Naming Conventions

  • Subject naming: {topic}-{key|value}
  • Schema names: Use namespaces (e.g., com.company.User)
  • Field names: Use camelCase for consistency

Compatibility Settings

  • Backward: Consumer upgrades first
  • Forward: Producer upgrades first
  • Full: Any upgrade order (recommended)
  • None: No compatibility checks (dangerous)

Monitoring

  • Schema registry metrics - JMX monitoring
  • Schema evolution alerts - Notify on breaking changes
  • Schema usage tracking - Which schemas are active

Production Deployment

High Availability

# Run multiple Schema Registry instances
# Use load balancer for failover
# Replicate _schemas topic across brokers

Security

# Enable authentication
schema.registry.inter.instance.protocol=https
schema.registry.ssl.keystore.location=/path/to/keystore
schema.registry.ssl.keystore.password=password

# Enable authorization
schema.registry.authorizer.class=io.confluent.kafka.schemaregistry.security.authorizer.rbac.RbacAuthorizer

Key Takeaways

  1. Schema Registry prevents data corruption - enforce compatibility rules
  2. Avro is the best choice for Kafka - compact, evolvable, integrated
  3. Plan schema evolution carefully - breaking changes are expensive
  4. Use full compatibility mode - safest for production
  5. Monitor schema usage - track what's actually being used

Next Steps

Ready to secure your Kafka cluster? Check out our next lesson on Security, Authentication, and Authorization where we'll learn how to protect your data and control access.