Event Sourcing and CQRS with Kafka

Event sourcing represents application state as a history of domain events. CQRS separates the model that accepts commands from models used to answer queries. Kafka can connect those models, but it does not automatically supply concurrency checks, aggregate indexes or permanent history.

The first decision is whether Kafka is the authoritative event store or the transport for changes committed elsewhere. That choice determines where commands are validated, which transaction accepts them, and how recovery reconstructs state.

Scope: Kafka 4.0.x behavior; application patterns are illustrative. Configuration fragments do not implement an event-store library or prove transactional behavior in an application framework.

An event backbone and an event store have different obligations

For an event backbone, a database can validate a command, commit the mutation and outbox row together, then publish the change through Kafka. Consumers build search indexes, caches and analytics views. Kafka carries the change; the database remains authoritative for command acceptance.

For an authoritative event store, accepting a command means durably appending sufficient domain history. The design must also answer:

• How is an append rejected when the aggregate version is stale?
• How are all events for one aggregate located and reconstructed?
• Which history can expire, and where can it be recovered afterward?
• How are writer ownership, schema changes and deletion obligations enforced?

Kafka provides ordered partition appends and replication. It does not provide an expected-version compare-and-append operation for each key or an indexed query returning that key’s history. A consumer-maintained store/index or another event store can supply those features, but they are additional architecture.

Kleppmann’s discussion of logs and derived data explains why an ordered change stream can maintain several views. It does not eliminate the need to define the source of truth and command boundary.

Assigning a version does not enforce concurrency

Suppose two command handlers load account version 7. Both validate a withdrawal against the same balance, assign version 8 and publish an event. Kafka can order both appends, but that order does not retroactively reject a command validated against stale state.

A conditional database update or expected-version append can accept one transition from 7 to 8 and reject the other. Alternatively, route commands through a single authoritative processing owner for the aggregate, with durable state and fencing during takeover. In either design, describe how ownership survives failover and how accepted commands become durable effects.

Do not launch concurrent transactions on a shared producer and call that optimistic concurrency control. The producer supports one open transaction; assigning expectedVersion + 1 locally is merely numbering.

These options have different operational costs; choose one rather than implying Kafka partition order implements all three.

Worked decision: when the order service needs a retained stream

At the first stage, the service accepts 200 order events/s, has one notification worker and needs retryable delivery. Its database already owns order validation. A database transaction plus outbox, relayed to an existing task queue, is a defensible starting design: the team gets durable command acceptance and its required work distribution without taking on a retained-stream platform solely for one handler. The relay and notification operation still need duplicate-safe behavior.

Now change the requirement. Search, fulfillment and analytics each need their own progress, and each must replay 72 hours without asking the command service to republish history. The database remains authoritative, but a retained Kafka stream becomes a stronger candidate. An order ID is a candidate partition key when the required order is per order; a merchant-wide invariant would need a different ownership design.

The bet has a cost: someone must own connector/relay lag, schemas, retained history, broker or provider capacity, and projection recovery. The task-queue alternative needs an explicit mechanism for independent historical replay; simply adding more workers to one queue does not supply that contract. Conversely, Kafka does not remove destination-specific retry and poison-event design.

Record what would change the decision: if independent replay disappears and one handler remains, the simpler queue path may suffice. If historical queries, conditional appends or legal retention dominate, compare an event store/archive design rather than assuming longer Kafka retention supplies every capability. Validate the chosen replay path with the oldest required data before migrating readers.

Preserve transitions separately from snapshots

A domain event such as MoneyWithdrawn(20) is a transition. Keeping only the last transition does not recover an account balance. A snapshot records state at a known version and can replace earlier snapshots if the necessary tail remains available.

# Event-history topic: use a deliberate retention/archive contract.
cleanup.policy=delete
retention.ms=31536000000

The example retains roughly a year under the timestamp/segment rules described in retention. It cannot support an arbitrary all-time audit after older history is deleted without another recoverable archive.

# Separate snapshot topic keyed by aggregate ID.
cleanup.policy=compact

A snapshot needs aggregate identity, schema version, aggregate version and a reliable replay boundary. For a partition-based materialization, that boundary may be a vector of input offsets. Verify that snapshot persistence and progress agree; a snapshot without its exact boundary can skip or repeat required state transitions.

Compaction is asynchronous and supplies no point lookup. A materialized snapshot index must be built and recovered. If a snapshot predates retained history, the missing interval makes reconstruction impossible; finding a snapshot row is not enough.

Partition keys are part of the contract

