Event-Driven Architectures and Event Sourcing

4.1 Log-Structured Event Streams

A log is an append-only sequence of records, each with a monotonically increasing offset. That is the whole data structure, and its power comes from what it does not do. It does not delete on read, it does not update in place, and it does not track per-consumer state.

A traditional message queue deletes a message once a consumer acknowledges it, so the broker tracks per-message state, delivery becomes a coordination problem, and replay is impossible. A log keeps records for a retention period and stores one number per consumer group. Replay is resetting that number.

Segments, indexes, and the read path

A partition is a directory of segment files. Each segment holds a contiguous range of offsets and comes with two index files: an offset index mapping a relative offset to a byte position, and a time index mapping a timestamp to an offset. Both indexes are sparse, with one entry per few kilobytes of data, so they stay small enough to memory-map.

  /var/lib/kafka/orders-3/
    00000000000000000000.log     1 GB, offsets 0 .. 4,182,993
    00000000000000000000.index   sparse: offset -> byte position
    00000000000000000000.timeindex sparse: timestamp -> offset
    00000000004182994.log        1 GB, offsets 4,182,994 .. 8,401,220
    00000000004182994.index
    00000000004182994.timeindex
    00000000008401221.log        active segment, open for append
    ...
    leader-epoch-checkpoint      epoch -> start offset, for truncation

  fetch(offset = 4,900,000):
    1. binary search the file names        -> segment 00000000004182994
    2. binary search the .index            -> byte position 61,440
    3. scan forward from there             -> at most a few KB
    4. sendfile() from that position       -> page cache straight to socket

  The broker process never touches the record bytes on a hot read.

Two mechanisms give the design its throughput, and both come from the operating system rather than from clever data structures.

Writes go to the page cache and the kernel flushes them later. The broker does not fsync per record by default. Durability comes from replication instead: a record is durable once enough replicas hold it, and the cluster tolerates the loss of any one machine's page cache. Section 3.1 priced an fsync, and this design avoids paying it per record.

Reads use sendfile, which Section 1.2 described. Bytes go from the page cache to the socket inside the kernel with no copy into user space. A consumer reading recent data reads from memory, and the bytes never enter the broker's address space. This is why one broker sustains gigabytes per second on modest CPU. It is also why a consumer that falls far behind hurts: its reads miss the page cache, hit the device, and evict the pages that current consumers need.

Replication and the durability contract

Each partition has a leader and a set of followers. A follower that keeps up belongs to the in-sync replica set. The producer's acks setting decides when a write is acknowledged: acks=0 returns immediately, acks=1 returns when the leader has it, and acks=all returns when every in-sync replica has it.

Leader epochs solve a subtler problem. Without them, a follower that rejoins after a leader change cannot always tell which of its records were committed, and truncation by high watermark alone can lose data or keep a divergent record. The leader epoch file maps each epoch to its starting offset, so a returning follower asks the leader where its epoch ended and truncates exactly there.

KRaft replaced ZooKeeper for cluster metadata. The controller quorum runs Raft over an internal metadata topic, which removes an external dependency, cuts failover time, and raises the supported partition count. ZooKeeper mode was removed in Kafka 4.0.

Tiered storage moves closed segments to object storage and keeps recent segments local. Local disk then sizes to the hot window rather than to the retention period, which makes long retention affordable and makes rebalancing fast, because a moved partition copies only local segments. Diskless designs go further and write partition data straight to object storage. KIP-1150 was accepted in March 2026 as a directional proposal, with the core and coordinator proposals still under discussion later that year, so treat it as a direction rather than a shipped capability.

Producer batching

A producer does not send one record per request. It accumulates records per partition in a buffer and sends a batch when the batch reaches batch.size or when linger.ms elapses. The batch is compressed as a unit, which compresses far better than per-record compression, and the broker stores it compressed and serves it compressed.

linger.ms is the one knob that teams leave at zero and regret. At zero, the producer sends as soon as a record arrives, so batches are small, compression is poor, and request rate is high. Setting it to 5 to 20 milliseconds adds that much latency and often multiplies throughput several times. For a pipeline where 20 milliseconds is invisible, it is free throughput.

