Designing Topics and Event Contracts

In plain terms

A checkout team changes an amount field while the warehouse still expects the old representation. Whether that release is safe depends on the contract shared by the two applications. A topic is a long-lived interface: its name, key, schema, meaning, retention, and owner tell consumers what they can depend on. The log and its delivery mechanisms provide the setting in which these dependencies must hold.

The event design article wrote the tracking plan, the human agreement about what an event means. This article is the machine half: how that agreement becomes a schema the broker’s clients enforce, how topics are cut and keyed so that the guarantees from the Kafka article hold, and how a change is made without a coordinated deploy of every consumer. The lab builds a schema registry small enough to read and tests selected compatibility and validation cases.

Where one topic ends and the next begins

Topic boundaries balance isolation and shared consumption. Putting unrelated events together couples topic-level permissions, retention, throughput, and recovery. Splitting every event type can increase metadata, consumer filtering or joining, and operational work, but may be appropriate for independent lifecycles. Compare owner, entity key, sensitivity, retention, consumers, and failure isolation. Differences are reasons to consider a separate topic, not a universal rule. A topic can carry several event types if their schema strategy and processing semantics are explicit.

The naming convention encodes the boundary, so that a topic’s name tells a reader what it holds and who to ask: owner.entity with a suffix for the sensitivity class, such as orders.order, growth.user, identity.user.sensitive, or, for change streams from a database, cdc.orders. In this convention, a version suffix (.v2) marks a migration to a new topic; Kafka itself does not require this naming rule. The lab’s first exercise applies the rule to twelve event types and gets nine topics. Ordering is the reason entity is in the rule: everything about one order that must be read in sequence has to share a topic and a key, because the Kafka article’s ordering guarantee stops at the partition boundary.

The key and the partition count

Choose a partition key for records whose partition append order must stay together. Order IDs preserve order-level history; customer IDs group a customer’s orders but may create hot partitions. Neither reconstructs event-time order across concurrent producers. A stable salt derived from order ID can spread one customer while keeping each order together, but it loses customer-wide order. Changing from anonymous ID to user ID at login can also move a journey across partitions; document the identity mapping and how consumers reconcile the two histories.

Estimate partition count from peak bytes and records per second, measured per-partition throughput, and the slowest required consumer path. Add headroom, then test skew and failure recovery. Partitions consume replica storage, metadata, open files, memory, and recovery work even when traffic is low; creating ten times as many in anticipation of growth is not free. The count need not divide every group size. Increasing it can remap keys, so plan ordering and migration rather than treating it as a routine scale knob.

A stable business ID is not enough if its encoded key changes. Key-based partitioners use serialized key bytes, and compaction compares keys within each partition. Encoding order O17 differently can move new records to another partition or leave two byte-level keys for the same order. A tombstone under the new key will not delete the old key. Test key bytes as well as value-schema compatibility; when changing a key serializer or partition count, define how old state is retired and how order is preserved across the transition.

Retention, compaction, and what kind of thing a topic holds

An event records an occurrence; a change record describes a state transition; a state record supplies the current value for a key. Before/after images depend on the CDC source and configuration. Preserve history with delete retention when replay needs every event. For a state topic, compaction can remove superseded values; it is asynchronous, and combining it with delete retention can also remove old state. A Kafka null value is a tombstone, not a JSON object containing a null field. After delete markers expire, a stale state store may miss deletions. The contract must say whether to append, merge, or delete, and how to recover after missing the required history.

The record on the wire: format and registry

The broker stores bytes; clients interpret the application schema. JSON is readable, but field names alone do not define units, required fields, or business meaning. Avro resolves a writer schema against a reader schema. Protobuf identifies fields by number in its binary format and supports generated APIs. Encoded size depends on the actual data and compression. Protobuf binary-compatible renames can still break generated code or JSON consumers; reserve removed field numbers and names. Choose a format with the readers and migration path in mind.

