Overview

Key Benefits

Compatibility Contract Evolution Rules

Backward Compatibility (Default)

// v1 schema - producer writes
{
 "type": "record",
 "name": "Transaction",
 "fields": [
 {"name": "id", "type": "string"},
 {"name": "amount", "type": "double"}
 ]
}

// v2 schema - backward compatible (adds field with default)
{
 "type": "record",
 "name": "Transaction",
 "fields": [
 {"name": "id", "type": "string"},
 {"name": "amount", "type": "double"},
 {"name": "timestamp", "type": "long", "default": 0}
 ]
}

Forward Compatibility

Full Compatibility (Bidirectional)

Serialization Format Comparison

Apache Avro

{
 "type": "record",
 "name": "AMLTransaction",
 "namespace": "com.acaciafund.aml",
 "fields": [
 {"name": "transaction_id", "type": "string"},
 {"name": "timestamp", "type": "long"},
 {"name": "amount", "type": "double"},
 {
 "name": "metadata",
 "type": {
 "type": "record",
 "name": "TransactionMetadata",
 "fields": [
 {"name": "channel", "type": "string"},
 {"name": "location", "type": "string"},
 {"name": "device_id", "type": ["null", "string"], "default": null}
 ]
 }
 },
 {
 "name": "flags",
 "type": {
 "type": "enum",
 "name": "TransactionFlag",
 "symbols": ["NORMAL", "SUSPICIOUS", "REVIEWED", "BLOCKED"]
 }
 }
 ]
}

from avro.schema import Schema from avro.datafile import DataFileReader, DataFileWriter
from avro.io import DatumReader, DatumWriter
import json

# Load schema
schema = json.load(open('transaction.avsc'))

# Write data
writer = DatumWriter(schema)
dfw = DataFileWriter(open('transactions.avro', 'wb'), writer, schema)
dfw.append({
 'transaction_id': 'TXN-001',
 'timestamp': 1623849600,
 'amount': 15000.00,
 'metadata': {'channel': 'online', 'location': 'PL-WAR'},
 'flags': 'NORMAL'
})
dfw.close()

# Read data (schema evolution)
reader = DatumReader()
dfr = DataFileReader(open('transactions.avro', 'rb'), reader)
for record in dfr:
 print(record)
dfr.close()

Google Protobuf

syntax = "proto3";

package acaciafund.aml;

message TransactionMetadata {
 string channel = 1;
 string location = 2;
 optional string device_id = 3;
}

message AMLTransaction {
 string transaction_id = 1;
 int64 timestamp = 2;
 double amount = 3;
 TransactionMetadata metadata = 4;
 TransactionFlag flags = 5;
}

enum TransactionFlag {
 NORMAL = 0;
 SUSPICIOUS = 1;
 REVIEWED = 2;
 BLOCKED = 3;
}

# Generate Python code from .proto
protoc --python_out=. aml_transaction.proto

# Use generated code
from aml_transaction_pb2 import AMLTransaction, TransactionMetadata

# Create transaction
txn = AMLTransaction(
 transaction_id="TXN-001",
 timestamp=1623849600,
 amount=15000.00,
 metadata=TransactionMetadata(
 channel="online",
 location="PL-WAR"
 ),
 flags=TransactionFlag.NORMAL
)

# Serialize
serialized = txn.SerializeToString()

# Deserialize
txn2 = AMLTransaction()
txn2.ParseFromString(serialized)

JSON Schema

{
 "$schema": "http://json-schema.org/draft-07/schema#",
 "title": "AMLTransaction",
 "type": "object",
 "required": ["transaction_id", "timestamp", "amount"],
 "properties": {
 "transaction_id": {"type": "string"},
 "timestamp": {"type": "integer"},
 "amount": {"type": "number"},
 "metadata": {
 "type": "object",
 "properties": {
 "channel": {"type": "string"},
 "location": {"type": "string"},
 "device_id": {"type": ["string", "null"]}
 }
 },
 "flags": {
 "type": "string",
 "enum": ["NORMAL", "SUSPICIOUS", "REVIEWED", "BLOCKED"]
 }
 }
}

