“Stream processing” sounds like a for loop over a queue: read a message, do a thing, write a result. That mental model dies the moment the thing you do has memory. Count events per user per minute, and now you have state - which user is at what count, and what happens to that count when the machine holding it dies mid-minute? Read late-arriving events, and now “per minute” is a lie, because messages do not arrive in the order they happened. Ask for exactly-once, and now a crash between “read”, “update the count”, and “write the result” cannot be allowed to double-count or drop, across three different systems at once. A durable log that just moves bytes is the easy half. The hard half is running stateful computation over that log at a million messages a second per topic without losing, duplicating, or mis-ordering a single result when a node inevitably falls over.

The core problem is that a stream processor has to hold three things together that fight each other: a durable ordered log as the transport, large mutable per-key state as the computation, and exactly-once correctness spanning both plus the output - all while a node can die at any instant. The log part is Kafka-shaped: partitioned, replicated, append-only. The processing part is where it gets interesting - local state that must survive failure, windows defined on when events happened not when they arrived, and a commit protocol that makes read-process-write-state atomic. This post builds the log properly and then builds the processing engine on top of it, because a stream processing platform is both. Let me do it end to end.

Functional Requirements (FR)

In scope:

  • Ingest from many producers into topics. Thousands of producers publish messages (key, value, event timestamp) to named topics. The system durably stores each message and acknowledges the write. This is the ingest path.
  • Durable, ordered, partitioned log. Each topic is split into partitions; within a partition, messages have a strict total order and a monotonic offset. The log is retained for a window regardless of consumption, so processing can replay.
  • Consumer groups over the log. Independent consumer groups each read the full stream at their own pace; within a group, partitions are split across members for parallel consumption. This is standard log delivery.
  • Stateful stream processing. Run operators over the streams - map, filter, groupBy, aggregate, join stream-to-stream and stream-to-table - where aggregations and joins hold large per-key state that must survive failures.
  • Event-time windowing. Group events into tumbling, hopping, and session windows keyed on the event timestamp (when it happened), tolerating out-of-order and late arrivals, not on when the processor happened to see the event.
  • Exactly-once processing. A message affects the output and the state exactly once, even across crashes and restarts - no double-counting, no lost updates, spanning the input offset, the state store, and the output topic together.
  • Scale to 1M messages/sec per topic with horizontal scale-out of both the log (brokers, partitions) and the processing (tasks, workers).

Explicitly out of scope (name it so you own the scope):

  • A SQL query planner / full SQL surface. A streaming SQL layer (ksqlDB / Flink SQL) compiles down to the operator graph we build; we design the runtime, not the parser and optimizer.
  • Cross-datacenter geo-replication. Mirroring topics across regions is a real feature but a separate design; we build a single-cluster platform and note the extension.
  • The connector ecosystem. Source and sink connectors (ingest from a database CDC feed, write to a warehouse) sit at the edges; we treat them as producers and consumers and design the core.
  • Schema registry and serialization formats. Enforcing payload schemas is an adjacent service; we treat values as opaque bytes and note where the registry hooks in.
  • Exactly-once into arbitrary external sinks. We guarantee exactly-once within the platform (log + state + output topic). A downstream non-transactional database requires that sink to cooperate (idempotent writes), which we describe but do not own.

The decision that drives everything: a stream processor is a deterministic operator graph running over a partitioned, replayable log, whose per-key state is itself backed by a log, and whose exactly-once guarantee comes from atomically committing input offsets, state changes, and outputs together. Log for transport, log for state, one commit for correctness - every hard part below is a consequence of that.

Non-Functional Requirements (NFR)

  • Scale: a hot topic at 1,000,000 messages/sec sustained (bursts higher), with the platform hosting hundreds of topics and tens of thousands of partitions across a broker fleet of tens of nodes, and a processing fleet of tens to hundreds of worker tasks. Reads exceed writes because each message is consumed by multiple groups and re-read on reprocessing.
  • Latency: end-to-end (produce to processed output) in the tens to low hundreds of milliseconds when processors are caught up. This is a low-latency continuous system, not a batch job, though it batches internally for throughput. Produce-ack p99 in single-digit to tens of milliseconds under the durable setting.
  • Availability: 99.9%+ on ingest and processing. A broker or worker failure must not lose accepted messages and must not stall a partition for more than the few seconds it takes to fail over and (for processing) restore state.
  • Consistency: within a partition, a strict total order and a single source of truth (the partition leader). Processing results are consistent with exactly-once semantics against that ordered input; reads see only committed messages.
  • Durability: very high. Once a produce is acked under acks=all with replication factor 3 and min.insync.replicas=2, the message survives any single broker loss with zero data loss. Processor state is durable too - it is backed by a replicated changelog, so a worker crash never loses committed state.
  • Correctness: exactly-once is the headline requirement, and it is not free. It spans three things at once (input progress, state, output), so it needs an atomic commit protocol, not just idempotent retries.