A schema registry stores schema versions under a subject and checks the configured compatibility policy. Confluent’s TopicNameStrategy commonly uses topic-value; RecordNameStrategy and TopicRecordNameStrategy offer different compatibility scopes for multiple record types. The classic Confluent payload prefix is a version byte plus a four-byte schema ID. Avro data follows that prefix; Protobuf also includes message indexes. Other header modes and registries differ. The ID identifies the writer schema; it does not make every reader compatible. Clients need schema resolution, compatible versions, and access to retained schema definitions. Distinguish a corrupt or unknown ID from a temporary registry outage before quarantining a record.

A subject version is a position in one subject’s history; a schema ID identifies a registered schema within its registry scope. Version 2 is not necessarily schema ID 2, and an identical schema can be associated with several subjects. A valid ID alone does not prove that a record belongs on this topic. Enforce the intended subject and producer permissions as well as payload validation. The lab assigns a fresh ID on every registration and does not implement schema reuse, subject authorization, or registry contexts.

Compatibility: which changes are safe

Backward means a new reader can read old data; upgrade readers first when relying only on that direction. Forward means an old reader can read new data; it does not promise that a new reader can replay old history. Full checks both directions. Non-transitive policies compare against the latest version; transitive policies compare against all previous registered versions. Select a policy for the versions that coexist and the history that must be replayed. In Avro, a reader default fills a field absent from the writer schema; it does not make that field optional when writing. Null requires a type that permits null, such as ["null", "string"]. The table assumes the same record name and otherwise unchanged fields:

The lab checks a small subset of Avro-style field resolution. A required rename without aliases breaks both directions; defaults and reader aliases can change that result. Numeric promotion is structural compatibility, not a guarantee of precision: long-to-double can round large integers. A unit change with the same schema may pass registry checks while breaking business meaning, so review units, time semantics, IDs, examples, and consumer expectations separately. Protobuf and JSON Schema have their own rules; do not apply the Avro table to them.

For example, v1 has a string field with a default, v2 removes it, and v3 adds an integer field under the same name with an integer default. Each step can pass BACKWARD against only the latest schema, but v3 cannot read v1’s string as an integer. BACKWARD_TRANSITIVE checks that older pair too. Keep schemas needed by retained records; keeping topic bytes without their schema definitions is insufficient for replay.

Making a breaking change anyway

A structural break may be rejected by the registry; a dollars-to-cents change with the same type may pass. For an incompatible contract, create a versioned topic or an explicitly versioned envelope with compatible readers. Define a cutover boundary, stable event IDs, and which version each consumer processes around that boundary. If dual-writing, use a durable source/outbox or suitable Kafka transactions and reconcile both outputs; two unrelated sends can leave a gap. Dual-read consumers must deduplicate the overlap. A continuously written old topic has no final end to drain until a boundary or producer stop is defined. Validate output counts and meaning, retain rollback capability, and retire only after all owners confirm cutover and the promised replay window is satisfied. An empty group list alone does not prove no one depends on the topic.

Validation and dead letters

Check compatibility at schema registration and in CI; validate records before sending and at trust boundaries. Types alone cannot prove that a numeric amount uses cents, a timestamp has the right meaning, or an ID is legitimate. Add domain checks and reconciliation. A dead-letter path needs durable writes, source identity, reason, owner, retention, and a replay procedure; commit source progress only after the required output or quarantine write succeeds. Restrict or redact originals according to their contents rather than making the dead-letter topic a public copy of sensitive data. Retry transient infrastructure failures separately from invalid records. The lab simulates producer validation with in-memory lists; no Kafka topic is written.

The operational contract

The schema is the smaller half of a topic’s contract. The rest is the promise the owner makes about how the topic behaves, and it is one page per topic, kept next to the schema and reviewed with it:

