It's 2:47 AM on a Tuesday. Your on-call phone buzzes. A critical downstream analytics dashboard has gone dark, and the incident channel fills up fast. A developer on the source team deployed a "minor update" that renamed customer_id to customerId — camelCase instead of snake_case, totally reasonable from their perspective. But that one change cascaded through your Kafka consumers, your Spark transforms, your dbt models, and finally into the BI layer where someone's Monday morning executive report now shows nothing but nulls.
This is schema evolution failure in its most common, most painful form. It's not exotic. It happens constantly in organizations where the implicit contract between producers and consumers of data is never made explicit. And the fix isn't just technical — it requires thinking about schemas as a first-class concern in your pipeline architecture, the same way you think about data quality, lineage, or access control.
By the end of this lesson, you'll understand not just how to react when schemas break, but how to design systems where schema changes are anticipated, versioned, communicated, and deployed without ever taking a pipeline offline. We're going to go deep on the mechanics, the tradeoffs, and the places where conventional wisdom breaks down.
What you'll learn:
This lesson assumes you are comfortable with the following:
If you haven't worked with a Schema Registry before, read through the Confluent documentation overview first. You don't need to be an expert, but you should know what it is.
Before we can handle schema changes safely, we need a precise vocabulary for them. The terms "breaking" and "non-breaking" get thrown around loosely, but they're always relative to a direction: backward compatibility, forward compatibility, and full compatibility mean fundamentally different things, and conflating them is where most teams go wrong.
A schema change is backward compatible if new code can read data written with the old schema. This is the perspective of the consumer upgrading while old data (or old producers still running) needs to remain readable.
Classic examples of backward-compatible changes:
INT to BIGINTClassic examples of backward-incompatible changes:
STRING to INT)A schema change is forward compatible if old code can read data written with the new schema. This is the perspective of consumers that haven't upgraded yet — they'll receive messages serialized with the new schema and need to handle them gracefully.
Forward compatibility requires:
Full compatibility means both backward AND forward compatible simultaneously. It's the strictest mode and the hardest to maintain. Essentially it means you can only add optional fields with defaults. You can never remove or rename anything — you can only deprecate.
Enum additions deserve special mention because they trip up even experienced engineers. In Avro, adding a new symbol to an enum is NOT backward compatible by default. If an old consumer receives a message containing an unknown enum value, it will throw a deserialization error unless the schema defines a default value for the enum. In Protobuf, unknown enum values are preserved as their integer representation, making it more naturally forward-compatible. This difference in behavior between serialization formats is where a lot of cross-system bugs live.
When a schema change request comes in, run it through this mental model:
Change Type Backward Forward Full
───────────────────────────────────────────────────────────
Add optional field (w/ default) ✓ ✓ ✓
Add required field (no default) ✗ ✓ ✗
Remove field ✓ ✗ ✗
Rename field ✗ ✗ ✗
Change type (compatible widening) ✓* ✓* ✓*
Change type (incompatible) ✗ ✗ ✗
Add enum value (Avro, no default) ✗ ✓ ✗
Add enum value (Protobuf) ✓ ✓ ✓
Make nullable field non-nullable ✗ ✓ ✗
*Type widening compatibility depends heavily on the serialization format. Avro has strict rules; JSON is effectively untyped and more permissive but correspondingly less safe.
The Schema Registry (we'll use Confluent Schema Registry as the canonical implementation, though AWS Glue Schema Registry and other alternatives follow similar principles) is the enforcement layer that prevents incompatible schemas from ever reaching production consumers.
The core mechanism: every Avro or Protobuf message produced to Kafka contains a magic byte followed by a 4-byte schema ID, not the schema itself. The consumer uses that schema ID to fetch the writer's schema from the registry, then uses the Avro/Protobuf schema resolution rules to convert the message to the reader's schema. This schema evolution at the reader level — not the wire level — is what makes compatibility tractable.
Schema Registry subjects (the unit of schema versioning, typically one per Kafka topic) can be configured independently:
import requests
import json
REGISTRY_URL = "http://schema-registry:8081"
def set_subject_compatibility(subject: str, compatibility: str) -> dict:
"""
compatibility: one of BACKWARD, BACKWARD_TRANSITIVE, FORWARD,
FORWARD_TRANSITIVE, FULL, FULL_TRANSITIVE, NONE
"""
url = f"{REGISTRY_URL}/config/{subject}"
payload = {"compatibility": compatibility}
response = requests.put(url, json=payload)
response.raise_for_status()
return response.json()
# Set FULL_TRANSITIVE for a critical orders topic
# TRANSITIVE variants check against ALL previous versions, not just the latest
set_subject_compatibility(
subject="orders-value",
compatibility="FULL_TRANSITIVE"
)
The difference between BACKWARD and BACKWARD_TRANSITIVE is critical and underappreciated. Non-transitive checks only validate the new schema against the immediately previous version. Transitive checks against every version in history. If you're on schema version 12 and a consumer is still running version 3 (it happens), non-transitive mode might let a breaking change slip through.
Warning: Many teams default to
BACKWARDand assume they're protected. They're not — they're protected against consumers running version N-1, but consumers lagging further back can still break. UseBACKWARD_TRANSITIVEorFULL_TRANSITIVEfor topics where consumer lag is a real operational concern.
Let's walk through the lifecycle of a real schema. We have an order_events topic for an e-commerce platform:
from confluent_kafka.schema_registry import SchemaRegistryClient, Schema
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka import Producer
REGISTRY_CONFIG = {"url": "http://schema-registry:8081"}
registry_client = SchemaRegistryClient(REGISTRY_CONFIG)
# Version 1: Original schema
ORDER_SCHEMA_V1 = json.dumps({
"type": "record",
"name": "OrderEvent",
"namespace": "com.wickedsmartdata.orders",
"fields": [
{"name": "order_id", "type": "string"},
{"name": "customer_id", "type": "string"},
{"name": "total_amount_cents", "type": "long"},
{"name": "currency", "type": "string"},
{"name": "status", "type": {
"type": "enum",
"name": "OrderStatus",
"symbols": ["PENDING", "CONFIRMED", "SHIPPED", "DELIVERED", "CANCELLED"]
}},
{"name": "created_at", "type": "long", "doc": "Unix timestamp in milliseconds"}
]
})
schema_v1 = Schema(ORDER_SCHEMA_V1, schema_type="AVRO")
schema_id_v1 = registry_client.register_schema(
subject_name="order_events-value",
schema=schema_v1
)
print(f"Registered schema V1 with ID: {schema_id_v1}")
Three months later, the product team wants to add a shipping_address field and a discount_amount_cents field. This is a clean additive change — but let's do it right:
# Version 2: Adding optional fields with defaults
ORDER_SCHEMA_V2 = json.dumps({
"type": "record",
"name": "OrderEvent",
"namespace": "com.wickedsmartdata.orders",
"fields": [
{"name": "order_id", "type": "string"},
{"name": "customer_id", "type": "string"},
{"name": "total_amount_cents", "type": "long"},
{"name": "currency", "type": "string"},
{"name": "status", "type": {
"type": "enum",
"name": "OrderStatus",
"symbols": ["PENDING", "CONFIRMED", "SHIPPED", "DELIVERED", "CANCELLED"],
"default": "PENDING" # Added default to protect against future enum additions
}},
{"name": "created_at", "type": "long", "doc": "Unix timestamp in milliseconds"},
# New fields — union with null gives us a default of null
{
"name": "discount_amount_cents",
"type": ["null", "long"],
"default": None,
"doc": "Discount applied in cents. Null if no discount."
},
{
"name": "shipping_address",
"type": ["null", {
"type": "record",
"name": "Address",
"fields": [
{"name": "street_line_1", "type": "string"},
{"name": "street_line_2", "type": ["null", "string"], "default": None},
{"name": "city", "type": "string"},
{"name": "state_province", "type": "string"},
{"name": "postal_code", "type": "string"},
{"name": "country_code", "type": "string"}
]
}],
"default": None
}
]
})
# Validate compatibility before registering
def check_compatibility(subject: str, schema_str: str) -> bool:
url = f"{REGISTRY_URL}/compatibility/subjects/{subject}/versions/latest"
payload = {"schema": schema_str}
response = requests.post(url, json=payload)
result = response.json()
return result.get("is_compatible", False)
is_compat = check_compatibility("order_events-value", ORDER_SCHEMA_V2)
print(f"Schema V2 compatible: {is_compat}") # Should be True
if is_compat:
schema_v2 = Schema(ORDER_SCHEMA_V2, schema_type="AVRO")
schema_id_v2 = registry_client.register_schema(
subject_name="order_events-value",
schema=schema_v2
)
Notice we check compatibility programmatically before registering. This should be baked into your CI/CD pipeline — failing a PR at schema registration time is infinitely cheaper than failing at 2:47 AM.
Streaming systems with Schema Registry handle compatibility at the serialization layer. But many pipelines run against SQL databases — PostgreSQL, Snowflake, BigQuery, Redshift — where schema changes mean DDL. Here the challenge is different: you can't use schema resolution rules. A query that references customer_id will fail if the column no longer exists.
The Expand-Contract pattern (also called parallel change or two-phase migration) is the canonical solution for zero-downtime column renames, type changes, and table restructuring.
Phase 1: Expand Add the new column alongside the old one. Update all writers to write to both columns. At this point, both columns have valid data, and all readers can safely read from either.
Phase 2: Migrate Readers Update each consumer/reader to use the new column. Since both columns exist and both have current data, readers can be migrated independently at their own pace.
Phase 3: Contract Once all readers are confirmed to use the new column, drop the old column.
Let's work through a realistic example: renaming user_email to email_address in a user_events table in Snowflake, while maintaining a live pipeline.
-- Phase 1: EXPAND
-- Add the new column
ALTER TABLE user_events ADD COLUMN email_address VARCHAR(320);
-- Update the ingestion pipeline to write to both columns
-- (shown as a Snowflake MERGE example for a CDC pipeline)
MERGE INTO user_events AS target
USING staging_user_events AS source
ON target.event_id = source.event_id
WHEN MATCHED THEN UPDATE SET
target.user_email = source.email, -- Keep writing old column
target.email_address = source.email, -- Also write new column
target.event_timestamp = source.event_timestamp
WHEN NOT MATCHED THEN INSERT (
event_id, user_email, email_address, event_timestamp
) VALUES (
source.event_id, source.email, source.email, source.event_timestamp
);
-- Backfill historical data where email_address is still null
UPDATE user_events
SET email_address = user_email
WHERE email_address IS NULL;
-- Phase 2: MIGRATE READERS
-- Each downstream model updates to use email_address instead of user_email
-- Example dbt model update:
-- Before:
SELECT
event_id,
user_email,
event_timestamp,
event_type
FROM {{ ref('user_events') }}
-- After:
SELECT
event_id,
email_address,
event_timestamp,
event_type
FROM {{ ref('user_events') }}
-- Phase 3: CONTRACT
-- Only after ALL downstream models confirmed migrated:
-- First, remove the dual-write from the ingestion pipeline
-- Then drop the column:
ALTER TABLE user_events DROP COLUMN user_email;
Tip: Before dropping the old column in Phase 3, run a query that scans your dbt
manifest.jsonor query logs to confirm no query has referenceduser_emailin the past 30 days. In Snowflake,QUERY_HISTORYmakes this straightforward. Dropping a column that something still reads gives you a 2:47 AM call with no warning.
The biggest operational challenge with expand-contract is tracking which readers are still on the old schema. For a small team, a migration tracking table works well:
CREATE TABLE schema_migration_tracking (
migration_id VARCHAR(100) PRIMARY KEY,
description TEXT,
table_name VARCHAR(200),
old_column VARCHAR(200),
new_column VARCHAR(200),
phase VARCHAR(20) CHECK (phase IN ('EXPAND', 'MIGRATING', 'CONTRACT', 'COMPLETE')),
expand_completed_at TIMESTAMP,
contract_eligible_at TIMESTAMP, -- Set when all readers confirmed migrated
contracted_at TIMESTAMP,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
INSERT INTO schema_migration_tracking VALUES (
'USR-EVENTS-EMAIL-RENAME-001',
'Rename user_email to email_address in user_events table',
'user_events',
'user_email',
'email_address',
'EXPAND',
CURRENT_TIMESTAMP,
NULL,
NULL,
CURRENT_TIMESTAMP
);
Sometimes you genuinely cannot avoid a breaking change. The data type is wrong in a way that causes actual analytical errors. The column name violates a new naming standard that ten downstream systems need to align on. The table structure needs a fundamental redesign.
In these cases, the strategy shifts from preventing downtime to minimizing it through controlled, observable migration.
The cleanest approach for breaking changes in streaming systems is to run two versions of the pipeline in parallel, topic-level:
orders-v1 ──► consumer-v1 ──► analytics-v1
│
orders-v2 ──► consumer-v2 ──► analytics-v2 (new)
│
[cutover point]
The producer writes to both orders-v1 and orders-v2 simultaneously (the dual-write period). Consumers migrate to the new topic at their own pace. Once all consumers have migrated, production stops writing to orders-v1.
from confluent_kafka import Producer
import json
class DualWriteOrderProducer:
"""
Dual-write producer for schema migration period.
Writes to both v1 (legacy) and v2 (new schema) topics.
V1 schema: flat structure with customer_id as string
V2 schema: nested customer object, customer_id as UUID type,
added fraud_score field
"""
def __init__(self, bootstrap_servers: str, schema_registry_url: str):
self.producer = Producer({"bootstrap.servers": bootstrap_servers})
self.v1_topic = "order_events-v1"
self.v2_topic = "order_events-v2"
self._migration_enabled = True # Feature flag — turn off when migration complete
def produce_order_event(self, order: dict) -> None:
# Always produce V2 (new canonical format)
v2_message = self._transform_to_v2(order)
self.producer.produce(
topic=self.v2_topic,
key=order["order_id"].encode(),
value=json.dumps(v2_message).encode()
)
# Also produce V1 during migration window
if self._migration_enabled:
v1_message = self._transform_to_v1(order)
self.producer.produce(
topic=self.v1_topic,
key=order["order_id"].encode(),
value=json.dumps(v1_message).encode()
)
self.producer.flush()
def _transform_to_v1(self, order: dict) -> dict:
"""Legacy flat format"""
return {
"order_id": order["order_id"],
"customer_id": str(order["customer"]["id"]), # Flatten nested object
"total_amount_cents": order["total_amount_cents"],
"currency": order["currency"],
"status": order["status"],
"created_at": order["created_at"]
}
def _transform_to_v2(self, order: dict) -> dict:
"""New structured format"""
return {
"order_id": order["order_id"],
"customer": {
"id": order["customer"]["id"],
"tier": order["customer"].get("tier", "STANDARD"),
"country_code": order["customer"].get("country_code")
},
"total_amount_cents": order["total_amount_cents"],
"currency": order["currency"],
"status": order["status"],
"fraud_score": order.get("fraud_score"), # New field in V2
"created_at": order["created_at"]
}
def disable_v1_writes(self) -> None:
"""Call this after all consumers confirmed migrated to V2"""
self._migration_enabled = False
print("V1 dual-write disabled. Migration complete.")
Type changes are among the nastiest schema migrations because they require data transformation, not just structural change. Consider changing order_total from DECIMAL(10,2) in dollars to BIGINT in cents — a common migration as pipelines mature and floating-point precision becomes a problem.
-- This is a BREAKING type change. We use expand-contract with a transform.
-- Phase 1: EXPAND with transformation
ALTER TABLE orders ADD COLUMN total_amount_cents BIGINT;
-- Backfill: transform the existing data
UPDATE orders
SET total_amount_cents = ROUND(order_total * 100)::BIGINT
WHERE total_amount_cents IS NULL;
-- Add a check constraint to catch bad transforms before they propagate
ALTER TABLE orders ADD CONSTRAINT chk_cents_reasonable
CHECK (total_amount_cents > 0 AND total_amount_cents < 1000000000); -- Max ~$10M order
-- Validate the backfill
SELECT
COUNT(*) as total_rows,
COUNT(CASE WHEN total_amount_cents IS NULL THEN 1 END) as null_cents,
COUNT(CASE WHEN ABS(total_amount_cents / 100.0 - order_total) > 0.01 THEN 1 END) as mismatches
FROM orders;
-- Should show 0 null_cents and 0 mismatches before proceeding
Warning: Floating-point to integer conversion for financial data is not always
value * 100. Edge cases include values like0.1 + 0.2in IEEE 754 floating point, which equals0.30000000000000004. Always useROUND()and validate the conversion explicitly. Financial systems that skip this step have paid dearly for it.
Object storage data lakes present a uniquely challenging schema evolution problem because the data is immutable once written. You can't update old Parquet files the way you can update a database row. Schema evolution here is about making new queries work against a mixture of old and new file formats.
Parquet's column-based storage naturally supports some evolution patterns. When you read a Parquet file, the reader reconstructs the schema from the file footer — every Parquet file is self-describing. Delta Lake and Apache Iceberg add a transaction log on top of Parquet that tracks the schema across all files in a table, which is what enables table-level schema operations.
from delta import DeltaTable
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, StringType, LongType, DoubleType
spark = SparkSession.builder \
.config("spark.sql.extensions", "io.delta.sql.DeltaSparkSessionExtension") \
.config("spark.sql.catalog.spark_catalog", "org.apache.spark.sql.delta.catalog.DeltaCatalog") \
.getOrCreate()
# Current table schema (simplified)
DELTA_TABLE_PATH = "s3://data-lake/orders/delta/"
# Adding a new column to a Delta table — this is safe and instant
# Delta stores the schema in the transaction log, not by rewriting files
spark.sql(f"""
ALTER TABLE delta.`{DELTA_TABLE_PATH}`
ADD COLUMN fraud_score DOUBLE COMMENT 'ML fraud probability 0.0-1.0, null for orders pre-2024'
""")
# For Spark DataFrames, use mergeSchema option when writing
(
new_orders_df
.write
.format("delta")
.option("mergeSchema", "true") # Allows new columns; raises error on type changes
.mode("append")
.save(DELTA_TABLE_PATH)
)
The mergeSchema option is powerful but has limits. It handles additive changes. For type changes or column drops, you need overwriteSchema, which rewrites the table metadata — but importantly, it does NOT rewrite existing Parquet files. Old files still have the old types, and Spark will apply the coercion rules when reading.
Apache Iceberg has arguably the most sophisticated schema evolution support of any open table format. Its explicit schema versioning and partition evolution features make it worth understanding in depth:
from pyiceberg.catalog import load_catalog
from pyiceberg.schema import Schema
from pyiceberg.types import (
NestedField, StringType, LongType, DoubleType,
StructType, ListType, MapType
)
catalog = load_catalog("glue", **{
"type": "glue",
"warehouse": "s3://data-lake/iceberg/"
})
table = catalog.load_table("orders.order_events")
# Iceberg's update_schema() gives you a transactional schema update
with table.update_schema() as update:
# Add a new column
update.add_column("fraud_score", DoubleType(), "ML fraud probability")
# Rename a column (without losing data — Iceberg uses field IDs, not names)
update.rename_column("user_email", "email_address")
# Move a column (cosmetic, no data impact)
update.move_after("fraud_score", "status")
# This is the key insight: Iceberg uses field IDs, not field names,
# for internal column tracking. A rename changes only the name mapping
# in the metadata — no data files are rewritten.
The field ID system is what makes Iceberg's rename support genuinely safe. When you write a Parquet file in an Iceberg table, columns are identified by their field ID in the file footer. The human-readable name is just a label maintained in the table metadata. This means rename_column is a pure metadata operation that takes milliseconds and requires zero data movement.
Delta Lake lacks this feature — Delta uses column names directly, so a rename is effectively a two-step expand-contract at the table format level.
For large-scale systems where multiple teams consume the same data, a formal multi-version architecture becomes necessary. The goal is to decouple producer releases from consumer releases completely.
Borrow from the REST API world: your data pipeline exposes versioned "endpoints" (topics, views, or table snapshots), and consumers pin to a specific version. The pipeline operator maintains older versions for a defined support window.
class VersionedDataService:
"""
Manages multiple versions of a dataset, providing version-specific
views while maintaining a single underlying source of truth.
Consumers read from versioned views. Breaking changes introduce
a new version. Old versions are maintained for a deprecation window.
"""
SUPPORTED_VERSIONS = ["v1", "v2", "v3"]
DEPRECATED_VERSIONS = ["v1"] # Still available, but sunset date announced
CURRENT_VERSION = "v3"
def __init__(self, spark: SparkSession, base_path: str):
self.spark = spark
self.base_path = base_path
def get_view_for_version(self, version: str):
if version not in self.SUPPORTED_VERSIONS:
raise ValueError(
f"Version {version} not supported. "
f"Supported versions: {self.SUPPORTED_VERSIONS}"
)
if version in self.DEPRECATED_VERSIONS:
print(f"WARNING: {version} is deprecated. Please migrate to {self.CURRENT_VERSION}")
return self.spark.read \
.format("delta") \
.load(f"{self.base_path}/{version}/")
def materialize_v1_view(self, canonical_df):
"""
V1 view: flat structure, customer_id as string, no fraud_score
Derived from the canonical V3 dataset.
"""
return canonical_df.selectExpr(
"order_id",
"customer.id AS customer_id", # Flatten nested struct
"total_amount_cents",
"currency",
"status",
"created_at"
)
def materialize_v2_view(self, canonical_df):
"""
V2 view: nested customer object introduced, still no fraud_score
"""
return canonical_df.selectExpr(
"order_id",
"customer",
"total_amount_cents",
"currency",
"status",
"created_at"
)
def materialize_v3_view(self, canonical_df):
"""
V3 view: canonical format, all fields available
"""
return canonical_df # V3 IS the canonical format
def refresh_all_versions(self, canonical_df) -> None:
"""
Called from the main pipeline job. Writes all supported versions
from the canonical dataset.
"""
version_materializers = {
"v1": self.materialize_v1_view,
"v2": self.materialize_v2_view,
"v3": self.materialize_v3_view,
}
for version in self.SUPPORTED_VERSIONS:
view_df = version_materializers[version](canonical_df)
(
view_df.write
.format("delta")
.mode("overwrite")
.option("overwriteSchema", "true")
.save(f"{self.base_path}/{version}/")
)
print(f"Refreshed {version} view successfully")
The canonical representation here is V3 — the most complete, structurally correct schema. All older versions are derived from it at write time. This is an important design choice: the alternative (deriving V3 from V1) would mean your new features are constrained by your oldest schema, which is exactly backwards.
Technical patterns are necessary but not sufficient. The real prevention is organizational and process-level governance.
Every schema change should go through a review process similar to a code review. In practice, this means:
# .github/workflows/schema-compatibility-check.yml
name: Schema Compatibility Check
on:
pull_request:
paths:
- 'schemas/**/*.avsc'
- 'schemas/**/*.proto'
jobs:
check-schema-compatibility:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v3
with:
fetch-depth: 0
- name: Install dependencies
run: pip install confluent-kafka requests
- name: Run compatibility check
env:
SCHEMA_REGISTRY_URL: ${{ secrets.SCHEMA_REGISTRY_URL }}
run: |
python scripts/check_schema_compatibility.py \
--changed-schemas $(git diff --name-only origin/main HEAD -- 'schemas/') \
--registry-url $SCHEMA_REGISTRY_URL \
--fail-on-incompatible
# scripts/check_schema_compatibility.py
import argparse
import sys
import json
import requests
from pathlib import Path
def check_all_changed_schemas(changed_schema_files: list[str], registry_url: str) -> bool:
all_compatible = True
for schema_file in changed_schema_files:
schema_path = Path(schema_file)
if not schema_path.exists():
print(f"Skipping deleted schema: {schema_file}")
continue
schema_content = schema_path.read_text()
# Derive subject name from file path convention
# e.g., schemas/orders/order_events.avsc -> orders.order_events-value
parts = schema_path.parts
subject = f"{parts[-2]}.{schema_path.stem}-value"
response = requests.post(
f"{registry_url}/compatibility/subjects/{subject}/versions/latest",
json={"schema": schema_content},
headers={"Content-Type": "application/vnd.schemaregistry.v1+json"}
)
if response.status_code == 404:
print(f"No existing schema for {subject} — new subject, skipping compatibility check")
continue
result = response.json()
is_compatible = result.get("is_compatible", False)
if is_compatible:
print(f"✓ {subject}: compatible")
else:
print(f"✗ {subject}: INCOMPATIBLE — this change will break consumers")
print(f" Messages: {result.get('messages', [])}")
all_compatible = False
return all_compatible
if __name__ == "__main__":
parser = argparse.ArgumentParser()
parser.add_argument("--changed-schemas", nargs="+", required=True)
parser.add_argument("--registry-url", required=True)
parser.add_argument("--fail-on-incompatible", action="store_true")
args = parser.parse_args()
compatible = check_all_changed_schemas(args.changed_schemas, args.registry_url)
if not compatible and args.fail_on_incompatible:
sys.exit(1)
Beyond technical enforcement, define a communication protocol:
Additive changes (new optional fields): Engineer notifies the data-users Slack channel, describes the new field, and merges. No approval required.
Deprecations (marking a field for future removal): Engineer opens a data schema RFC issue, notifies all known consumers, sets a minimum 30-day sunset date before the field is eligible for removal. Documented in schema metadata.
Breaking changes requiring versioning: Requires approval from a data platform lead. Requires a migration plan document. Requires consumer team acknowledgment. Dual-write period must be defined and scheduled.
This exercise simulates a realistic schema evolution scenario you'll encounter in production.
Scenario: You're the data platform engineer at a fintech company. Your transaction_events Kafka topic uses Avro with the following V1 schema:
{
"type": "record",
"name": "TransactionEvent",
"namespace": "com.fintech.transactions",
"fields": [
{"name": "transaction_id", "type": "string"},
{"name": "account_id", "type": "string"},
{"name": "amount", "type": "double"},
{"name": "currency", "type": "string"},
{"name": "transaction_type", "type": {
"type": "enum",
"name": "TransactionType",
"symbols": ["DEBIT", "CREDIT", "TRANSFER"]
}},
{"name": "timestamp", "type": "long"}
]
}
You need to make the following changes:
risk_score field (float, 0.0–1.0, optional)amount needs to change from double to long (storing cents, not dollars) — this is a breaking type changeREVERSAL needs to be added to the enumaccount_id needs to be renamed to account_number (per new naming standards)Tasks:
For each of the four changes, classify them using the compatibility matrix from the first section.
Write the V2 Avro schema that safely handles the risk_score addition and the REVERSAL enum addition without breaking existing consumers. Verify it using the Schema Registry compatibility API.
Design the expand-contract migration plan for the amount type change in the downstream PostgreSQL transactions table. Write the SQL for all three phases and the validation query you'd run before Phase 3.
Design the dual-write and versioned topic strategy for the account_id → account_number rename. Write the Python producer code that handles the dual-write period, and the consumer migration strategy.
Write the CI/CD YAML and Python compatibility check script for your team's GitHub Actions workflow that validates all four changes before merge.
Challenge: What's the correct order to deploy these four changes? Does the ordering matter? Draw the dependency graph between the changes and justify your deployment sequence.
Teams often set NONE to get an urgent hotfix deployed, intending to restore proper compatibility mode afterward. Then they forget, or the team turns over, and six months later a legitimate breaking change goes undetected. Treat Schema Registry compatibility mode as infrastructure — managed as code, reviewed in PRs, never modified manually in production.
JSON Schemas look rigorous but are only validated if you explicitly validate them at the producer. Unlike Avro, which serializes with strict type enforcement, JSON messages go over the wire as text. A field declared as integer in the JSON Schema can contain a string value in the actual message if the producer doesn't validate before sending. For type-sensitive pipelines, prefer Avro or Protobuf over JSON with JSON Schema.
Backfill operations on large tables (running UPDATE ... WHERE column IS NULL after adding a new column) can take hours and cause lock contention. Always run backfills in batches and monitor table lock activity:
-- Batch backfill to avoid locking the table for hours
DO $$
DECLARE
batch_size INT := 10000;
rows_updated INT;
BEGIN
LOOP
UPDATE transactions
SET amount_cents = ROUND(amount * 100)::BIGINT
WHERE amount_cents IS NULL
AND ctid IN (
SELECT ctid FROM transactions
WHERE amount_cents IS NULL
LIMIT batch_size
);
GET DIAGNOSTICS rows_updated = ROW_COUNT;
EXIT WHEN rows_updated = 0;
RAISE NOTICE 'Updated % rows, sleeping briefly...', rows_updated;
PERFORM pg_sleep(0.1); -- 100ms pause between batches
END LOOP;
END $$;
Schema version and data version are different things. Schema V2 can contain records created during the V1 era (backfilled or transformed). When you query "how many V2 records exist," you might mean "records that were created by the V2 producer" or "records that conform to the V2 schema structure." Make this distinction explicit in your data lineage metadata, and add a schema_version field to records if you need to track producer version explicitly.
When you disable dual-write and stop producing to the old topic, you need to ensure all consumer groups on the old topic have processed all remaining messages before declaring migration complete. Don't turn off the old topic just because the producer stopped — there may be hours of unconsumed messages sitting in the partition:
from confluent_kafka.admin import AdminClient
from confluent_kafka import TopicPartition
def get_consumer_group_lag(bootstrap_servers: str, group_id: str, topic: str) -> dict:
admin_client = AdminClient({"bootstrap.servers": bootstrap_servers})
# Get topic end offsets
from confluent_kafka import Consumer
consumer = Consumer({
"bootstrap.servers": bootstrap_servers,
"group.id": "lag-checker-temp"
})
metadata = consumer.list_topics(topic)
partitions = [
TopicPartition(topic, p)
for p in metadata.topics[topic].partitions.keys()
]
end_offsets = consumer.get_watermark_offsets
# Fetch committed offsets for the consumer group
# (Requires AdminClient.list_consumer_group_offsets — Kafka 2.4+)
result = admin_client.list_consumer_group_offsets([group_id])
lag_by_partition = {}
# Calculate and return lag per partition
# Actual implementation depends on confluent-kafka version
return lag_by_partition
If you receive a 409 when trying to register a schema, the registry is telling you the schema is incompatible with the configured mode. Before changing the compatibility mode, understand why it's incompatible:
def diagnose_incompatibility(subject: str, new_schema: str) -> None:
"""
Detailed incompatibility diagnosis
"""
# Check against each previous version to find where compatibility breaks
versions_response = requests.get(f"{REGISTRY_URL}/subjects/{subject}/versions")
versions = versions_response.json()
for version in versions:
check_url = f"{REGISTRY_URL}/compatibility/subjects/{subject}/versions/{version}"
result = requests.post(
check_url,
json={"schema": new_schema}
).json()
if not result.get("is_compatible", False):
print(f"Incompatible with version {version}:")
for msg in result.get("messages", []):
print(f" - {msg}")
Schema evolution is one of those disciplines that separates data engineers who prevent fires from those who fight them. The core insight from this lesson is that there's no single solution — the right approach depends on your serialization format, pipeline architecture, consumer diversity, and change type.
Here's the strategic summary:
FULL_TRANSITIVE for critical topics.ALTER TABLE. Always use the three-phase approach.Next Steps in This Learning Path:
The investment you make in schema governance now pays compound interest. Every producer who learns to make additive changes by default, every engineer who knows to run the compatibility check before merging, every consumer that declares its version dependency explicitly — these practices accumulate into a system that evolves gracefully rather than catastrophically.
Learning Path: Data Pipeline Fundamentals