Schema Design Patterns for Event Sourcing on Kafka

Event sourcing captures state changes as an immutable, ordered sequence of events. Apache Kafka’s durable, partitioned log is a good fit for this, but the long-term health of an event-sourced system is decided by schema design. A poorly designed schema produces tight producer/consumer coupling, painful reprocessing, and an event log you cannot safely evolve. This guide gives concrete schema patterns for platform, SRE, and backend engineers running Kafka at scale.

The physical layout of your topics—partition counts, keying, and retention—is inseparable from logical schema design; see Topics, Partitions & Data Modeling. A single-partition topic preserves global order but caps consumer parallelism at one, so partition count and keying are part of the schema decision, not an afterthought. The patterns below assume a working grasp of those physical concepts.

The Event Envelope: A Wrapper for Metadata

Avoid publishing a “naked” business event. Wrap each event in an envelope that carries infrastructure metadata—data that is not part of the domain but is needed for tracing, debugging, schema routing, and deduplication. A bare event like {"userId": "123", "amount": 100} gives a consumer no way to tell what type it is, when it happened, or how to deserialize it.

At minimum the envelope should carry: a unique event ID, an event-type discriminator, a creation timestamp, and a payload-schema version. You can adopt the CloudEvents specification—whose Kafka binding maps attributes to record headers (binary mode) or serializes the whole event into the record value (structured mode)—or define a lightweight Avro envelope carried in the record value. A custom Avro envelope:

{
  "type": "record",
  "name": "EventEnvelope",
  "namespace": "com.mycompany.platform",
  "fields": [
    {"name": "event_id", "type": "string", "doc": "UUIDv4 for deduplication and tracing"},
    {"name": "event_type", "type": "string", "doc": "Fully qualified event type, e.g. com.mycompany.orders.OrderPlaced"},
    {"name": "timestamp", "type": {"type": "long", "logicalType": "timestamp-millis"}, "doc": "Event creation time at the source"},
    {"name": "source", "type": "string", "doc": "Service instance / host that produced the event"},
    {"name": "data_schema_version", "type": "int", "doc": "Monotonically increasing version of the payload schema"},
    {"name": "trace_id", "type": ["null", "string"], "default": null, "doc": "Distributed tracing ID"},
    {"name": "payload", "type": "bytes", "doc": "Avro-encoded binary of the business event"}
  ]
}

Note the timestamp field: a logical type annotates a base type, so logicalType must live inside the type object ({"type": "long", "logicalType": "timestamp-millis"}). Placing logicalType as a sibling of "type": "long" on the field is silently ignored by most Avro implementations—the field deserializes as a plain long—which is a common and hard-to-spot bug.

Because the envelope is decoupled from the payload, you can add metadata fields like tenant_id or correlation_id—each with a default so the change stays backward compatible—without touching domain schemas. The event_type field is the discriminator a consumer uses to pick the right schema for the payload bytes. This builds on Designing Event-Driven Schemas for Kafka Topics.

The tradeoff to make on purpose. An opaque-bytes payload buys you a stable envelope and per-event-type schema independence, but it costs you registry-enforced payload compatibility—the registry validates the envelope, not the bytes inside it. If you would rather the registry police every payload too, put the typed record directly in the record value (next section) and let the envelope’s metadata travel in Kafka record headers instead. Pick one and apply it per topic; mixing the two on the same topic is the source of most “why won’t this deserialize” tickets.

Strongly Typed Domain Events with a Schema Registry

With the envelope in place, define each domain event as a distinct, independently versioned record using Avro, Protobuf, or JSON Schema. A Schema Registry (Confluent’s, or any registry implementing the same API) stores these schemas and enforces compatibility checks so a producer cannot push a change that breaks existing consumers. An OrderPlaced event:

{
  "type": "record",
  "name": "OrderPlaced",
  "namespace": "com.mycompany.orders",
  "fields": [
    {"name": "order_id", "type": "string"},
    {"name": "customer_id", "type": "string"},
    {"name": "total_amount", "type": "double"},
    {"name": "currency", "type": "string", "default": "USD"},
    {"name": "line_items", "type": {
      "type": "array",
      "items": {
        "type": "record",
        "name": "LineItem",
        "fields": [
          {"name": "product_id", "type": "string"},
          {"name": "quantity", "type": "int"},
          {"name": "unit_price", "type": "double"}
        ]
      }
    }}
  ]
}

Compatibility modes: pick the guardrail, then pick the rollout order

The registry’s compatibility mode decides both what change is legal and who you upgrade first. For an event-sourced log the choice is load-bearing, because a full replay forces every historical version through current consumer code.

BACKWARD is the registry default and the right pick if you can always roll consumers ahead of producers. But for an append-only log you may replay end to end, FULL_TRANSITIVE is the strongest practical default: a replay can resurrect a v1 record years later, and only a transitive check guarantees current code still reads it. The Confluent Schema Registry compatibility documentation details each mode.