Consumer groups and the rebalance protocol

A consumer group divides partitions among its members. Each partition goes to exactly one member, so parallelism caps at the partition count, and a group with more members than partitions leaves the extras idle.

The group coordinator is a broker chosen by hashing the group identifier. The protocol runs in two phases. In JoinGroup, every member sends its subscription and the coordinator picks one member as the group leader. In SyncGroup, the leader computes the assignment and sends it to the coordinator, which distributes it. Putting the assignment logic in a client rather than in the broker is what lets a group use a custom assignor.

The classical rebalance stops the world. Every member revokes every partition, the assignment is computed, and every member starts again. Processing stops for the whole group during that window.

Cooperative sticky assignment fixes most of it. Members keep the partitions they will retain and revoke only those that move, over two short rounds instead of one long stop. Static group membership goes further: each member gets a stable group.instance.id, so a rolling restart triggers no rebalance at all as long as the member returns within the session timeout.

Consumer lag is the primary health metric, and it must be read in seconds rather than in records. A thousand-record lag means one second on a busy topic and one hour on a quiet one. Compute lag time from the timestamp of the next unconsumed record, which the time index makes cheap.

Change data capture

Change data capture reads a database's write-ahead log and emits one event per row change. It gives a complete change stream with no application code and no polling load on the database. Debezium is the common implementation, reading PostgreSQL logical replication, MySQL binary logs, and MongoDB change streams.

Section 2.4 introduced the outbox pattern. Change data capture is the efficient relay for it: the outbox row appears in the write-ahead log, the connector publishes it, and no polling query runs.

Two operational facts matter more than the setup. A logical replication slot in PostgreSQL retains write-ahead log segments until the consumer reads them, so a stopped connector fills the primary's disk and eventually stops the database. Monitor replication slot lag as a first-class alert with a page, not a ticket. And an initial snapshot of a large table produces a burst that overwhelms downstream consumers, so use an incremental snapshot mechanism that interleaves snapshot chunks with live changes.

4.2 Event Sourcing and CQRS

Event sourcing stores the sequence of changes rather than the current state. Current state is a fold over the events. The event log is the source of truth and it is append-only.

What the pattern buys

Four properties follow, and none of them is available from a state-storing design without extra work.

The audit trail is free, because the events are the audit trail. Temporal queries are free, because folding events up to time T gives the state at time T. Debugging by replay works, because a bug reproduces from the exact event sequence. And a new read model can be built at any time, because a new projection replays history.

The write path is a specific shape. A command arrives, the aggregate loads by replaying its events from the last snapshot, it validates the command against its invariant, and it appends new events with an expected version. The expected version is optimistic concurrency: if another writer appended in the meantime, the append fails and the command retries against the new state.

class EventStore:
    """Append with an expected version. The uniqueness constraint on
    (stream_id, version) is what makes concurrency control work."""

    def append(self, stream_id: str, expected_version: int,
               events: list[Event]) -> None:
        with self.db.transaction():
            current = self.db.scalar(
                "SELECT COALESCE(MAX(version), 0) FROM events "
                "WHERE stream_id = %s", stream_id)
            if current != expected_version:
                raise ConcurrencyConflict(stream_id, expected_version, current)

            rows = [(stream_id, expected_version + i + 1, e.type,
                     e.payload, e.occurred_at)
                    for i, e in enumerate(events)]
            self.db.executemany(
                "INSERT INTO events "
                "(stream_id, version, type, payload, occurred_at) "
                "VALUES (%s, %s, %s, %s, %s)", rows)

    def load(self, stream_id: str) -> tuple[list[Event], int]:
        snap = self.snapshots.latest(stream_id)          # may be None
        from_version = snap.version if snap else 0
        events = self.db.query(
            "SELECT version, type, payload FROM events "
            "WHERE stream_id = %s AND version > %s ORDER BY version",
            stream_id, from_version)
        return events, (events[-1].version if events else from_version)

The three costs

Replay cost grows with history. An aggregate with 500,000 events takes too long to load. Snapshots solve it: store the folded state every N events and replay only the tail. Pick N so that replay stays under a few milliseconds, which is usually between 100 and 1,000 events. A snapshot is a cache, not a source of truth, so it must be safe to delete and rebuild.