topic            events.checkout                owner: checkout team      support: #checkout-data
kind             events (immutable)             key: stable actor_id; anonymous/user mapping is explicit
ordering         partition append order for a stable key                   partitions: 24            replication: 3, min.insync 2, acks=all
schema           events.checkout-value, Avro, compatibility FULL_TRANSITIVE; current v2; history in the registry
rate             60 records/s average, 900/s peak (launch days); records under 2 KB
retention        30 days total replay; local and remote retention configured consistently
sensitivity      pseudonymous identifiers; restricted according to linkage risk; ACL: checkout-api writes, listed groups read
consumers        raw-layer writer, fraud lane, funnel job (three groups; see the catalog)
change policy    compatible changes any time with a note; breaking changes as events.checkout.v2 with 60 days dual write
quality          dead-letter rate under 0.1%; oldest unprocessed event age alert at 5 minutes; incident channel #data-incidents

The rate and record-size lines support capacity tests; the replay window bounds backfills; the sensitivity line guides access review; and the consumer list identifies migration dependencies. Brokers expose configuration such as retention and partition count, but not the full business contract. Keep the document and actual settings in sync, including the key schema, compatibility policy, alerts, and owners.

This is a proposed operational contract. The lab schema below covers authenticated checkout records and requires user_id; it does not implement the anonymous identity mapping. The actor key is serialized separately from the value, and its schema and evolution policy need their own contract. The registry model accepts a policy on each call for comparison experiments; a production subject’s policy must be centrally configured and access-controlled.

The subscription company’s topics, contracted

Anti-patterns

• The firehose topic. Everything in events, so that anyone who may read anything may read everything.
• Topic per screen. Sixty topics, each with one reader, joined at query time with no ordering across them.
• The constant key. Every record uses the same tenant key, concentrating the load in one partition while the mapping stays fixed. Adding group members does not split that partition’s ordinary group assignment.
• Schemaless JSON on a shared topic. The producer added a field, renamed another, and the consumers found out from null columns.
• Compatibility off “for the migration”. Left off, and the next breaking change ships without anyone noticing.
• Changing meaning under the same name. Dollars became cents in schema v3; every check passed.
• Dropping invalid records. A dead-letter topic nobody created, so the collector skipped them and the totals came out 3% low.
• Mixed kinds in one topic. Events and state records interleaved, and a consumer that appended the state records as if they had happened.

Lab

The lab is a deliberately limited in-memory registry model. It supports primitive fields and nullable strings, reader defaults, three directional modes and their transitive forms. It omits nested records, aliases, enums, logical types, real registry APIs, and Avro binary encoding. Unknown modes and unsupported schemas are rejected rather than treated as compatible. Run both setup blocks before each independent exercise. Values on the wire are ordinary JSON behind a Confluent-style prefix, not production Avro or Confluent JSON Schema serialization.

import json, random, zlib, copy, math
from collections import Counter, defaultdict

PROMOTIONS = {("int", "long"), ("int", "double"), ("long", "double")}
MODES = {"BACKWARD", "FORWARD", "FULL"}
MODES |= {m + "_TRANSITIVE" for m in list(MODES)}
TYPES = {"string", "int", "long", "double", "boolean"}

def matches(value, kind):
    if kind == ["null", "string"]:
        return value is None or isinstance(value, str)
    if kind == "string":
        return isinstance(value, str)
    if kind == "boolean":
        return type(value) is bool
    if kind in ("int", "long"):
        bits = 32 if kind == "int" else 64
        return type(value) is int and -(2 ** (bits - 1)) <= value < 2 ** (bits - 1)
    if kind == "double":
        try:
            return type(value) in (int, float) and math.isfinite(float(value))
        except OverflowError:
            return False
    return False