The tension to state up front: throughput and low latency want to process in memory and ack fast; durability and exactly-once want to persist and coordinate, which costs latency. We resolve it by keeping the hot path in memory and local (local state, batched I/O, sequential log appends) and pushing durability off to asynchronously replicated logs (the message log and the state changelog), then reconciling them with a periodic atomic checkpoint rather than a synchronous write per message. Correctness is bought in bulk, at checkpoint boundaries, not paid per event.

Back-of-the-Envelope Estimation (BoE)

Real numbers with the arithmetic, because they set the partition count, the state budget, and the checkpoint interval.

Ingest throughput and bandwidth (the hot topic):

Target:               1,000,000 messages/sec on one topic at peak
Average message size: ~1 KB (key + value + headers, packed)
Write bandwidth:      1M/sec * 1 KB = 1 GB/sec inbound for this topic alone

Fan-out on read: consumed by ~3 groups + re-read on reprocessing (~1x) => ~4x
Read bandwidth:  1 GB/sec * 4 = ~4 GB/sec outbound for this topic

Reads dominate writes because the same log is read by every processing job and every reprocess. Size the network and page cache for the fan-out, not just the ingest.

Partition count (the parallelism unit):

Per-partition sustainable throughput ~ 20K-50K msgs/sec (append + replicate + fetch).
To hit 1M/sec on one topic:  1M / ~30K = ~33 partitions minimum.
Add headroom for parallel consumers and future growth -> pick ~60-100 partitions.
Parallelism ceiling: one partition -> one task per operator, so 100 partitions
   lets a processing job run up to 100 parallel tasks. Partition count is a
   first-class capacity decision, and it is hard to change later (see bottlenecks).

Log storage:

Write volume:        1 GB/sec = 86,400 sec * 1 GB ≈ 86 TB/day (this topic).
Retention:           3 days replay window -> 86 * 3 = ~260 TB (one copy).
Replication factor 3: 260 TB * 3 = ~780 TB of raw disk for one hot topic.
Across all topics: multiple PB. Sequential append/scan, page-cache friendly,
   so it is disk-heavy but disk-efficient - not random-access.

Processing state (the part the log post does not have):

Say the job counts events per user over a 1-minute window.
Active keys (distinct users in flight): ~50M
State per key: key (~16B) + count/window aggregate (~64B) + overhead ~ 100-150B
   50M * ~128B = ~6.4 GB of live state for one operator.
A stream-stream join buffering both sides for a window can be far larger:
   1M/sec * 60s window * 1 KB = ~60 GB buffered per join side.
=> State is GB to tens of GB per operator, sharded across tasks, held locally
   on the workers (RocksDB-style), NOT in a remote database on the hot path.

Changelog and checkpoint volume:

State updates get logged to a compacted changelog topic (for recovery).
If 30% of the 1M/sec messages mutate state: 300K state-updates/sec * ~128B
   = ~38 MB/sec of changelog per operator - modest next to the 1 GB/sec input.
Checkpoint interval: ~every 10-60s. Longer = cheaper but more replay on recovery;
   shorter = fast recovery but more overhead. 10s is a common middle.

The headline shape: ingest is ~1 GB/sec of sequential log, reads are a multiple of that from fan-out, storage is petabytes of sequential log, and the genuinely new scarce resource versus a plain queue is per-key processing state - GB to tens of GB, held locally, that must be made durable without a synchronous write per event. Size for sequential I/O, fan-out, partition count, and local state.

High-Level Design (HLD)