Schema change is permanent. An event written in 2019 must still be readable in 2029, because the log never gets rewritten. Upcasting solves it: a small function converts a version 1 event into a version 2 event at load time, and the chain of upcasters grows over time. Keep events small and factual, because a small factual event survives schema change and a large event carrying derived data does not.

Deletion conflicts with an append-only log. A privacy regulation that gives a person the right to erasure meets a log that never deletes. Crypto shredding resolves it: encrypt each subject's personal data with a per-subject key, store the keys in a mutable store, and delete the key on an erasure request. The events remain and their personal content becomes unreadable.

Command query responsibility segregation

CQRS uses different models for writes and reads. The write model enforces invariants and stays normalized. The read model serves queries and denormalizes freely. A projection keeps the read model current from the write model's events.

The separation exists because the two sides have different requirements. The write side needs a small consistency boundary and an aggregate shape. The read side needs whatever shape answers the query in one lookup, which is usually a wide denormalized row or a document. Forcing one schema to serve both produces a schema that serves neither well.

   command                 +--------------------------+
   SubmitOrder ----------> | load aggregate:          |
                           |   snapshot + event tail  |
                           | validate invariant       |
                           | append with expected ver |
                           +------------+-------------+
                                        |
   EVENT LOG (append only, the source of truth)
   +-----+-----+-----+-----+-----+-----+-----+-----+-----+
   | e1  | e2  | e3  | e4  | e5  | e6  | e7  | e8  | e9  |
   +-----+-----+-----+-----+-----+-----+-----+-----+-----+
      |             |                    |            |
      v             v                    v            v
  +---------+  +-----------+     +--------------+  +-----------+
  | order   |  | customer  |     | search index |  | analytics |
  | summary |  | timeline  |     | (Elastic)    |  | warehouse |
  | Postgres|  | Redis     |     |              |  | Iceberg   |
  +---------+  +-----------+     +--------------+  +-----------+
   each projection tracks its own position in the log and can be
   deleted and rebuilt independently, at its own pace

Eventual consistency reaches the user interface

A projection lags the write. A user submits an order and then reads the order list, and the projection has not caught up. The user sees their action apparently fail.

Four techniques address it, and the right one depends on the interaction. Read the write side for a short window after a write, which is correct and loads the write model. Return the expected version from the write and have the read wait for the projection to reach it, which is precise and adds latency. Render the result optimistically on the client from the command it just sent, which is instant and lies if the command later fails. Or route that user's reads to the projection instance that consumed their write, which gives read your writes with sticky routing.

Projection rebuilds deserve a design of their own. A rebuild replays the whole log, which can take hours. Doing it in place makes the read model unavailable for that time. The usual approach builds the new projection alongside the old one, tracks both positions, and switches reads over once the new one catches up, which is the same expand-and-contract shape that a schema migration uses.

4.3 Exactly-Once Processing Semantics

Three delivery levels exist, and the names hide how much work the third one takes.

At-most-once delivers zero or one times. The consumer commits its offset before processing, and a crash after the commit loses the record. It is correct for telemetry where a gap is acceptable.

At-least-once delivers one or more times. The consumer commits after processing, and a crash before the commit reprocesses. This is the default almost everywhere, and it requires the downstream effect to be idempotent.

Exactly-once means the effect happens once, whatever fails. It does not mean the message is delivered once, because that is impossible over an asynchronous network. It means the visible result is as if it were.

The three mechanisms

The idempotent producer attaches a producer identifier and a monotonically increasing sequence number per partition to every batch. The broker tracks the last sequence number it accepted per producer per partition and discards a duplicate that a retry created. This removes duplicates caused by producer retries and nothing else.

Transactions let a producer write to several partitions and commit the consumer offsets atomically. A transaction coordinator, itself a broker, holds the transaction state in an internal topic. The producer registers, begins, writes, and commits. The coordinator writes commit or abort markers into every partition the transaction touched.