def schema_check(schema):
    if not isinstance(schema, dict) or set(schema) != {"type", "name", "fields"}:
        raise ValueError("model schema requires only type, name, fields")
    if schema["type"] != "record" or not isinstance(schema["name"], str) or not schema["name"]:
        raise ValueError("named record required")
    if not isinstance(schema["fields"], list):
        raise ValueError("fields must be a list")
    seen = set()
    for f in schema["fields"]:
        if not isinstance(f, dict) or not {"name", "type"} <= set(f) or set(f) - {"name", "type", "default"}:
            raise ValueError("unsupported field declaration")
        if not isinstance(f["name"], str) or not f["name"] or f["name"] in seen:
            raise ValueError("field names must be unique nonempty strings")
        seen.add(f["name"])
        kind = f["type"]
        if not (isinstance(kind, str) and kind in TYPES) and kind != ["null", "string"]:
            raise ValueError("unsupported field type")
        if "default" in f and not matches(f["default"], kind):
            raise ValueError("default does not match field type")

def fields(schema):
    return {f["name"]: f for f in schema["fields"]}

def compatible_type(reader, writer):
    r = reader if isinstance(reader, list) else [reader]
    w = writer if isinstance(writer, list) else [writer]
    return all(any(x == y or (x, y) in PROMOTIONS for y in r) for x in w)

def can_read(reader, writer):
    """Resolution checks for the declared subset, not a full Avro checker."""
    schema_check(reader); schema_check(writer)
    problems, r, w = [], fields(reader), fields(writer)
    if reader["name"] != writer["name"]:
        problems.append("record names differ")
    for name, field in r.items():
        if name not in w and "default" not in field:
            problems.append(f"reader needs {name!r} and the writer never wrote it (no default)")
        elif name in w and not compatible_type(field["type"], w[name]["type"]):
            problems.append(f"{name!r} is {w[name]['type']} in the data but {field['type']} for the reader")
    return problems

def check(old, new, compatibility):
    if compatibility not in MODES:
        raise ValueError("unknown compatibility mode")
    mode = compatibility.removesuffix("_TRANSITIVE")
    problems = []
    if mode in ("BACKWARD", "FULL"):
        problems += ["BACKWARD: " + p for p in can_read(new, old)]
    if mode in ("FORWARD", "FULL"):
        problems += ["FORWARD: " + p for p in can_read(old, new)]
    return problems

def validate(record, schema):
    schema_check(schema)
    if not isinstance(record, dict):
        return "record must be an object"
    for field in schema["fields"]:
        name, kind = field["name"], field["type"]
        if name not in record:
            return f"missing {name}"
        value = record[name]
        if not matches(value, kind):
            return f"{name} should be {kind}, got {type(value).__name__}"
    if set(record) - set(fields(schema)):
        return "unknown record fields"
    return None

class Registry:
    """In-memory immutable-by-copy schema snapshots; policy is supplied per call."""

    def __init__(self):
        self._schemas, self.subjects, self.next_id = {}, {}, 1

    def get(self, schema_id):
        if type(schema_id) is not int or not 0 < schema_id < 2 ** 32:
            raise ValueError("schema id must be a positive uint32")
        return copy.deepcopy(self._schemas[schema_id])

    def register(self, subject, schema, compatibility="BACKWARD"):
        if not isinstance(subject, str) or not subject:
            raise ValueError("subject must be a nonempty string")
        if compatibility not in MODES:
            raise ValueError("unknown compatibility mode")
        schema_check(schema)
        history = self.subjects.get(subject, [])
        against = history if compatibility.endswith("_TRANSITIVE") else history[-1:]
        for schema_id in against:
            problems = check(self.get(schema_id), schema, compatibility)
            if problems:
                raise ValueError(f"{subject} v{len(history) + 1} rejected under {compatibility}: " + "; ".join(problems))
        schema_id = self.next_id
        self._schemas[schema_id] = copy.deepcopy(schema)
        self.subjects.setdefault(subject, []).append(schema_id)
        self.next_id += 1
        return schema_id

CHECKOUT_V1 = {"type": "record", "name": "checkout_started", "fields": [
    {"name": "event_id", "type": "string"}, {"name": "event_time", "type": "string"},
    {"name": "user_id", "type": "string"}, {"name": "amount_cents", "type": "long"},
    {"name": "currency", "type": "string"}, {"name": "item_count", "type": "int"}]}