Two planes stacked. The log plane is a cluster of brokers storing topic partitions as replicated append-only logs (the Kafka-shaped transport). The processing plane is a fleet of worker nodes that run the operator graph as parallel tasks (one task per partition per operator), each holding local state and reading from and writing to the log. A coordination layer (a Raft-based controller for the log, a job manager for the processing) tracks leadership, task assignment, and checkpoints.

   PRODUCERS                          PROCESSING PLANE (operator graph as tasks)        SINKS / CONSUMERS
  ┌─────────┐                        ┌──────────────────────────────────────────┐      ┌──────────────┐
  │producers│  produce(T,key,val,    │  Job Manager (checkpoint coordinator,      │      │ output topic │
  │(events) │◄─ ─ ─ event_ts) ─ ─ ─ ▶│  task assignment, failure detection)       │      │ consumers /  │
  └────┬────┘        │               │                                            │      │ external sink│
       │             │               │   ┌ task p0 ┐ ┌ task p1 ┐ ┌ task p2 ┐      │      └──────▲───────┘
       ▼             │               │   │ source  │ │ source  │ │ source  │      │             │
 ┌───────────────────┼── LOG PLANE (BROKER CLUSTER) ──────────────┼──────────────┼─────────────┘
 │                   │   │            │   │ mapFilter│ │ mapFilter│ │ mapFilter│   │
 │  ┌ Broker A ┐  ┌ Broker B ┐  ┌ Broker C ┐         │   │ groupBy │ │ groupBy │ │ groupBy │
 │  │T-p0 LEAD │  │T-p0 foll │  │T-p0 foll │         │   │  +STATE │ │  +STATE │ │  +STATE │  each task:
 │  │append log│─▶│replicate │─▶│replicate │◀ fetch ─┘   │ (local  │ │ (local  │ │ (local  │  - reads its
 │  │T-p1 foll │  │T-p1 LEAD │  │T-p2 LEAD │             │ RocksDB)│ │ RocksDB)│ │ RocksDB)│    partition
 │  └────┬─────┘  └──────────┘  └──────────┘            └────┬────┘ └────┬────┘ └────┬────┘  - updates state
 │       │ leader appends, ISR acks -> committed             │           │          │        - emits output
 │       │                                                   ▼           ▼          ▼        - checkpoints
 │  ┌──────────────── CHANGELOG TOPICS (compacted) ──────────────────────────────────────┐  │
 │  │  state backup per task: key -> latest value, replicated like any topic              │  │
 │  └─────────────────────────────────────────────────────────────────────────────────────┘ │
 │  ┌──────────────── CONTROLLER (Raft quorum) ──────────────────────────────────────────┐  │
 │  │  partition leadership + ISR, broker liveness, offset topic, task assignment metadata │ │
 │  └─────────────────────────────────────────────────────────────────────────────────────┘ │
 └────────────────────────────────────────────────────────────────────────────────────────────┘

Ingest flow (a producer writing an event):

  1. A producer publishes an event with a key, a value, and an event timestamp (when it actually happened - this matters for windowing later). It asks the controller which broker leads T-partition p, where p = hash(key) % numPartitions, so all events for one key stay on one partition and stay ordered. It caches this.
  2. The producer sends the message (batched with others for throughput) to the leader broker for T-p. The leader appends it to the tail of the partition’s on-disk commit log and assigns the next sequential offset.
  3. The leader replicates the record to follower brokers; once the in-sync replicas (ISR) have it, it is committed. Under acks=all the leader acks the producer only now, so an acknowledged message already survives a broker loss.

Processing flow (a stateful operator over the log):

  1. A processing job is compiled into an operator graph (source -> map/filter -> groupBy+aggregate -> sink) and deployed as tasks: one task per input partition per stateful stage. Task p0 owns partition 0 end to end.
  2. The source operator in each task fetches from its partition at its current offset. It runs the record through the stateless operators (map, filter) in memory.
  3. At a stateful operator (aggregate, join), the task reads and writes local state (a RocksDB-style embedded store on the worker’s disk), keyed by the record’s key. Because the key hashes to this partition, all records for a key land in this one task, so the state for that key is local - no cross-node lookup on the hot path.
  4. The operator emits output records to a downstream operator or, at the sink, produces them to an output topic. State mutations are also appended to a changelog topic (compacted) so the state can be rebuilt after a crash.
  5. Periodically the job manager triggers a checkpoint: every task atomically records “I have consumed input through offset N, my state is at version V, my output through this point is committed.” This checkpoint is what makes recovery exactly-once (deep dive 4).

Recovery flow: if a worker dies, the job manager reassigns its tasks to another worker, which restores state from the latest checkpoint plus the changelog tail, rewinds the input to the checkpointed offset, and resumes. Because processing from that offset is deterministic and the checkpoint bundles offset + state + output atomically, nothing is lost or double-counted.

The structural insight: the log plane gives durable ordered replayable input; the processing plane keeps state local for speed and backs it with a changelog for durability; and a periodic atomic checkpoint over offset + state + output is the single mechanism that turns an in-memory, at-least-once pipeline into an exactly-once one.

Component Deep Dive

Four hard parts, naive-first then evolved: (1) the durable partitioned log itself - storage and replication, the stated hard part; (2) large per-key processing state that survives failure; (3) event-time windowing with out-of-order data; (4) exactly-once across consume-process-write-state.

1. The log: mutable queue -> replicated append-only partitioned commit log

We need to store 1M messages/sec durably, ordered per key, replayable, and survive a broker death. This is the transport the whole platform sits on.

Naive approach: an in-memory queue per topic, delete on consume

Keep a list per topic, producers append to the tail, consumers pop from the head, the broker deletes a message once acknowledged and tracks per-message delivery state.

enqueue(msg) -> list.append(msg)
dequeue()    -> msg = list.popHead(); on ack -> delete msg

Where it breaks:

  • No replay. A stream processor must reprocess history (bug fix, new job, state rebuild). Delete-on-consume leaves nothing to replay from.
  • No independent fan-out. Every job and every reprocess needs its own full pass over the stream. A pop-once queue would need N copies or N-way per-message cursors, plus per-message ack bookkeeping the broker has to mutate randomly.
  • Random-access I/O and no durability. Per-message deletes and delivery-state updates are random writes with index churn and locking - you cannot sustain 1M/sec of that. In memory, a crash loses everything unacked.
  • No fault tolerance. One copy on one machine dies with the machine.