The consumer side uses isolation.level=read_committed. Such a consumer never returns records from an aborted transaction, and it never returns records beyond the last stable offset, which is the offset of the earliest still-open transaction. That is the mechanism, and it has a consequence worth stating: an open transaction blocks read-committed consumers from advancing past it, so a hung transaction stalls consumers even though the partition keeps accepting writes.

  partition log:
  offset:  100  101  102  103  104  105  106  107  108
  content: [A]  [B]  [T1] [T1] [C]  [T2] [T1] [COM] [T2]
                      ^open transaction T1 begins here
                                                ^T1 commit marker

  Last Stable Offset (LSO) = 102 while T1 is open.
  A read_committed consumer can read 100 and 101 and then waits,
  even though C at offset 104 is a plain record that committed.

  Once the T1 commit marker lands at 107, the LSO jumps to 108 and
  the consumer receives 102, 103, 104 and 106. T2 records stay hidden
  until T2 resolves.

  transaction.timeout.ms bounds how long this can last. A producer
  that dies mid-transaction is fenced and its transaction aborted
  after that timeout, not before.

// Read, process, write, and commit offsets as one atomic unit.
producer.initTransactions();

while (running) {
    ConsumerRecords<String, String> records = consumer.poll(POLL_TIMEOUT);
    if (records.isEmpty()) continue;

    producer.beginTransaction();
    try {
        for (ConsumerRecord<String, String> record : records) {
            for (ProducerRecord<String, String> out : transform(record)) {
                producer.send(out);
            }
        }
        // The offsets commit inside the same transaction as the output.
        producer.sendOffsetsToTransaction(
                offsetsOf(records), consumer.groupMetadata());
        producer.commitTransaction();
    } catch (ProducerFencedException | OutOfOrderSequenceException e) {
        producer.close();          // another instance took over; stop
        throw e;
    } catch (KafkaException e) {
        producer.abortTransaction();   // safe to retry the whole batch
    }
}

The guarantee stops at the broker boundary

The transaction covers writes into the log and the offset commit. It does not cover an external system. A pipeline that reads the log, writes PostgreSQL, and commits an offset involves two systems with no shared transaction.

Two options close the gap. Make the external write idempotent using a natural key or a deduplication table keyed by the record's topic, partition, and offset, which is the simplest correct answer. Or use a sink connector that implements two-phase commit against the external system, which Flink's two-phase-commit sink provides and which inherits the blocking behavior that Section 2.4 described.

The cost of transactions in Kafka is a throughput reduction in the range of 10 to 30 percent plus the commit latency, which is why it is not the default. Idempotent consumers with at-least-once delivery are simpler and cheaper, and they reach the same outcome for most workloads.

Deduplication windows and out-of-order records

An idempotent consumer needs to recognize a record it already processed. The key is usually a business identifier plus a version, or the topic, partition, and offset triple.

The store for those keys must be bounded, because it cannot grow forever. A time-windowed store keeps keys for longer than the maximum retry and reprocessing window. Getting the window wrong in either direction has a cost: too short and a late duplicate slips through, too long and the store dominates memory.

Out-of-order records are a separate problem. Order is guaranteed within a partition and nowhere else, so any join or aggregation across partitions sees records out of order. Three answers exist. Key both streams so that related records land in the same partition, which removes the problem entirely. Buffer by event time and emit when a watermark passes, which Section 8 of Volume I covers and which trades latency for completeness. Or make the operation order-insensitive, which is what a CRDT-shaped aggregation does.

4.4 Schema Registries and Event Contracts

An event is a contract between a producer and consumers that the producer does not know about. A change that breaks a consumer is discovered in production unless something validates the change first.

Compatibility modes

A schema registry stores schemas by subject, assigns each one an identifier, and rejects a new version that violates the configured compatibility rule. Producers write the identifier in the record header, and consumers fetch the schema by identifier and cache it.

The upgrade order column is the operational content of the table. Backward compatibility means consumers must be upgraded first, because the new consumer reads both old and new data. Forward compatibility means producers go first. Full compatibility removes the ordering constraint entirely, which is why it is the right default for an event that many teams consume.

Avro and Protobuf differ in where the schema lives