registry = Registry()
print("registered checkout_started v1 as schema id", registry.register("events.checkout-value", CHECKOUT_V1))
# registered checkout_started v1 as schema id 1
print("a valid record:", validate({"event_id": "E1", "event_time": "2026-03-11T09:03:17Z", "user_id": "U1", "amount_cents": 6300, "currency": "USD", "item_count": 2}, CHECKOUT_V1))
# a valid record: None
print("a bad record:  ", validate({"event_id": "E2", "event_time": "2026-03-11T09:03:17Z", "user_id": "U1", "amount_cents": "63.00", "currency": "USD", "item_count": 2}, CHECKOUT_V1))
# a bad record:   amount_cents should be long, got str

The helpers validate the writer’s record, frame JSON with the schema ID, then validate the header and payload when decoding. Reader resolution fills defaults only when the writer schema lacks the field. A writer-declared field missing from a payload is malformed data, not an evolution default. The decoder returns an error for these cases so the caller can choose recovery or quarantine.

def encode(schema_id, record):
    """Validated JSON with a classic Confluent-style prefix; not Avro binary."""
    if type(schema_id) is not int or not 0 < schema_id < 2 ** 32:
        raise ValueError("schema id must be a positive uint32")
    try:
        writer = registry.get(schema_id)
    except KeyError:
        raise ValueError(f"unknown schema id {schema_id}") from None
    problem = validate(record, writer)
    if problem:
        raise ValueError(problem)
    return bytes([0]) + schema_id.to_bytes(4, "big") + json.dumps(record, allow_nan=False).encode()

def unique_object(pairs):
    result = {}
    for name, value in pairs:
        if name in result:
            raise ValueError("duplicate JSON key")
        result[name] = value
    return result

def reject_constant(value):
    raise ValueError(f"invalid JSON constant {value}")

def decode(message, reader_schema):
    if not isinstance(message, bytes) or len(message) < 6 or message[0] != 0:
        return None, "invalid frame"
    schema_id = int.from_bytes(message[1:5], "big")
    try:
        writer = registry.get(schema_id)
    except (KeyError, ValueError):
        return None, f"unknown schema id {schema_id}"
    try:
        raw = json.loads(message[5:].decode("utf-8"), object_pairs_hook=unique_object, parse_constant=reject_constant)
        problem = validate(raw, writer)
        if problem:
            return None, problem
        problems = can_read(reader_schema, writer)
        if problems:
            return None, problems[0]
        result, writer_fields = {}, fields(writer)
        for f in reader_schema["fields"]:
            name = f["name"]
            value = raw[name] if name in writer_fields else copy.deepcopy(f["default"])
            result[name] = float(value) if f["type"] == "double" else value
        return result, None
    except (ValueError, UnicodeError, TypeError, OverflowError) as error:
        return None, str(error)

1. Draw the topic boundaries. Twelve event types are listed with their owning team, sensitivity class, the entity they are keyed by, and their rate. Group them into topics by owner, entity, and sensitivity, name each topic by the convention, and print the result with each topic’s key and combined rate.

EVENTS = [
    ("product_viewed", "growth", "none", "user", 2000), ("cart_item_added", "growth", "none", "user", 400),
    ("checkout_started", "checkout", "none", "user", 60), ("payment_submitted", "checkout", "payment", "order", 50),
    ("order_placed", "orders", "none", "order", 45), ("order_shipped", "logistics", "address", "order", 44),
    ("order_refunded", "orders", "none", "order", 3), ("subscription_paused", "subscriptions", "none", "subscription", 5),
    ("subscription_resumed", "subscriptions", "none", "subscription", 4), ("user_logged_in", "identity", "credential", "user", 120),
    ("support_ticket_opened", "support", "personal", "ticket", 8), ("email_sent", "marketing", "personal", "user", 900),
]
topics = {}
for name, owner, pii, key, rate in EVENTS:
    topic = f"{owner}.{key}" + (".sensitive" if pii != "none" else "")
    entry = topics.setdefault(topic, {"key": key, "events": [], "rate": 0})
    entry["events"].append(name)
    entry["rate"] += rate