The mutable-queue model conflates storage with consumption tracking and cannot replay - fatal for a processing platform.

Evolved approach: partitioned append-only commit log, replicated leader-follower

Store each partition as an immutable append-only log of records, each with a monotonic offset, in segment files on disk. Nothing is deleted on consume; retention is time/size based. Split the topic into N partitions across brokers for horizontal scale, routed by hash(key) % N. Replicate each partition R times (typically 3) with one leader and followers.

partition T-p0 on disk (segment files, rolled by size/time):
   00000000000000000000.log   offsets 0        .. 4,999,999
   00000000000005000000.log   offsets 5,000,000.. (active, appended)
   each segment has a sparse .index (offset -> byte pos) for O(log) seek.

append(record): write to tail of active segment; offset = last + 1   (sequential)
read(offset):   binary-search sparse index -> byte pos -> sequential scan

replication (RF 3), acks=all:
   leader appends -> ISR followers fetch and ack -> record COMMITTED -> ack producer
   consumers read only COMMITTED records (never a message a failover could erase)

Why this is right:

  • Sequential I/O gives GB/sec. Appends go to the segment tail - pure sequential writes, no index rewrite per message. Reads are sequential scans. This is how one broker sustains ~1 GB/sec, and how the platform hits 1M/sec by spreading partitions across brokers.
  • Zero-copy reads. A fetch is served by sendfile()-ing bytes straight from page cache to socket. A caught-up processor reads from RAM, disk untouched.
  • Replay and fan-out are free. Rewinding is seek(offset); many jobs read the same immutable log at different offsets. This is exactly what stream reprocessing and multi-job platforms need.
  • Durability across broker death. Only records replicated to the whole ISR are committed. With acks=all and min.insync.replicas=2, an acked message is on >=2 brokers; on a leader death the controller elects a new leader from the ISR in seconds with zero committed-message loss. unclean.leader.election stays off - prefer consistency (no data loss) over availability for a durable log.
  • Partitioning is the parallelism unit. N partitions give N times the throughput and up to N parallel processing tasks. Ordering is per-partition (per-key), not topic-global - which is the ordering a keyed aggregation actually needs. The partition key is the most consequential and least reversible decision (changing N remaps keys), so pick it deliberately.

This log is the foundation. The processing plane treats it as a durable, ordered, replayable tape - and, crucially, uses the same log machinery to make its own state durable (next).

2. Processing state: remote database per event -> local state store backed by a changelog

Aggregations and joins hold large per-key state (counts, running sums, join buffers). At 1M/sec that state is read and written on nearly every message, and it must survive a worker crash. Where does it live?

Naive approach: keep state in an external database, read-modify-write per event

For each event, SELECT the current aggregate from Postgres/Redis, update it, UPDATE it back.

on_event(e):
    cur = db.get(e.key)              # network round trip
    cur = aggregate(cur, e)
    db.put(e.key, cur)               # network round trip
    emit(cur)

Where it breaks:

  • Two network round trips per event. At 1M/sec that is 2M DB ops/sec against a shared database - it becomes the bottleneck and dwarfs the processing cost. A ~1ms round trip caps you at ~1K/sec per thread; you would need thousands of threads fighting one database.
  • Read-modify-write races. Concurrent tasks updating the same key need locks or transactions in the DB, serializing the hot path and adding latency.
  • Coupling and blast radius. The processing job’s throughput and availability are now hostage to an external database’s. A DB hiccup stalls the whole stream.
  • No clean recovery boundary. The DB state and the input offset advance independently, so a crash leaves them inconsistent - you cannot tell which events were already applied.

Remote state per event makes the network the bottleneck and destroys the locality the partitioning gave you.

Evolved approach: local embedded state store, made durable by a compacted changelog topic

Because records are partitioned by key, all events for a key are routed to one task. So put that key’s state in the task: an embedded key-value store (RocksDB-style, on the worker’s local disk with an in-memory cache) that the operator reads and writes with no network hop. Make it durable by writing every state change to a compacted changelog topic in the log plane - one changelog partition per task.

task p3 owns input partition 3 (all keys where hash(key)%N == 3):
   local state store (embedded KV, memtable + SSTs on local disk):
      key -> aggregate/window/join-buffer         # O(1) local read/write, no network
   changelog topic  __job-state-p3  (compacted):
      on state mutation: produce (key -> new value) to changelog   # async, batched
      compaction keeps only the latest value per key -> changelog ~ size of state

on_event(e):
    cur = local.get(e.key)          # local, microseconds
    cur = aggregate(cur, e)
    local.put(e.key, cur)           # local
    changelog.produce(e.key, cur)   # durable backup, batched, off the hot path
    emit(cur)