Compatibility Contract Transition Matrix

Deployment Strategy

Multi-Tenant Registry Deployment

Confluent Schema Registry Tenancy

# Tenant-specific registry endpoints
kafka-schema-registry --tenant=finance --port=8081
kafka-schema-registry --tenant=marketing --port=8082

# Schema registration per tenant
curl -X POST -H "Content-Type: application/vnd.schemaregistry.v1+json" \
 --data '{"schema": "{\"type\":\"record\",\"name\":\"Transaction\",\"fields\":[{\"name\":\"id\",\"type\":\"string\"}]}"}' \
 http://localhost:8081/tenants/finance/subjects/transaction-value/versions

Apicurio Multi-Tenant Architecture

# Tenant A namespace
apiVersion: v1
kind: Namespace
metadata:
 name: tenant-a
 labels:
 tenant: tenant-a

# Tenant B namespace
apiVersion: v1
kind: Namespace
metadata:
 name: tenant-b
 labels:
 tenant: tenant-b

CI/CD Integration for Schema Validation

GitHub Actions Workflow

name: Schema Validation

on:
 push:
 paths:
 - 'schemas/**/*.proto'
 - 'schemas/**/*.avsc'

jobs:
 validate-schema:
 runs-on: ubuntu-latest
 steps:
 - uses: actions/checkout@v3
 
 - name: Set up Protobuf
 uses: arduino/setup-protoc@v1
 
 - name: Validate Protobuf schemas
 run: |
 protoc --proto_path=schemas --cpp_out=dist schemas/*.proto
 
 - name: Validate Avro schemas
 run: |
 python3 scripts/validate_avro.py schemas/*.avsc
 
 - name: Check compatibility
 run: |
 # Compare against latest schema in registry
 curl -s http://localhost:8081/subjects/transaction-value/versions/latest/schema > latest.avsc
 python3 scripts/check_compatibility.py latest.avsc schemas/transaction.avsc

Schema Compatibility Check Script

#!/usr/bin/env python3
"""Check schema compatibility between old and new schemas."""

import json
import sys
from avro.schema import Parse

def check_backward_compatibility(old_schema_str: str, new_schema_str: str) -> bool:
 """Check if new schema is backward compatible with old schema."""
 old_schema = Parse(old_schema_str)
 new_schema = Parse(new_schema_str)
 
 old_fields = {f.name: f for f in old_schema.fields}
 new_fields = {f.name: f for f in new_schema.fields}
 
 # Check removed fields
 for name in old_fields:
 if name not in new_fields:
 print(f"ERROR: Field '{name}' was removed")
 return False
 
 # Check type changes
 for name in new_fields:
 if name in old_fields:
 if old_fields[name].type != new_fields[name].type:
 print(f"ERROR: Field '{name}' type changed from {old_fields[name].type} to {new_fields[name].type}")
 return False
 
 # Check new required fields
 for name, field in new_fields.items():
 if name not in old_fields:
 if field.default is None:
 print(f"ERROR: New required field '{name}' added without default")
 return False
 
 return True

if __name__ == "__main__":
 if len(sys.argv) != 3:
 print("Usage: check_compatibility.py old_schema.avsc new_schema.avsc")
 sys.exit(1)
 
 with open(sys.argv[1]) as f:
 old_schema = f.read()
 with open(sys.argv[2]) as f:
 new_schema = f.read()
 
 if check_backward_compatibility(old_schema, new_schema):
 print(" Schemas are backward compatible")
 sys.exit(0)
 else:
 print(" Schemas are NOT backward compatible")
 sys.exit(1)

Performance Considerations

Schema Registry Performance

Optimization Strategies

Security Considerations

Schema Registry Access Control

Schema Sanitization

Deployment Checklist

Conclusion