Schema evolution leaks the table

Raw CDC inherits source DDL, so an ALTER TABLE RENAME silently breaks downstream consumers — unless a schema registry enforces a compatibility mode that rejects the incompatible registration before any event ships.

Previously

Now the pipeline can faithfully replay history and stream live changes — but it is emitting raw row deltas with whatever column names the DBA picked, which means the next ALTER TABLE leaks straight into every consumer. So we need something that catches incompatible changes BEFORE the bytes leave the producer.

Scene 07

Schema evolution leaks the table

Watch
Try it
Predict
Capture

What to watch for

The connector is streaming change events; the schema registry holds schema v1. Without a separate place to store schema versions, every consumer would have to redeploy on every DDL change. The component that does that storage is the schema registry, and the rules it enforces are called schema evolution. Watch what happens when the DBA adds a new column.

Implementation

Highlighted lines are the ones running in the diagram right now.

SchemaRegistry.register

Gate a candidate schema by compatibility mode.

1def register(subject, new_schema, mode):
2    old = versions[subject][-1]
3    if mode == NONE:
4        return accept(new_schema)
5    if mode == BACKWARD:
6        ok = can_read(reader=new_schema, data_of=old)
7    elif mode == FORWARD:
8        ok = can_read(reader=old, data_of=new_schema)
9    elif mode == FULL:
10        ok = (can_read(new_schema, old) and
11              can_read(old, new_schema))
12    return accept(new_schema) if ok else reject()

Connector.publishWithSchema

1def publish_with_schema(event, subject):
2    schema = avro_schema_of(event)
3    verdict = registry.register(subject, schema, MODE)
4    if verdict == REJECTED:
5        halt('cannot emit — incompatible schema')
6    schema_id = verdict.id
7    payload = magic_byte + schema_id + avro_encode(event)
8    kafka.produce(topic=subject, value=payload)
9    # consumer.deserialize() looks up schema by id

renameColumn

Avro rename = remove + add; aliases bridge the gap.

1def rename_column(old_name, new_name, aliases_enabled):
2    new_schema = drop_field(current, old_name)
3    if aliases_enabled:
4        new_schema = add_field(new_schema, new_name,
5                               aliases=[old_name])
6    else:
7        new_schema = add_field(new_schema, new_name)
8    # under BACKWARD, old reader looks for old_name:
9    #   with alias  -> resolves via aliases[]  -> accept
10    #   no alias    -> field missing           -> reject
11    return registry.register(subject, new_schema, MODE)

Sources

Where this sits in Build a CDC pipeline (Debezium + outbox)

Scene 07 of 12. Raw CDC inherits the source DDL — an ALTER TABLE RENAME COLUMN silently breaks downstream unless a Schema Registry enforces a compatibility mode that rejects the change at registration.

Up next. CDC of business tables couples consumers to internal table shape; the way out is to stop treating tables as the contract and start emitting domain events from a table that exists only to be published.

All 12 scenes in Build a CDC pipeline (Debezium + outbox) · Every curriculum

Built with Arqly

Every scene in Build a CDC pipeline (Debezium + outbox) builds on the one before it.All 12 Build a CDC pipeline (Debezium + outbox) scenes