recovery of task p3 on a new worker:
    replay __job-state-p3 (compacted) -> rebuild local store    # bounded by state size
    resume input partition 3 from the checkpointed offset

Why this is right:

  • Local reads and writes. No network per event; state ops are memory/local-disk speed, so a single task keeps up with tens of thousands of events/sec and the job scales by adding tasks (partitions), not by hammering one database.
  • Locality follows partitioning. The same hash(key) % N that shards the log also shards the state, so state is always where the events are. No cross-task coordination for single-key aggregation.
  • Durability via the log, again. The changelog is just another replicated topic, so state inherits the log plane’s durability and replication. Compaction keeps only the latest value per key, so the changelog stays roughly the size of the live state, not the size of the update history.
  • Bounded, deterministic recovery. To restore a lost task, replay its compacted changelog into a fresh local store on another worker - time proportional to state size, not to the whole input history. (For big state, also keep periodic local snapshots so you replay only the changelog tail.)
  • Standby replicas for fast failover. Optionally keep warm standby tasks that continuously consume the changelog, so on failure a standby with near-current state takes over without a full restore.

Local state backed by a changelog is the move that makes stateful processing fast (local) and durable (log) at once - the same “log is truth, working copy is a replay of it” idea as the message log, applied to state.

3. Windowing and time: process-time windows -> event-time windows with watermarks

Most stream jobs group events into windows - “count per user per minute.” The subtle, interview-critical part is which minute: when the event happened (event time) or when the processor saw it (processing time). Messages arrive out of order and late, so these differ.

Naive approach: tumbling windows on the processor’s wall clock

Bucket each event into a window by System.now() at the moment of processing; close and emit a window when the clock passes its end.

on_event(e):
    w = floor(now() / 60s)          # processing-time bucket
    windows[w][e.key] += 1
    if now() >= w.end: emit(windows[w]); drop windows[w]

Where it breaks:

  • Wrong answers on out-of-order data. An event that happened at 12:00:59 but arrives at 12:01:02 (network delay, a producer catching up after a blip) is counted in the 12:01 window, not 12:00. The per-minute counts are wrong, and they change depending on ingestion lag.
  • Non-reproducible. Replaying the same log later, with different timing, buckets events differently - so a reprocess for a bug fix produces different numbers than the original run. For a system whose recovery story is “replay the log,” that is fatal.
  • No principled handling of lateness. There is no notion of “this event is late” versus “on time”; you either silently misplace it or drop it with no rule.

Processing-time windows are easy and almost always wrong for analytics, because they measure the pipeline’s mood, not the data’s reality.

Evolved approach: event-time windows driven by watermarks

Window by the event timestamp carried in each record (stamped by the producer at the source). Track progress in event time with a watermark: a monotonically advancing assertion that “no more events with timestamp <= W are expected.” A window closes and emits when the watermark passes its end, not when the wall clock does.

each record carries event_ts (set at the source, immutable through the pipeline).
watermark W = max_event_ts_seen - allowed_lateness   (per partition, then merged)

on_event(e):
    w = window_for(e.event_ts)          # bucket by EVENT time
    state[w][e.key] = aggregate(state[w][e.key], e)   # in the local state store
on_watermark(W):
    for each window w where w.end <= W:
        emit(state[w]); retain briefly for late updates, then evict     # fire the window

late event (event_ts < W - allowed_lateness):
    route to a side output / dead-letter, or drop by policy (explicit, not silent).

Why this is right:

  • Correct, reproducible results. Counts reflect when events happened, so out-of-order arrival does not corrupt them, and a replay of the same log yields the same windows every time - which is exactly what the exactly-once recovery story needs.
  • Watermarks make “done” decidable. Without them you never know when a window is complete (more late data could always arrive). The watermark is the explicit, tunable bound: allowed_lateness trades completeness (wait longer, catch more stragglers) against latency (emit sooner). State that.
  • Windows live in the state store. Open windows are just keyed entries (window, key) -> aggregate in the local RocksDB-style store from deep dive 2, so they are durable and recoverable like any other state, and closed windows are evicted to bound memory.
  • Session and hopping windows fall out. Session windows merge events within a gap; hopping windows overlap. All are the same “assign event_ts to one or more window keys, aggregate in state, fire on watermark” machinery - only the window-assignment function changes.
  • Late data is a policy, not an accident. Events past the allowed lateness go to a side output for reconciliation or are dropped by an explicit rule, never silently misbucketed.

Event time plus watermarks is what makes a stream processor produce correct aggregates over messy, delayed, out-of-order reality - and produce the same correct aggregates on every replay.

4. Exactly-once: at-least-once and hope -> atomic checkpoint of offset + state + output

The whole thing runs read -> process -> update state -> write output. A crash can happen between any of these. Exactly-once means each input message affects state and output exactly once, spanning all three. This is the stated headline requirement and the hardest part.

Naive approach: at-least-once, commit the input offset after writing output

Process the event, write the output, then commit the consumer offset. On crash, reprocess from the last committed offset.

