Integrating the Confluent Schema Registry

The Confluent Schema Registry turns a Debezium-to-Avro CDC path into a contract-enforced stream by validating every writer schema before a record reaches a…

The Confluent Schema Registry turns a Debezium-to-Avro CDC path into a contract-enforced stream by validating every writer schema before a record reaches a topic, so an incompatible ALTER TABLE fails a CI check instead of a downstream consumer at 3 a.m. This page wires the registry into the JSON to Avro transformation layer specifically — the AvroConverter properties, the subject-naming strategy, the BACKWARD_TRANSITIVE policy that lets a live stream survive source DDL, the CI ordering that registers schemas before the migration ships, and the wire-format prefix that lets any consumer decode a record it has never seen. All behavior is validated against PostgreSQL 14 through 17, Debezium 2.x, and Confluent Platform 7.x registries.

Skip the registry and the failure is deferred, not avoided. A producer that self-registers schemas will happily accept a breaking field the instant someone runs a migration, and the break surfaces hours later as a decode exception in an analytics job rather than as a rejected write at the source. The registry is the enforcement point that converts a silent, late, hard-to-attribute failure into a loud, early, obvious one — but only if you configure the compatibility policy and the registration ordering deliberately, because the defaults protect less than most teams assume.

CI-gated schema evolution order for a Debezium to Avro stream Five stages run left to right. First a CI job registers the new writer schema against the value subject. The registry gate then runs a BACKWARD_TRANSITIVE compatibility check across every prior version and returns is_compatible true before the migration is allowed to proceed. Consumers that understand the new field are deployed next, then the source DDL is applied, and finally producers begin emitting records that carry the new 4-byte schema id in the Confluent wire prefix. A red note warns that reversing this order ships DDL before the schema exists, so producers emit an id consumers cannot resolve. Register and verify before the DDL ships — never after is_compatible: true 1 2 3 4 5 CI job register writer schema Registry gate BACKWARD_TRANSITIVE check Deploy consumers understand new field Ship source DDL ALTER TABLE Producers emit new schema id on wire Reverse this order and the DDL ships before the schema exists — producers emit a schema id consumers cannot resolve, and the stream stalls.
The registry is a gate, not a logbook: the BACKWARD_TRANSITIVE check runs in CI before any DDL, so an incompatible change is caught while it is still a pull request rather than a stalled topic.

Compatibility Mode Semantics

The registry stores schemas under a subject — by default <topic>-value and <topic>-key under the TopicNameStrategy — and evaluates each new schema against the subject’s history according to its compatibility mode. For a CDC stream the mode is the single decision that determines which source DDL you can run without coordinating a consumer redeploy first. The table below maps each mode to the DDL it survives and the consequence it imposes on the running pipeline.

Compatibility mode Allowed schema change Reader / writer checked against CDC consequence
NONE Anything Nothing No protection. A breaking ALTER TABLE ships silently and consumers fail on the next record. Never use for a governed stream.
BACKWARD (default) Add optional field, delete field New schema vs. the immediately previous version New consumers read data written by the last writer. Does not protect a consumer replaying a compacted topic from an older version.
BACKWARD_TRANSITIVE Add optional field, delete field New schema vs. all prior versions New consumers read every historical record, including offset 0 of a compacted topic. The correct default for CDC.
FORWARD Add field, delete optional field Previous schema vs. the new version Old consumers read data written by the new writer. Useful when producers upgrade ahead of consumers.
FORWARD_TRANSITIVE Add field, delete optional field All prior schemas vs. the new version Every deployed consumer, old or new, reads the newest writer.
FULL / FULL_TRANSITIVE Add or delete only optional fields Both directions Strictest. Producers and consumers deploy in any order, but every added column must carry an Avro default.

Two properties turn these modes from advisory into enforced. Setting auto.register.schemas=false on the converter stops a producer registering a schema at runtime, so the only path to a new version is the CI job that runs the compatibility check first. Setting the subject policy to BACKWARD_TRANSITIVE — rather than leaving the fresh-registry default of BACKWARD — is what protects a log-compacted CDC topic whose earliest retained record may have been written under any prior schema. Both are set once, per subject, before the first record flows.

Diagnostic Patterns

The registry sits downstream of a replication slot, so a stalled stream is often a slot problem masquerading as a serialization problem. Confirm the slot is live before you touch the registry.

sql
-- Publisher: is the slot feeding the serializer still advancing?
SELECT slot_name,
       active,                                            -- expect true
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal,
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)) AS unconfirmed
FROM pg_replication_slots
WHERE slot_name = 'debezium_pg_analytics';
-- active = false with retained_wal climbing means the serializer stopped acknowledging WAL.

Once the slot is confirmed healthy, interrogate the registry directly. These calls belong in both the CI gate and the on-call runbook.

bash
# The effective compatibility mode for a subject (falls back to global if unset).
curl -s http://schema-registry:8081/config/orders-value | jq .
# Expect: {"compatibilityLevel":"BACKWARD_TRANSITIVE"}