Avro writes no field names or identifiers in the payload. The data is a positional encoding, and reading it requires the exact writer schema. The reader supplies its own schema, and Avro resolves the two by field name, applying defaults for fields the writer did not have. That resolution step is what gives Avro its strong evolution story, and it is why an Avro consumer cannot read a record without contacting the registry.

Protobuf writes a field number and a wire type per field, as Section 1.3 described. A reader skips unknown fields using the wire type, so it decodes without the writer schema at all. Evolution rules follow from the encoding: adding a field with a new number is safe, removing one is safe if the number is reserved, and changing a type is safe only within a compatible wire type group.

The practical difference is where a mistake is caught. Avro catches it at registration, because resolution is explicit and the registry can compute compatibility precisely. Protobuf catches some at registration and lets others through, because the wire format tolerates changes that the application does not.

Rules that keep an event contract stable

Five rules cover most of what goes wrong.

Never change the meaning of an existing field. Adding status = "PENDING_V2" to an enumeration that consumers switch on is a breaking change even though the schema validates. Meaning lives outside the schema, so it needs a review rather than a tool.

Give every field a default. A field with a default is removable later and addable now in both directions, which is what makes full compatibility achievable.

Never reuse an identifier or a name. A reused Protobuf field number decodes old data into a new field with a wrong type or a wrong meaning. Mark removals reserved and let the tool enforce it.

Version the event type, not only the schema. When a change is genuinely breaking, publish OrderSubmitted.v2 alongside v1, run both for a migration window, and retire v1 once consumer lag on it reaches zero. A breaking change forced through one subject is an outage.

Keep events factual and small. An event that carries derived data becomes wrong when the derivation changes, and it cannot be corrected because the log is immutable. Carry the facts and let each consumer derive what it needs.

Dead letter queues and retry topologies

A consumer that cannot process a record has three options: retry, skip, or park it. Retrying forever blocks the partition, because offsets advance in order. Skipping loses data silently. Parking it in a dead letter queue keeps the partition moving and preserves the record for inspection.

A tiered retry topology separates transient failures from permanent ones. The consumer retries in memory a few times with backoff, which handles a brief dependency blip. A record that still fails goes to a retry topic with a delay, and a separate consumer reads that topic after the delay. A record that fails the last retry tier goes to the dead letter topic.

  orders ---> [ consumer ]
                  |  transient failure
                  |  retry 3x in process, 50ms / 200ms / 1s backoff
                  |  still failing
                  v
            orders.retry.5s ---> [ delay consumer ] --+
                  ^                                    | still failing
                  |                                    v
                  |                          orders.retry.1m ---> ...
                  |                                    | still failing
                  |                                    v
                  |                          orders.retry.10m
                  |                                    | still failing
                  |                                    v
                  |                              orders.DLQ
                  |                                    |
                  +----- operator replays after fix ---+

  Every message in the DLQ carries headers: original topic, partition,
  offset, the exception class, the stack trace, the attempt count, and
  the timestamp of the first failure.

Three rules make a dead letter queue useful rather than a place where data goes to be forgotten. Attach enough context to diagnose without the original logs, which means the original coordinates, the error, and the attempt history. Alert on the arrival rate rather than on the depth, because a non-empty queue is normal and a rising rate is not. And build the replay tool at the same time as the queue, because a dead letter queue with no replay path is a data loss mechanism with extra steps.

Poison pill records deserve a specific mention. A record that deterministically crashes the consumer will crash it again on restart, and a consumer that crashes before committing never advances. The result is an infinite crash loop on one record. Catch deserialization errors separately from processing errors and route them straight to the dead letter topic, because a record that cannot be deserialized will never succeed.

Sources and further reading

Jay Kreps, "The Log: What every software engineer should know about real-time data's unifying abstraction", 2013. Martin Kleppmann, "Making Sense of Stream Processing". The Kafka improvement proposals KIP-98 for the idempotent producer and transactions, KIP-429 for cooperative rebalancing, KIP-345 for static membership, KIP-405 for tiered storage, and KIP-1150 for diskless topics. Greg Young's talks on event sourcing and on the distinction between event sourcing and CQRS. The Confluent Schema Registry documentation on compatibility modes. The Debezium documentation on incremental snapshots and replication slot management.