on_event(e):
    out = process(e)
    output_topic.produce(out)       # (1)
    local_state.put(e.key, update)  # (2)
    consumer.commit(e.offset)       # (3) after (1) and (2)

Where it breaks:

  • Crash between (1)/(2) and (3) double-applies. If the worker dies after producing output and mutating state but before committing the offset, recovery re-reads the event, produces the output again, and applies the state update again. The count is wrong; the downstream sees a duplicate. At-least-once double-counts on every failure.
  • Crash between (1) and (2) tears state and output apart. Output written, state not - or vice versa. Now the emitted result and the internal state disagree, and no offset marks the boundary.
  • Three independent commits, no atomicity. The output topic, the state store, and the offset advance are three separate systems committing independently. There is no single point that says “all three are exactly here,” so recovery cannot restore a consistent line.

At-least-once plus separate commits cannot be exactly-once, because the failure window straddles three uncoordinated writes.

Evolved approach: periodic aligned checkpoint that atomically bundles offset, state, and output

Make the three commit together, atomically, at periodic checkpoint boundaries, so recovery always rolls back to a consistent line where input offset, state version, and emitted output all agree. Two complementary mechanisms, used together:

(a) Aligned checkpoint barriers (the snapshot). The job manager periodically injects a numbered checkpoint barrier into the source streams. Each operator, on receiving the barrier on all its inputs, snapshots its state (offset for sources, state store for aggregations) tagged with that checkpoint id, then forwards the barrier. When every operator has snapshotted, the checkpoint is complete - a globally consistent cut across the whole graph: “at checkpoint C, every task had consumed exactly up to these offsets and its state was exactly this.”

Job Manager: inject barrier(C) into all source partitions
Source task: on barrier(C): snapshot {input_offset} as of C, forward barrier
Stateful op: on barrier(C) on ALL inputs (align): snapshot local state store
             (upload SSTs / changelog position) as of C, forward barrier
Sink op:     on barrier(C): flush + prepare output as of C
Job Manager: when all tasks ack C -> checkpoint C is COMPLETE and durable

Recovery: restore every task's state to the last COMPLETE checkpoint C,
          rewind every source to C's offsets, resume.
          => every message between C and the crash is reprocessed from a
             consistent line: state and offset move back together, no tear.

(b) Transactional output (the two-phase commit to the log). Reprocessing after recovery would still re-emit output unless the output is tied to the checkpoint. So the sink writes output inside a transaction on the output topic, and the transaction commits only when the checkpoint completes. Consumers of the output read with read_committed, so they never see output from an uncommitted (later rolled-back) attempt.

between checkpoints C and C+1:
   sink.produce(out) inside transaction Tx(C+1)     # each stamped (producer_id, seq)
on checkpoint C+1 complete:
   commit Tx(C+1)                                   # output becomes visible atomically
on failure before C+1:
   abort Tx(C+1) (or it is never committed) -> its output is discarded,
   state + offsets roll back to C -> reprocessing re-emits under a new Tx. No dup.

idempotent producer underneath: (producer_id, per-partition sequence) lets the
broker drop duplicate produces from retries, so even producer-level retries are safe.

Why this is right:

  • Offset, state, and output move as one unit. Recovery restores to checkpoint C for all three simultaneously, so there is no torn state, no double-applied update, and no visible duplicate output. That is exactly-once across the whole read-process-write-state loop.
  • Correctness is bought in bulk, not per event. The expensive coordination (barrier alignment, transaction commit) happens every checkpoint interval (say 10s), not per message. The hot path stays at-least-once-cheap in memory; the checkpoint makes the observable result exactly-once. This is how you get exactly-once at 1M/sec without a synchronous distributed commit per event.
  • The changelog and snapshots make state recovery bounded. A restored task rebuilds its state from the last checkpoint’s snapshot plus the changelog tail (deep dive 2), so recovery is proportional to state-since-checkpoint, not the whole history.
  • read_committed hides in-flight work. Downstream consumers only ever see output from completed checkpoints, so a failed-and-retried interval leaves no duplicate visible - the aborted transaction’s records are skipped.
  • The honest caveat. Exactly-once holds within the platform (input log + state + output topic). If the sink is an external non-transactional system (a REST call, a plain database), you fall back to at-least-once delivery + an idempotent sink (dedup by a message key or a (checkpoint_id, key) idempotency token on the write). Say it plainly: end-to-end exactly-once requires either a transactional sink or an idempotent one; the platform guarantees its own half and hands the sink a stable dedup key.

Exactly-once is not a delivery flag you flip - it is a periodic atomic checkpoint that snapshots offsets and state on a consistent cut and commits output transactionally on that same cut, so a crash always rolls the entire pipeline back to a line where everything agrees.

API Design & Data Schema

Ingest / admin API (write and control plane)