An aggregate key keeps its events on one partition under stable routing. Partition expansion, custom partitioners or time-based routing can move future events elsewhere and break that assumption. Plan a cutover or merge protocol before changing the mapping.

A tenantId:aggregateId key distinguishes identities; hashing it does not provide tenant isolation, locality or independent scaling. Authorization usually applies to resources such as topics, and quotas/resource placement require separate controls. A large tenant or hot aggregate can still dominate shared capacity.

Log order also differs from event-time order. For reconstruction, use the application’s accepted version/order, not a casual sort by wall-clock timestamps. Clock skew and delayed publication can make timestamp order differ from the accepted command sequence.

Projections must tolerate replay

A projection consumes events and updates a query model. If its database update succeeds and offset commit fails, the event can be delivered again. A uniqueness-guarded event ID plus the mutation in one database transaction is one approach; atomic state-and-offset storage is another.

Version-aware updates need domain-specific rules. Ignoring an older full-state replacement may be safe. Ignoring an out-of-order delta can lose a required contribution. A gap in aggregate versions may require waiting, fetching missing history or rebuilding, not simply accepting the largest number.

Retries can reorder projection effects if failed events move to another topic while later events proceed. Define whether the view must preserve per-aggregate order and which failures may be quarantined. A saga adds compensating business actions; compensation is not a database rollback and itself needs durable identity and retries.

CQRS also exposes freshness to users. After command acceptance, the read model may lag. Decide whether the client waits for a particular aggregate version, reads an authoritative path, or displays an explicit pending state. “Eventually consistent” is not a response-time or freshness promise.

Rebuild without treating an empty poll as EOF

Rebuilding a projection is a migration with a bounded target, not just a consumer started at offset zero.

1. Create a separate destination version. Keep the serving projection intact.
2. Identify required partitions and capture target end offsets using the intended isolation semantics. Record the boundary, schema version and rebuild identity.
3. Restore an appropriate snapshot or seek to the required available start positions. Confirm retention covers the remaining tail.
4. Consume through each captured target. An empty poll() does not establish completion; compare progress with the target positions.
5. Validate counts, invariants and selected entities at an equivalent boundary. Comparing a historical rebuild with a continuously moving live table can produce misleading differences.
6. Catch up the new projection to the live tail under a controlled cutover policy, then switch readers. Keep the previous version for rollback until the new one is trusted.

A captured vector of partition offsets is a replay boundary, not necessarily a global transactional snapshot. If cross-partition business invariants require a consistent cut, the application must define one. For transactional inputs, unresolved transactions can constrain the visible boundary.

Rate-limit rebuild reads and destination writes so they do not consume the live pipeline’s recovery margin. Cold history may miss broker page cache and alter live latency. Disable or isolate irreversible effects such as emails during replay; rebuilding a query table must not resend the year’s notifications.

Schema evolution is a recovery concern

A consumer that understands today’s events may still fail while replaying last year’s history. Preserve event meaning and identity, test old schemas, and distinguish additive defaults from semantic transformations. Upcasting can adapt representation; it cannot invent facts absent from an old event.

Snapshots need migration/versioning too. A changed projection can require rebuilding from events instead of reading an incompatible snapshot. Retention and archive policy therefore constrain which future schema migrations remain possible.

Before selecting Kafka as the event store, rehearse two concurrent commands for the same aggregate, a crash after projection mutation, a missing snapshot tail and a full rebuild while live traffic continues. If command authority or replay completeness remains unclear, use Kafka as an event backbone and keep those responsibilities in a store that explicitly implements them.

Continue with stateful stream processing for joins and time, or pipeline sizing for replay budgets.

Whiteboard: choose who may accept an order

Guided variant. Two command handlers read aggregate version 7 and both try to accept a conflicting change at version 8. The chosen design uses a conditional database mutation and an outbox row in one transaction. Complete the two outcomes and mark when an event is allowed to reach Kafka. Would merely publishing both events with a version-8 field reject the stale command?

Independent problem. Recommend either DB/outbox→task queue or DB/outbox→Kafka for the three-subscriber, 72-hour replay stage. Put six items on the board: acceptance boundary; ordering key; duplicate-effect rule; replay-completeness argument; operating owner/cost; and evidence that would make you reverse the choice. Then introduce a merchant-wide credit limit shared by many orders. Explain whether keeping order-ID partitioning alone enforces it.

Pass criterion: defend the choice using requirements and failure boundaries, then identify exactly which part changes when the business invariant spans several keys. Product names without an acceptance and recovery argument do not pass.