for topic, t in sorted(topics.items()):
    print(f"{topic:28} key={t['key']:13} {t['rate']:5d}/s  {', '.join(t['events'])}")
# checkout.order.sensitive     key=order            50/s  payment_submitted
# checkout.user                key=user             60/s  checkout_started
# growth.user                  key=user           2400/s  product_viewed, cart_item_added
# identity.user.sensitive      key=user            120/s  user_logged_in
# logistics.order.sensitive    key=order            44/s  order_shipped
# marketing.user.sensitive     key=user            900/s  email_sent
# orders.order                 key=order            48/s  order_placed, order_refunded
# subscriptions.subscription   key=subscription      9/s  subscription_paused, subscription_resumed
# support.ticket.sensitive     key=ticket            8/s  support_ticket_opened
print(len(EVENTS), "event types became", len(topics), "topics; one topic per owner, entity, and sensitivity class")
# 12 event types became 9 topics; one topic per owner, entity, and sensitivity class

2. Choose the key. Generate 15,000 status events for 5,000 orders, one customer placing 30% of them. For four candidate keys, measure the share of records in the busiest of twelve partitions and the number of orders whose three events land in different partitions.

rng = random.Random(0)
orders = []
for n in range(1, 5001):
    customer = "C1" if rng.random() < 0.3 else f"C{rng.randint(2, 400)}"
    order = f"O{n}"
    for status in ("placed", "paid", "shipped"):
        orders.append({"customer_id": customer, "order_id": order, "country": rng.choice(["US", "US", "US", "KR", "DE"]), "status": status})

def partition(key, count=12):
    return zlib.crc32(key.encode()) % count

def report(name, key_of):
    loads = Counter(partition(key_of(e)) for e in orders)
    by_order = defaultdict(set)
    for e in orders:
        by_order[e["order_id"]].add(partition(key_of(e)))
    split = sum(len(ps) > 1 for ps in by_order.values())
    print(f"key {name:24} busiest partition {max(loads.values()) / len(orders):5.1%}, orders whose events span partitions: {split}")

report("order_id", lambda e: e["order_id"])
report("customer_id", lambda e: e["customer_id"])
report("country", lambda e: e["country"])
report("customer_id + salt for C1", lambda e: e["customer_id"] + (e["order_id"][-1] if e["customer_id"] == "C1" else ""))
# key order_id                 busiest partition  8.9%, orders whose events span partitions: 0
# key customer_id              busiest partition 35.4%, orders whose events span partitions: 0
# key country                  busiest partition 60.4%, orders whose events span partitions: 3837
# key customer_id + salt for C1 busiest partition 12.9%, orders whose events span partitions: 0

3. Test six changes. Propose six changes to checkout_started: a nullable field with a default, a required field, a removed field, a rename, a widened type, and a changed type. Check each under backward, forward, and full compatibility and print the verdicts, then print why the rename fails.

def with_change(schema, **change):
    new = {"type": "record", "name": schema["name"], "fields": [dict(f) for f in schema["fields"]]}
    if "add" in change:
        new["fields"].append(change["add"])
    if "remove" in change:
        new["fields"] = [f for f in new["fields"] if f["name"] != change["remove"]]
    if "rename" in change:
        old_name, new_name = change["rename"]
        for f in new["fields"]:
            f["name"] = new_name if f["name"] == old_name else f["name"]
    if "retype" in change:
        name, kind = change["retype"]
        for f in new["fields"]:
            f["type"] = kind if f["name"] == name else f["type"]
    return new