# Dry-run a candidate schema BEFORE registering it — this is the CI gate.
curl -s -X POST \
  http://schema-registry:8081/compatibility/subjects/orders-value/versions/latest \
  -H 'Content-Type: application/vnd.schemaregistry.v1+json' \
  -d @order-v2.avsc.json | jq .
# Expect: {"is_compatible": true}  -> the migration may proceed

Verify the wire prefix on the topic to prove producers are actually framing records for the registry. A value that does not begin with the magic byte 0x00 was not written by a Confluent-framed serializer, and every consumer’s schema lookup will fail before Avro decoding starts.

bash
# First 5 bytes of a value: 0x00 magic + 4-byte big-endian schema id.
kcat -C -b kafka:9092 -t orders.avro -c 1 -f '%s' | xxd | head -1
# 00000000: 00 00 00 00 11 ...   -> magic 0x00, schema id 17
  • is_compatible: false in CI — the candidate schema breaks a prior version. Block the migration; do not override with NONE.
  • retained_wal > 5 GB and climbing — the serializer has stalled; the registry is a symptom, not the cause. Check the slot and the consumer, not the schema.
  • First value byte ≠ 0x00 — a non-Confluent producer wrote to the topic; consumers will fail schema resolution on every message.

Safe Deployment Sequence

The whole point of the registry is to make schema evolution a reviewed, ordered operation. Follow the five stages the diagram shows, keeping the compatibility check ahead of the DDL and a revert one command away.

1. Set the subject policy once, before the first schema. A fresh registry defaults to BACKWARD; a compacted CDC topic needs transitive coverage.

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

2. Configure the converter to trust the registry, not self-register. On the Debezium connector, both converters point at the registry and registration is disabled in production so no unreviewed producer can mint a version.

properties
key.converter=io.confluent.connect.avro.AvroConverter
value.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://schema-registry:8081
value.converter.schema.registry.url=http://schema-registry:8081
value.converter.auto.register.schemas=false
value.converter.use.latest.version=true

3. Register and verify in CI. Run the compatibility dry-run above; on is_compatible: true, POST the schema to .../subjects/orders-value/versions. If the check fails, the pipeline fails the build — the DDL never ships.

4. Deploy consumers, then ship the DDL. Roll out consumers that understand the new field first, then apply ALTER TABLE. Because the change is backward compatible, consumers already running against the old schema keep decoding while the new ones come up.

5. Let producers pick up the new writer schema, revert if needed. The serializer resolves and caches the new schema id on its next produce. If a downstream break appears, the revert is to lower nothing at the source — instead pin producers to the previous schema id (use.latest.version=false with the prior local schema) and roll consumers back; the registry keeps every version addressable by id, so no data is lost.

Pipeline Integration

On the Python side, the serializer fetches and caches the schema id on first use and prepends the wire prefix automatically, so the registry is contacted once per schema rather than once per record. Building the consumer to survive a brief registry outage is what keeps the whole path from turning the registry into a new single point of failure.

python
from confluent_kafka.schema_registry import SchemaRegistryClient
from confluent_kafka.schema_registry.avro import AvroSerializer
from confluent_kafka import Producer
import time, random

sr = SchemaRegistryClient({"url": "http://schema-registry:8081"})
with open("order.avsc") as f:
    serialize = AvroSerializer(sr, f.read(),
                               conf={"auto.register.schemas": False,
                                     "use.latest.version": True})

producer = Producer({"bootstrap.servers": "kafka:9092",
                     "enable.idempotence": True, "acks": "all"})

def emit_with_backoff(row, ctx, max_tries=5):
    backoff = 0.5
    for attempt in range(max_tries):
        try:
            producer.produce(topic="orders.avro", key=str(row["id"]),
                             value=serialize(row, ctx))       # cached id after first call
            return
        except Exception as exc:                              # registry unreachable, transient
            time.sleep(backoff + random.random() * 0.25)      # jitter avoids a reconnect storm
            backoff = min(backoff * 2, 30)
    log_dlq(row, "schema-registry unreachable")               # divert, never block the stream
  • Idempotent apply keyed on __lsn. Carry the source LSN in the Avro payload and have consumers drop any record whose LSN is not newer than the stored high-water mark, so an at-least-once replay after a connector restart converges instead of duplicating.
  • Cache-first resolution. Trust the serializer’s schema-id cache; do not validate against the registry on every call. A registry blip must degrade to backoff-and-DLQ, not a stalled pipeline.
  • Failover awareness. After a primary failover the connector may resume from an earlier offset and replay records under a schema id the registry still holds — the id is stable across the failover, so decode succeeds and the LSN guard suppresses the duplicates. The slot mechanics behind that replay are covered in Debezium connector configuration.

Downstream, the registry-framed records are partitioned, keyed, and dead-lettered by event routing and Kafka integration; a schema whose serialization fails is diverted to the dead-letter queue there rather than retried in place. The compression codec applied to those framed payloads is a separate decision covered in selecting Avro codecs for CDC payloads.

Authoritative references