Subject naming decides how many event types a topic may hold

Compatibility is scoped to a subject, and the subject-naming strategy is the setting people forget until it bites:

• TopicNameStrategy (default) registers under <topic>-value and effectively allows one record type per topic—a second type fails the topic’s compatibility check.
• RecordNameStrategy registers under the fully qualified record name, so one topic can carry many event types (e.g. OrderPlaced, OrderShipped, OrderCancelled on one orders-events topic), each evolved independently.
• TopicRecordNameStrategy registers under <topic>-<recordName>, the same multi-type freedom but scoped per topic, so the same record name can evolve differently on different topics.

Event sourcing almost always wants multiple related event types in one aggregate-keyed topic to preserve ordering, so it almost always wants one of the non-default strategies. Note the cost: RecordName/TopicRecordName perform no intra-topic compatibility checking across types, so the discipline that TopicNameStrategy gave you for free now lives in your code review.

If you embed payloads in an envelope as in the previous section, register the payload schemas yourself (the bytes are opaque to the registry’s automatic serializer). If you instead put the typed record directly in the record value, the standard KafkaAvroSerializer registers it under the subject your naming strategy dictates.

Partitioning and Keying for Event-Sourced Streams

In event sourcing the aggregate ID is almost always the correct message key. For Order aggregates, use order_id. Kafka routes records with the same key to the same partition (given a fixed partition count and the default partitioner), which preserves per-aggregate ordering and lets stateful processors with Kafka Streams or ksqlDB co-locate an aggregate’s events.

Set the key on the Kafka record, not only inside the value. The payload still carries the aggregate ID for consumer-side validation:

ProducerRecord<String, byte[]> record = new ProducerRecord<>(
    "orders-events",
    orderPlacedEvent.getOrderId(),  // key = aggregate ID -> stable partition
    serializeEnvelope(envelope)     // value = serialized envelope
);
producer.send(record, (metadata, exception) -> {
    if (exception != null) {
        logger.error("Failed to publish event", exception);
    } else {
        logger.debug("Published to partition {} offset {}",
            metadata.partition(), metadata.offset());
    }
});

One reassurance and one gotcha:

• Adaptive partitioning does not touch keyed records. The default partitioner’s load-aware behavior introduced in KIP-794 (partitioner.adaptive.partitioning.enable, default true) only applies to records with a null key. A non-null key is always hashed to a partition, so your per-aggregate ordering is safe regardless of that setting—do not disable it expecting it to change ordering behavior.
• A null key silently breaks ordering. If serialization yields a null key (a mapping bug, or forgetting to set it), that event scatters across partitions by the sticky/adaptive logic and loses its aggregate ordering—with no error. Assert a non-null key in the producer path and alert if any aggregate topic shows records on more partitions than it has distinct recent keys.

Pick a partition count from your throughput ceiling and target consumer parallelism (max parallelism per group equals the partition count), keeping in mind that increasing partitions later changes key-to-partition mapping and breaks per-key ordering for existing keys—so over-provisioning modestly up front is cheaper than resharding a live event log. Cruise Control can rebalance existing partitions across brokers without downtime, but it does not change a topic’s partition count.

Schema Evolution Strategies for Long-Lived Events

Event-sourced logs live for years, so evolution is inevitable. The envelope’s data_schema_version is an explicit marker consumers can branch on, but the goal is to keep version proliferation low. Three strategies:

1. Prefer additive, default-bearing changes. Add optional fields with defaults; avoid renaming or removing fields that older data depends on. To retire a field, stop populating it and document it as deprecated rather than deleting it from a schema that historical events were written against.

2. Upcast in one place. When a consumer reads an older version, upcast it to the current internal model before business logic runs. This keeps version branching in a single layer instead of scattered conditionals.

3. Replay-test compatibility. Periodically replay the full log (or a representative window) through the latest consumer code in staging. This is what catches the evolution mistakes that compatibility checks alone miss—chiefly the ones a transitive check cannot see, like a semantic change (units switched from cents to dollars) that is structurally compatible but corrupts every replayed aggregate.

An upcaster in a Kafka Streams topology:

KStream<String, EventEnvelope> stream = builder.stream("orders-events");
KStream<String, OrderPlacedV2> upcasted = stream.mapValues(envelope -> {
    if (envelope.getDataSchemaVersion() == 1) {
        OrderPlacedV1 v1 = deserialize(envelope.getPayload(), OrderPlacedV1.class);
        return upcastToV2(v1);
    }
    return deserialize(envelope.getPayload(), OrderPlacedV2.class);
});

The Avro specification’s schema-resolution rules define exactly how a reader schema reconciles with a different writer schema; the registry’s compatibility modes are built on top of those rules.