proposals = [
    ("add optional coupon_code (default null)", with_change(CHECKOUT_V1, add={"name": "coupon_code", "type": ["null", "string"], "default": None})),
    ("add required channel (no default)", with_change(CHECKOUT_V1, add={"name": "channel", "type": "string"})),
    ("remove item_count", with_change(CHECKOUT_V1, remove="item_count")),
    ("rename user_id to customer_id", with_change(CHECKOUT_V1, rename=("user_id", "customer_id"))),
    ("amount_cents long -> double", with_change(CHECKOUT_V1, retype=("amount_cents", "double"))),
    ("amount_cents long -> string", with_change(CHECKOUT_V1, retype=("amount_cents", "string"))),
]
print(f"{'change':40} {'BACKWARD':9} {'FORWARD':9} FULL")
# change                                   BACKWARD  FORWARD   FULL
for label, proposal in proposals:
    verdicts = ["ok" if not check(CHECKOUT_V1, proposal, mode) else "breaks" for mode in ("BACKWARD", "FORWARD", "FULL")]
    print(f"{label:40} {verdicts[0]:9} {verdicts[1]:9} {verdicts[2]}")
# add optional coupon_code (default null)  ok        ok        ok
# add required channel (no default)        breaks    ok        breaks
# remove item_count                        ok        breaks    breaks
# rename user_id to customer_id            breaks    breaks    breaks
# amount_cents long -> double              ok        breaks    breaks
# amount_cents long -> string              breaks    breaks    breaks
print("why the rename breaks:", check(CHECKOUT_V1, proposals[3][1], "FULL"))
# why the rename breaks: ["BACKWARD: reader needs 'customer_id' and the writer never wrote it (no default)", "FORWARD: reader needs 'user_id' and the writer never wrote it (no default)"]

4. Read across versions. Register version 2 with the nullable coupon field and its default under full compatibility, encode one record under each version, and decode each with a consumer on the other version. Then decode a record carrying an id the registry does not know.

CHECKOUT_V2 = {"type": "record", "name": "checkout_started", "fields": CHECKOUT_V1["fields"] + [{"name": "coupon_code", "type": ["null", "string"], "default": None}]}
v2_id = registry.register("events.checkout-value", CHECKOUT_V2, compatibility="FULL")

base = {"event_id": "E9", "event_time": "2026-03-11T09:03:17Z", "user_id": "U1", "amount_cents": 6300, "currency": "USD", "item_count": 2}
old_message = encode(1, base)
new_message = encode(v2_id, dict(base, coupon_code="SPRING10"))
print("v2 consumer reads a v1 message:", decode(old_message, CHECKOUT_V2))
# v2 consumer reads a v1 message: ({'event_id': 'E9', 'event_time': '2026-03-11T09:03:17Z', 'user_id': 'U1', 'amount_cents': 6300, 'currency': 'USD', 'item_count': 2, 'coupon_code': None}, None)
print("v1 consumer reads a v2 message:", decode(new_message, CHECKOUT_V1))
# v1 consumer reads a v2 message: ({'event_id': 'E9', 'event_time': '2026-03-11T09:03:17Z', 'user_id': 'U1', 'amount_cents': 6300, 'currency': 'USD', 'item_count': 2}, None)
print("a message under an unregistered id:", decode(bytes([0]) + (99).to_bytes(4, "big") + old_message[5:], CHECKOUT_V1))
# a message under an unregistered id: (None, 'unknown schema id 99')
print("wire format: magic byte", old_message[0], "+ schema id", int.from_bytes(old_message[1:5], "big"), "+", len(old_message) - 5, "payload bytes")
# wire format: magic byte 0 + schema id 1 + 131 payload bytes

5. Validate at the producer. Generate a thousand checkout records, with every fortieth carrying the amount as a decimal string and every hundred-and-twenty-fifth missing the user id. Validate each against version 1 before encoding; send the valid ones and dead-letter the rest with the reason and headers.