POST /topics                          # admin: create a topic
  { "name": "events", "partitions": 96, "replication_factor": 3,
    "retention_ms": 259200000,        # 3 days
    "min_insync_replicas": 2, "cleanup_policy": "delete" }
  201 Created

PRODUCE (to the partition leader, batched under the hood)
  send(topic="events", key="user_8f2a", value=<bytes>, event_ts=1752451200123,
       headers={...}, acks="all")
  routing: partition = key ? hash(key) % partitions : round_robin
  -> { partition: 41, offset: 880453120, timestamp: 1752451200123 }
     (returned only after commit to the ISR under acks=all)

Job / processing API (define and run the operator graph)

POST /jobs                            # submit a processing job (operator graph)
  { "name": "per-user-minute-count",
    "source": { "topic": "events" },
    "graph": [
      { "op": "filter",  "expr": "type == 'click'" },
      { "op": "keyBy",   "key": "user_id" },
      { "op": "window",  "type": "tumbling", "size_ms": 60000,
                         "time": "event", "allowed_lateness_ms": 10000 },
      { "op": "aggregate","fn": "count" }
    ],
    "sink": { "topic": "user_click_counts", "semantics": "exactly_once" },
    "parallelism": 96, "checkpoint_interval_ms": 10000 }
  -> { "job_id": "job-771", "state": "RUNNING", "tasks": 96 }

GET  /jobs/{job_id}         -> status, per-task offsets, consumer lag, last checkpoint
POST /jobs/{job_id}/rescale -> { "parallelism": 128 }   # repartition state, rebalance tasks
POST /jobs/{job_id}/restart -> { "from_checkpoint": "C-5591" }  # replay from a checkpoint

Consumer API over output (read plane)

SUBSCRIBE  subscribe(group_id="dashboard", topics=["user_click_counts"])
FETCH      poll(max_bytes=1MB, max_wait_ms=100, isolation="read_committed")
           -> batch: [ { partition, offset, key, value, event_ts, headers }, ... ]
COMMIT     commit(offsets={ user_click_counts-3: 480200 })
SEEK       seek(partition, offset|timestamp|beginning|end)   # replay / reset

Data model: log segments, local state, changelog, checkpoints, metadata

The heart is not a table. It is the on-disk log segment, the embedded state store, and small metadata stores for coordination. SQL vs NoSQL barely applies to the hot path - the message store is deliberately a purpose-built append-only log, and the state store is an embedded LSM-tree, because a general SQL database’s random-access, index-per-row, mutable model is the opposite of a GB/sec sequential-append, local-state workload.

On-disk partition log (source of truth, per partition per broker):

dir: /data/events-41/
   00000000000000000000.log       # length-prefixed records, append-only
   00000000000000000000.index     # sparse offset -> byte position
   00000000000000000000.timeindex # sparse event_ts -> offset (seek by time)
record: { offset(implicit), length, crc, event_ts, key_len, key,
          value_len, value, headers, producer_id, producer_epoch, sequence }

Local state store (per task, embedded LSM-tree, on worker disk):

state store for task p41 (RocksDB-style):
   key: (window_start, user_id)   value: { count, ... }   # windowed aggregate
   plus in-memory memtable cache; periodic local snapshots for fast restore.
   NOT a remote DB - local, so state ops are microseconds.

Changelog topic (compacted, backs the state store):

topic __job-771-state-p41   cleanup_policy = compact
   key: (window_start, user_id)   value: latest aggregate
   compaction keeps only the latest value per key -> size ~ live state size.

Checkpoint metadata (managed by the job manager, stored durably):

checkpoint C:
   { checkpoint_id, per_task: { input_offsets, state_snapshot_ref,
     changelog_positions }, output_txn_id, status: PENDING|COMPLETE }

Cluster metadata (Raft-based controller quorum):

partition_state: (topic,partition) -> { leader, replica_set[], isr[], leader_epoch }
broker_registry: broker_id -> { host, port, rack, last_heartbeat, status }
task_assignment: job_id -> { task -> worker, standbys[] }

Why this storage split:

StoreHoldsWhy this choice
Local-disk log segmentsthe messagessequential append/scan, zero-copy reads, cheap segment-drop retention
Embedded LSM state storeper-key processing statelocal (no network per event), fast writes, snapshot-able
Compacted changelog topicdurable backup of statereuses log durability/replication; compaction keeps size ~ state
Raft controller quorumleadership, ISR, assignmentsmall, strongly consistent, always available for failover
Checkpoint metadataconsistent offset+state+output cutthe anchor that makes recovery exactly-once

Bottlenecks & Scaling

Where it breaks first, in order, and the fix for each.

1. A topic outgrows one broker (throughput and storage). One log on one machine caps at that machine’s NIC and disk, well below 1M/sec. Fix: partitioning - split the topic across brokers for N times the throughput and storage and up to N parallel tasks. Size partition count generously up front (it is hard to change; hash%N remaps keys), and spread leadership evenly so no broker leads all the hot partitions.