Snapshots and Compaction

Reconstructing state by replaying the entire log gets slower as the log grows. The common remedy is to periodically write each aggregate’s current state to a compacted topic. Log compaction retains at least the latest record per key, which is exactly “current state per aggregate.”

Key the snapshot topic by aggregate ID; the value is the serialized aggregate state, ideally tagged with the source event offset it reflects. On startup a consumer loads the latest snapshot per key, then replays only events after that offset. Create the snapshot topic with cleanup.policy=compact:

kafka-topics --bootstrap-server localhost:9092 \
  --create --topic orders-snapshots \
  --partitions 12 --replication-factor 3 \
  --config cleanup.policy=compact \
  --config min.cleanable.dirty.ratio=0.1 \
  --config segment.ms=3600000 \
  --config max.compaction.lag.ms=3600000

How these settings actually behave:

• min.cleanable.dirty.ratio (default 0.5) sets the fraction of the log that must be “dirty” (uncompacted) before the cleaner runs. Lowering it makes compaction run more often. Do not set it near 0.01: it forces near-constant cleaning and significant broker CPU/IO load for little benefit. A modest value like 0.1 is a more defensible “compact aggressively” choice.
• max.compaction.lag.ms (default effectively disabled, Long.MAX_VALUE) is the better lever when you want bounded snapshot freshness: it guarantees a record becomes eligible for compaction within that window regardless of the dirty ratio. min.compaction.lag.ms (default 0) sets a lower bound if you instead need records to survive uncompacted for a while.
• segment.ms forces a segment to roll on a time bound. Only the active segment is exempt from compaction, so a smaller segment time lets recent keys be compacted sooner—at the cost of more, smaller segment files.

A snapshot is not a free win. Compaction guarantees at least the latest value per key but makes no promise about when an old value disappears, so a snapshot consumer must be correct against duplicates—load the latest, and let the source-offset tag break ties. And a full snapshot per aggregate can dwarf the events that produced it; if a serialized snapshot approaches the broker’s max.message.bytes, that is the signal your aggregate has grown too large and should be split rather than snapshotted whole.

See the Kafka log-compaction documentation for the exact guarantees, including tombstone (null-value) delete retention via delete.retention.ms.

Operational Concerns: Monitoring, Dead Letters, and Reprocessing

Monitor the registry, by the right metric. If you run a leader-based registry deployment, watch the leader-role JMX MBean—kafka.schema.registry:type=master-slave-role (value 1 = leader/primary, 0 = follower) in current Confluent builds. If no node reports as leader, schema registrations fail even though cached schema lookups keep working, which produces a confusing “reads fine, writes rejected” outage. Also alert on the Jersey per-endpoint error rate (kafka.schema.registry:type=jersey-metrics, e.g. the request-error-rate for the schema-registration endpoints). There is no schema-registry-master-slave-rate metric—do not alert on it.

Dead-letter the poison events. Some records will fail deserialization or processing. Route them to a dedicated dead-letter topic for inspection rather than blocking the partition. The envelope’s event_id and source make the origin traceable. Two failure modes worth distinguishing before you retry: a transient failure (a downstream timeout) is safe to replay from the DLQ, but a poison failure (a payload no current schema can deserialize) will fail identically on every retry—an automatic retry loop on those silently rebuilds consumer lag. Tag DLQ records with the exception class so you can route the two differently.

Reprocess via offset reset—on an inactive group. To replay after a consumer fix, reset offsets with kafka-consumer-groups. The reset is rejected unless the target group has no active members, so stop the group’s consumers first:

# All consumers in the group must be stopped, or the reset is rejected.
kafka-consumer-groups --bootstrap-server localhost:9092 \
  --group order-processor-v2 \
  --topic orders-events \
  --reset-offsets --to-earliest \
  --execute

Run --dry-run (the default when --execute is omitted) first to preview the new offsets. To replay only a window, use --to-datetime <ISO-8601, UTC> or --by-duration <ISO-8601 duration> instead of --to-earliest. Reprocessing under a fresh consumer group is the cleanest pattern—it leaves the live group untouched and sidesteps the inactive-group requirement entirely. Note that --to-datetime maps to offsets by the broker’s stored record timestamps, so its accuracy depends on whether the topic uses CreateTime (producer clock) or LogAppendTime (broker clock)—and under CreateTime, an out-of-order or back-dated producer timestamp can place the resolved offset earlier or later than you expect, so verify the preview against a known event before you execute.

These patterns—the metadata envelope, registry-enforced typed events, aggregate-keyed partitioning, single-layer upcasting, snapshotting onto a compacted topic, and operational tooling—form a coherent approach to event sourcing on Kafka. Applied consistently, they keep an event log evolvable and operable over the years it will outlive any single service that writes to it.