rng = random.Random(1)
produced, dead_letter = [], []
for n in range(1, 1001):
    record = {"event_id": f"E{n}", "event_time": "2026-03-11T09:03:17Z", "user_id": f"U{rng.randint(1, 300)}",
              "amount_cents": rng.randint(500, 20000), "currency": "USD", "item_count": rng.randint(1, 5)}
    if n % 40 == 0:
        record["amount_cents"] = str(record["amount_cents"] / 100)
    if n % 125 == 0:
        del record["user_id"]
    problem = validate(record, CHECKOUT_V1)
    if problem:
        dead_letter.append({"reason": problem, "original": record, "headers": {"topic": "events.checkout", "producer": "checkout-api", "schema_id": 1}})
    else:
        produced.append(encode(1, record))
print(len(produced), "messages produced,", len(dead_letter), "sent to events.checkout.dlq")
# 968 messages produced, 32 sent to events.checkout.dlq
print("dead-letter reasons:", Counter(d["reason"] for d in dead_letter))
# dead-letter reasons: Counter({'amount_cents should be long, got str': 24, 'missing user_id': 8})
print("one dead letter:", json.dumps(dead_letter[0])[:150])
# one dead letter: {"reason": "amount_cents should be long, got str", "original": {"event_id": "E40", "event_time": "2026-03-11T09:03:17Z", "user_id": "U212", "amount_ce

6. Compaction and tombstones. Build a small customer-state log with updates and one deletion as a null value. Compact it, build the state table with a consumer that honours tombstones and one that ignores them, and then show what happens to a consumer that stopped before the deletion and resumes after the broker has removed old tombstones.

customer_updates = [("C1", {"segment": "basic"}), ("C2", {"segment": "basic"}), ("C1", {"segment": "premium"}),
                    ("C3", {"segment": "basic"}), ("C2", None), ("C4", {"segment": "premium"}), ("C1", {"segment": "premium", "country": "KR"})]

def compact(log):
    latest = {}
    for key, value in log:
        latest[key] = value
    return list(latest.items())

def build_table(table, log, honour_tombstones=True):
    for key, value in log:
        if value is None:
            if honour_tombstones:
                table.pop(key, None)
        else:
            table[key] = value
    return table

print("the latest-value view keeps one entry per key, tombstone included:", compact(customer_updates))
# the latest-value view keeps one entry per key, tombstone included: [('C1', {'segment': 'premium', 'country': 'KR'}), ('C2', None), ('C3', {'segment': 'basic'}), ('C4', {'segment': 'premium'})]
print("consumer honouring tombstones:", build_table({}, customer_updates))
# consumer honouring tombstones: {'C1': {'segment': 'premium', 'country': 'KR'}, 'C3': {'segment': 'basic'}, 'C4': {'segment': 'premium'}}
print("consumer ignoring tombstones: ", build_table({}, customer_updates, honour_tombstones=False))
# consumer ignoring tombstones:  {'C1': {'segment': 'premium', 'country': 'KR'}, 'C2': {'segment': 'basic'}, 'C3': {'segment': 'basic'}, 'C4': {'segment': 'premium'}}
stale = build_table({}, customer_updates[:4])
without_tombstones = [(k, v) for k, v in compact(customer_updates) if v is not None]
print("a consumer that stopped before the delete and resumes after the broker removed old tombstones:", build_table(stale, without_tombstones))
# a consumer that stopped before the delete and resumes after the broker removed old tombstones: {'C1': {'segment': 'premium', 'country': 'KR'}, 'C2': {'segment': 'basic'}, 'C3': {'segment': 'basic'}, 'C4': {'segment': 'premium'}}

7. Write the contract. Produce the one-page operational contract for orders.order from the subscription company’s table, in the format of the events.checkout example, and add the migration plan you would follow if amount_cents had to become a decimal string.

References

Checked on 2026-09-09. Avro field examples follow the linked 1.12 specification; registry framing and subject strategies are Confluent-specific. The Python model implements only the subset stated in the lab.

• Avro 1.12 specification: defaults and schema resolution
• Confluent schema evolution and transitive compatibility
• Confluent serializers, subjects, and wire formats
• Protobuf proto3 language guide
• Kafka 4.1 design: log compaction