2. Hot key / hot partition (skew). A celebrity user_id hashes a disproportionate share onto one partition, hotspotting one broker and one processing task. Fix: higher-cardinality or salted keys (user_id + bucket) with a downstream merge so the hot entity spreads across sub-partitions; pre-aggregate hot keys locally before the shuffle (combiner) to cut the volume that hits the single task; detect skew from per-partition lag metrics.

3. Processing state too large for one worker / lost on crash. Tens of GB of per-key state cannot sit in one JVM heap, and a worker death would lose it. Fix: embedded LSM state store on local disk (spills past RAM), compacted changelog for durability, periodic snapshots + standby replicas so failover restores fast without replaying the whole changelog. Rescale by repartitioning state across more tasks.

4. Exactly-once coordination overhead. Per-event distributed commit would be far too slow at 1M/sec. Fix: periodic aligned checkpoints + transactional output (deep dive 4) - buy correctness in bulk at 10s boundaries, keep the per-event hot path in memory at-least-once-cheap. Tune the checkpoint interval: longer = less overhead but more reprocessing on recovery.

5. Consumer/processor lag and backpressure. A slow operator (heavy state, a slow sink) makes the job fall behind the tail; naive buffering blows up memory. Fix: credit-based backpressure propagated upstream so a slow sink slows the source (bounded buffers, never unbounded), plus more tasks/partitions and autoscaling on lag (tail offset minus committed offset) as the primary health signal.

6. Broker failure loses/stalls a partition. A single-copy partition dies with its broker. Fix: RF 3 with ISR, acks=all + min.insync.replicas=2; the controller elects a new leader from the ISR in seconds with no committed-message loss. Unclean leader election off (consistency over availability).

7. Rebalance / checkpoint storms. Frequent task membership changes or overlapping checkpoints stall processing and thrash state restoration. Fix: static membership + cooperative sticky assignment so only moved tasks pause; incremental checkpoints (upload only changed SSTs) so checkpointing large state is cheap; tuned session timeouts so a GC pause is not read as death.

8. Controller / job manager as a single point of failure. If the thing tracking leadership, assignment, and checkpoints dies, nothing can fail over. Fix: run the controller as a replicated Raft quorum (3 or 5 nodes); run the job manager with a standby that recovers from the durable checkpoint metadata. Keep their state small so they stay fast and available.

9. Storage growth (petabytes of log). Retention at ~1 GB/sec per topic fills disks fast. Fix: time/size retention (drop aged-out segments by unlinking whole files), compaction for changelog/table topics (latest per key), and tiered storage - hot tail on local disk, older segments offloaded to object storage, still replayable.

10. Read fan-out saturates leaders. Many jobs and reprocesses reading the same hot partitions concentrate egress on leader brokers. Fix: rely on page cache + zero-copy so hot reads never hit disk; enable follower fetching (read from a same-zone in-sync follower) to spread read load and cut cross-zone bandwidth.

Shard keys, stated plainly: the log, the processing tasks, the local state, and the changelog all shard by the message key via hash(key) % numPartitions - because the key is simultaneously the unit of ordering (per-key order within a partition), the unit of parallelism (one task per partition), and the unit of state locality (a key’s state lives in the one task that owns its partition). That single alignment is what lets stateful processing be both parallel and local.

Wrap-Up

The trade-offs that define this design:

  • Log for transport, log for state. Messages live in a partitioned append-only commit log, and per-key processing state lives in a local store backed by a compacted changelog - the same “immutable log is truth, working copy is a replay of it” idea applied twice. We trade a simple database for sequential-I/O throughput, replay, and local state locality.
  • Local state over remote state. Because the key that partitions the log also partitions the state, state sits in the task that owns the key - no network per event. We trade a shared database’s convenience for the throughput that makes 1M/sec viable, and back it with a changelog for durability.
  • Event time over processing time. Windows are keyed on when events happened, advanced by watermarks, so results are correct on out-of-order data and identical on every replay. We trade some latency (allowed_lateness) for correctness and reproducibility.
  • Exactly-once is built, not free. It is a periodic aligned checkpoint that atomically snapshots input offsets and state on a consistent cut, plus transactional output committed on that same cut. We buy correctness in bulk at checkpoint boundaries rather than per event, and fall back to at-least-once + an idempotent sink for non-transactional external systems.
  • Correctness over raw availability at the seams. acks=all, ISR-only reads, and no unclean leader election prefer no-data-loss over staying up; a torn checkpoint rolls back rather than serving inconsistent results.

One-line summary: a durable, ordered, replicated, partitioned commit log ingesting 1M messages/sec per topic, over which a deterministic operator graph runs as one task per partition holding local per-key state backed by a compacted changelog, windowing on event time with watermarks, and achieving exactly-once by periodically checkpointing input offsets, state, and transactional output together on a consistent cut so any crash rolls the whole pipeline back to a line where everything agrees.