# System Design: [Stream processing engine name]

> Reference design for a distributed stream processing engine that runs
> long-lived, stateful jobs over unbounded event streams. It covers event time
> versus processing time, windows, watermarks, triggers and late data, keyed
> state with periodic checkpoints, exactly-once state with transactional sinks,
> backpressure and rescaling. The concepts follow the Dataflow model (Akidau
> et al., PVLDB vol. 8 no. 12, 2015) and asynchronous barrier snapshots
> (Carbone et al., 2015); the text and the worked design are original. It
> follows the structure of the System Design Document template. Replace every
> `[bracketed]` value and every row marked *Example*.

<!--
Guidance: How to use this template
- Decide first what "correct" means for your results: complete per event-time
  window, or fresh within a latency budget. That choice drives watermarks,
  triggers and the sink design (sections 7.1 to 7.4).
- Fill the capacity inputs in 3.1 from real source metrics; state size and the
  checkpoint interval, not CPU, usually decide the cluster shape.
- Keep every deep dive's options visible even when the choice looks obvious;
  reviewers and interviewers will ask about the alternatives.
- Run a failure drill (kill a worker at peak) before launch and record the
  measured recovery time in section 9.
-->

## Document control

<!-- Guidance: A design is a proposal until it is approved; record who decided and when. -->

| Field | Value |
|---|---|
| Document ID | [SDD-SPE-001] |
| System / feature | [Stream processing engine for real-time aggregation] |
| Version | [0.1] |
| Status | [Draft / In review / Approved / Implemented / Superseded] |
| Owner | [@owner] |
| Author(s) | [@author] |
| Reviewers | [@reviewer-1], [@reviewer-2] |
| Last updated | YYYY-MM-DD |
| Classification | [Internal] |
| Delivery guarantee | [Exactly-once state; exactly-once or idempotent output per sink] |
| Related | Message Queue System Design (`system-design/design-message-queue`), CDC Pipeline System Design (`system-design/design-cdc-pipeline`), Data Pipeline Specification (`ai-data/data-pipeline-spec`) |

### Approvals

| Role | Name | Decision | Date |
|---|---|---|---|
| Tech lead | [@name] | [Approved / Changes requested] | YYYY-MM-DD |
| Staff / principal engineer | [@name] | [Decision] | YYYY-MM-DD |
| SRE / operations | [@name] | [Decision] | YYYY-MM-DD |
| Data platform owner | [@name] | [Decision] | YYYY-MM-DD |
| Security | [@name] | [Decision] | YYYY-MM-DD |

## Revision history

<!-- Guidance: Note what changed in the design, not the wording. -->

| Version | Date | Author | Change summary |
|---|---|---|---|
| 0.1 | YYYY-MM-DD | [@name] | First draft for design review |
| 0.2 | YYYY-MM-DD | [@name] | *Example:* Moved from processing-time to event-time windows after the replay incident |
| 0.3 | YYYY-MM-DD | [@name] | *Example:* Added unaligned checkpoints for jobs under sustained backpressure |
| 1.0 | YYYY-MM-DD | [@name] | *Example:* Approved |

## Contents

<!-- Guidance: Keep aligned with the headings. -->

1. [Overview](#1-overview)
2. [Requirements](#2-requirements)
3. [Capacity estimation](#3-capacity-estimation)
4. [API design](#4-api-design)
5. [Data model](#5-data-model)
6. [High-level design](#6-high-level-design)
7. [Detailed design](#7-detailed-design)
8. [Scalability and performance](#8-scalability-and-performance)
9. [Reliability, security and operations](#9-reliability-security-and-operations)
10. [Trade-offs and alternatives](#10-trade-offs-and-alternatives)
11. [Delivery plan and risks](#11-delivery-plan-and-risks)
12. [Appendices](#12-appendices)

## 1. Overview

<!-- Guidance: One page that a busy reviewer can read alone: problem, proposed solution in a paragraph, and the decisions that need their attention. -->

### 1.1 Problem statement

[*Example:* Product teams compute campaign metrics (impressions, clicks,
spend per campaign and country per minute) with a nightly batch job and a
fragile set of consumers that count in memory. Dashboards are 12 hours stale,
counts change after every consumer restart, and mobile events that arrive
late (devices offline, then reconnecting) are either dropped or counted in the
wrong minute. Advertisers dispute numbers that do not match the batch
reports.]

### 1.2 Proposed solution

[*Example:* A shared stream processing engine runs declarative jobs that read
partitioned event logs, assign each event to event-time windows, keep
per-key state in an embedded log-structured store on local SSD, and take a
consistent distributed snapshot every 30 s by injecting barriers into the
streams. Watermarks track event-time progress per source partition; triggers
emit early speculative results every 10 s, an on-time result when the
watermark passes the window end, and corrections for late events for up to
10 minutes. Sinks either commit transactionally on checkpoint completion or
write idempotent upserts keyed by window. Jobs rescale by restoring a
savepoint with a different parallelism; state moves in key groups.]

### 1.3 Decisions requested from reviewers

| ID | Question | Recommended option | Section |
|---|---|---|---|
| Q-01 | *Example:* Event time or processing time for business metrics? | Event time with heuristic watermarks | 7.1 |
| Q-02 | *Example:* How long do we accept late events? | 10 minutes allowed lateness; later events to a side output | 7.2 |
| Q-03 | *Example:* Checkpoint mechanism? | Aligned barrier snapshots, unaligned when backpressured | 7.3 |
| Q-04 | *Example:* Exactly-once to sinks? | Two-phase commit for log sinks, idempotent upsert for databases | 7.4 |
| Q-05 | *Example:* Checkpoint interval? | 30 s, bounded by freshness for transactional readers | 3.4, 7.4 |
| [Q-nn] | [Question] | [Option] | [Section] |

### 1.4 45-minute interview path

<!-- Guidance: Use this order when presenting the design in an interview or a short review; the same document then serves both. -->

| Minutes | Cover | Output on the whiteboard |
|---|---|---|
| 0 to 5 | Clarify the use case: what is aggregated, freshness target, how late events can be, correctness expectation | Core requirements FR-01 to FR-05, NFR-01 to NFR-04 |
| 5 to 10 | Estimate ingest, state size and recovery time | Parallelism 64, ~25 GB state, recovery of a few minutes |
| 10 to 20 | Draw sources, coordinator, workers, state backend, checkpoint storage, sinks | Diagram in section 6 |
| 20 to 35 | Deep dive: pick event time and watermarks (7.1) plus checkpoints and exactly-once (7.3, 7.4); if the interviewer leans on correctness of late data, use 7.2 instead | Barrier and two-phase commit sequence |
| 35 to 42 | Backpressure, rescaling, hot keys | Key groups, credit-based flow control |
| 42 to 45 | Trade-offs and what changes at 10x | Table in section 10 |

## 2. Requirements

<!-- Guidance: Split core requirements from below-the-line items so nobody designs for what is out of scope. Non-functional requirements need numbers and a measurement. -->

### 2.1 Core functional requirements

| ID | Requirement | Priority |
|---|---|---|
| FR-01 | *Example:* Read events from partitioned logs and write results to logs, key-value stores and an analytical store | Must |
| FR-02 | *Example:* Group events by key and assign them to tumbling, sliding and session windows by event time | Must |
| FR-03 | *Example:* Emit early, on-time and late results per window as configured by a trigger | Must |
| FR-04 | *Example:* Keep per-key state across restarts with no loss and no double counting | Must |
| FR-05 | *Example:* Stop a job with a savepoint and restart a new version or parallelism from it | Must |
| FR-06 | *Example:* Route events later than the allowed lateness to a side output for audit | Should |
| FR-07 | *Example:* Join two streams by key within an event-time interval | Should |
| [FR-nn] | [Requirement] | [Must / Should / Could] |

### 2.2 Non-functional requirements

| ID | Quality | Target | Measured as |
|---|---|---|---|
| NFR-01 | Throughput | *Example:* 2 million events/s per job at peak | Source lag stays flat during load test |
| NFR-02 | Freshness | *Example:* p99 of 60 s from event arrival to visible on-time result | End-to-end marker events |
| NFR-03 | Correctness | *Example:* Final window values equal a batch recomputation over the same input | Daily reconciliation job |
| NFR-04 | Recovery | *Example:* Output fresh again within 5 minutes after a worker loss at peak | Chaos drill |
| NFR-05 | Isolation | *Example:* One job's backpressure does not slow another job | Multi-tenant load test |
| [NFR-nn] | [Quality] | [Target] | [Measurement] |

### 2.3 Below the line (out of scope)

- [*Example:* A SQL interface for ad-hoc queries; jobs are defined in code or the pipeline format in 4.2 for this release]
- [*Example:* Ingestion from devices; the engine reads from the existing message logs (`system-design/design-message-queue`)]
- [*Example:* Machine-learning feature serving; results land in stores that a feature service reads]
- [*Example:* Exactly-once delivery to external HTTP endpoints that offer no idempotency key]
- [Item deferred, with the release that will take it]

## 3. Capacity estimation

<!-- Guidance: Show inputs, formulas and results, and end every estimate with the design consequence it forces. Round generously and say which way you rounded. -->

### 3.1 Assumptions

| Input | Symbol | Value | Source |
|---|---|---|---|
| Peak events per second (one large job) | $R$ | *Example:* 2 000 000 | Source log metrics x growth |
| Average event size | $e$ | *Example:* 1 KB | Sampled payloads |
| Events per second one stateful task handles | $c$ | *Example:* 40 000 | Benchmark of the window operator |
| Active keys (campaign x country) | $K$ | *Example:* 5 million | Last 30 days |
| Window size and allowed lateness | $w$, $L$ | *Example:* 1 min, 10 min | Product requirement |
| Window state per key and window | $s_w$ | *Example:* 100 B | Measured |
| Deduplication horizon and ID size | $h$, $s_d$ | *Example:* 600 s, 16 B | Source retry behaviour |
| Checkpoint interval | $T_c$ | *Example:* 30 s | 7.3 |

### 3.2 Ingest and parallelism

$$
B = R \times e = 2 \times 10^{6} \times 1\ \text{KB} = 2\ \text{GB/s}
$$

$$
P = \frac{R}{c} \times 1.25 = \frac{2 \times 10^{6}}{4 \times 10^{4}} \times 1.25 = 62.5 \Rightarrow 64\ \text{tasks}
$$

Consequence: the keyed shuffle moves roughly the same 2 GB/s (16 Gbit/s
aggregate) across the network on top of ingest, so worker network interfaces
(*Example:* 25 Gbit/s) are sized for ingest, shuffle and checkpoint upload
together, and the job is split into at least 64 parallel tasks from day one. The maximum parallelism
(number of key groups) is fixed at 1 024 so the job can grow 16 times without
rehashing state.

### 3.3 State size

Open windows per key are one current window plus the lateness horizon:

$$
S_{win} = K \times \left(1 + \frac{L}{w}\right) \times s_w = 5 \times 10^{6} \times 11 \times 100\ \text{B} = 5.5\ \text{GB}
$$

$$
S_{dedup} = R \times h \times s_d = 2 \times 10^{6} \times 600 \times 16\ \text{B} = 19.2\ \text{GB}
$$

Total keyed state is about 25 GB, or about 0.39 GB per task at parallelism 64.
Consequence: deduplication, not the windows, dominates state, so it gets its
own TTL and is the first thing to shrink (shorter horizon or idempotent
producers upstream). State lives in an embedded log-structured store on local
SSD rather than on the heap, which keeps garbage collection pauses away from
the watermark path and enables incremental checkpoints.

### 3.4 Checkpoints and recovery

With *Example* 20% of state changing per interval, each incremental
checkpoint uploads 5 GB, or about 167 MB/s sustained to object storage.
Consequence: checkpoint upload is a steady background load that must be
budgeted in network capacity, not an occasional burst.

After a worker loss the job rewinds to the last checkpoint. Worst-case backlog
is one interval plus detection and restore time:

$$
\text{backlog} = R \times (T_c + t_{detect} + t_{restore}) = 2 \times 10^{6} \times (30 + 10 + 4)\ \text{s} = 88 \times 10^{6}\ \text{events}
$$

Spare capacity is $64 \times 40\,000 - 2 \times 10^{6} = 560\,000$ events/s,
so catching up takes about $88 \times 10^{6} / 5.6 \times 10^{5} \approx 157$ s.
Consequence: recovery to fresh output is under 3 minutes at peak, which meets
NFR-04 but not a 60 s freshness promise during incidents; raising headroom to
1.5x or shortening the interval are the two levers.

### 3.5 Output

Each of the 5 million keys emits one on-time row per minute, about 83 000
rows/s, plus early firings every 10 s (up to 500 000 rows/s if every key is
active). Consequence: sinks must accept batched upserts, and early firings are
enabled only for keys a dashboard actually reads.

### 3.6 Summary

| Quantity | Estimate | Design consequence |
|---|---|---|
| Ingest | 2 GB/s (16 Gbit/s aggregate) | Worker NICs sized for ingest + shuffle + checkpoints; shuffle sized like ingest |
| Parallelism | 64 tasks, 1 024 key groups | Rescale up to 16x without rehashing |
| Keyed state | ~25 GB (dedup 19.2 GB) | Embedded LSM store on SSD; TTL on dedup state |
| Checkpoint upload | ~167 MB/s | Incremental checkpoints; network budget |
| Recovery | ~3 min to fresh at peak | Meets NFR-04; more headroom if freshness tightens |
| Output | 83 000 to 500 000 rows/s | Batched upserts; early firings opt-in |

## 4. API design

<!-- Guidance: Define the contracts users touch: job submission, pipeline definition, lifecycle operations and the event envelope. Use RFC 9457 problem details for errors. -->

### 4.1 Control plane API

| Method and path | Purpose | Notes |
|---|---|---|
| `POST /v1/jobs` | Submit a job definition and artifact | Idempotency key header; returns job ID |
| `GET /v1/jobs/{id}` | Status, parallelism, last checkpoint, lag | |
| `POST /v1/jobs/{id}/savepoints` | Take a savepoint, optionally stop the job | Returns savepoint URI |
| `POST /v1/jobs/{id}/rescale` | Restart from a savepoint with new parallelism | Bounded by max parallelism |
| `POST /v1/jobs/{id}/upgrade` | Deploy a new version from a savepoint | State compatibility checked first |
| `DELETE /v1/jobs/{id}` | Cancel a job | Savepoint taken unless `force=true` |

### 4.2 Pipeline definition

A job declares what is computed, where in event time, when results are
emitted and how refinements relate (the four questions of the Dataflow model,
paraphrased):

```yaml
job: campaign-metrics
version: 14
source:
  log: ad-events
  timestamp_field: event_time
  watermark:
    strategy: bounded_out_of_orderness
    max_delay: 30s
    idle_partition_timeout: 60s
key_by: [campaign_id, country]
window:
  type: tumbling
  size: 1m
  allowed_lateness: 10m
trigger:
  early: every 10s processing_time
  on_time: at_watermark
  late: every_element
accumulation: accumulating   # each firing carries the full value so far
aggregate: {impressions: count, clicks: count, spend_micros: sum}
dedup: {field: event_id, horizon: 10m}
late_side_output: ad-events-too-late
sink:
  type: upsert_store
  table: campaign_minute
  key: [campaign_id, country, window_start]
checkpoint: {interval: 30s, mode: aligned, unaligned_after: 60s}
```

### 4.3 Output record

Every emitted result states which firing it is, so consumers can tell a
speculative value from a final one:

```json
{"campaign_id": "c-7781", "country": "VN", "window_start": "2026-10-22T09:41:00Z", "window_end": "2026-10-22T09:42:00Z", "impressions": 18240, "clicks": 311, "spend_micros": 9120000, "pane": {"timing": "LATE", "index": 3, "is_last": false}, "watermark": "2026-10-22T09:42:31Z"}
```

### 4.4 Errors

| Status | Problem type | When |
|---|---|---|
| 409 | `state-incompatible` | Upgrade changes a stateful operator's ID or state schema without migration |
| 409 | `rescale-out-of-bounds` | Requested parallelism exceeds max parallelism |
| 422 | `invalid-pipeline` | Window, trigger or watermark settings contradict each other |
| 503 | `checkpoint-in-progress` | Savepoint requested while the previous one is still running |

## 5. Data model

<!-- Guidance: Choose storage by access pattern. Show entities, keys, partitioning and the access-pattern table: each query, the key or index that serves it, and why the partition key was chosen. -->

### 5.1 Entities

```mermaid
erDiagram
  JOB ||--o{ JOB_VERSION : has
  JOB_VERSION ||--o{ OPERATOR : contains
  OPERATOR ||--o{ TASK : "runs as"
  TASK ||--o{ KEY_GROUP : owns
  JOB ||--o{ CHECKPOINT : takes
  CHECKPOINT ||--o{ STATE_HANDLE : references
  JOB {
    string job_id PK
    string owner
    int max_parallelism
  }
  JOB_VERSION {
    string job_id PK
    int version PK
    string artifact_uri
    string pipeline_hash
  }
  OPERATOR {
    string operator_uid PK
    string kind
    string state_schema
  }
  TASK {
    string task_id PK
    int subtask_index
    string worker_id
  }
  KEY_GROUP {
    int key_group PK
    string task_id FK
  }
  CHECKPOINT {
    string job_id PK
    bigint checkpoint_id PK
    string status
    datetime completed_at
    string kind
  }
  STATE_HANDLE {
    string handle_id PK
    string operator_uid FK
    int key_group_start
    int key_group_end
    string object_uri
  }
```

A key maps to a key group by hash modulo max parallelism; key groups map to
tasks in contiguous ranges. Rescaling moves whole key groups, never single
keys, which keeps state handles coarse.

### 5.2 Storage choices

| Data | Store | Key | Why |
|---|---|---|---|
| Job and version metadata | *Example:* Relational database | `job_id`, `version` | Small, audited, transactional |
| Leader and checkpoint pointer | Coordination service (`system-design/design-distributed-lock-coordination`) | `jobs/{id}/leader`, `jobs/{id}/latest-checkpoint` | Leader election and an atomic pointer swap |
| Live keyed state | Embedded LSM store on local SSD per task | `(key_group, key, namespace)` | Fast local reads and writes; large state |
| Checkpoint files | Object storage | `/{job}/{checkpoint_id}/{operator}/{handle}` | Durable, cheap, shared files across incremental checkpoints |
| Input and output streams | Partitioned log | Partition by key | Replayable from offsets |

### 5.3 Access patterns

| Access pattern | Served by | Partition key and rationale |
|---|---|---|
| Update window state for one event | Local LSM, key `(key_group, campaign_id, country, window_start)` | Key group from the business key, so all events for a key reach one task |
| Fire all windows ending before the watermark | Event-time timer queue per task, ordered by timestamp | Timers stored with the key group, so they move with state on rescale |
| Check duplicate `event_id` | Local LSM, key `(key_group, campaign_id, country, event_id)` with TTL 10 min | Key group from the business key, so deduplication runs in the same task as the window; it inherits that key's skew (7.5) |
| Find the latest completed checkpoint | Coordination service pointer | One small value per job, updated atomically |
| Restore state for key groups 0 to 15 | Checkpoint metadata lists handles per key-group range | Range layout lets a task fetch only its ranges |
| Read results for a dashboard | Sink table keyed `(campaign_id, country, window_start)` | Upsert target; idempotent by construction |

## 6. High-level design

<!-- Guidance: One diagram with every component on the data path and the control path, and where state lives. Name components by responsibility. -->

```mermaid
flowchart LR
  subgraph sources["Partitioned logs"]
    s1[(ad-events p0..p127)]
  end
  subgraph cp["Control plane"]
    api[Job API] --> coord[Job coordinator]
    coord --> ckc[Checkpoint coordinator]
    coord <--> cs[(Coordination service)]
  end
  subgraph workers["Workers"]
    src[Source tasks + watermarks] -->|keyed shuffle| win[Window + dedup tasks]
    win --> snk[Sink tasks]
    win --- lsm[(Local LSM state)]
  end
  s1 --> src
  ckc -->|inject barriers| src
  lsm -->|incremental upload| obj[(Object storage)]
  snk --> outlog[(Output log, transactional)]
  snk --> kv[(Upsert store)]
  win -->|too late| side[(Late side output)]
```

| Component | Responsibility | Scales by |
|---|---|---|
| Job coordinator | Schedules tasks on workers, restarts on failure, one active per job | Leader election; one per job |
| Checkpoint coordinator | Triggers checkpoints, collects acknowledgements, publishes the completed pointer | Part of the coordinator |
| Source tasks | Read partitions, extract timestamps, generate watermarks | Source partitions |
| Window and dedup tasks | Keyed state, timers, triggers | Key groups |
| Sink tasks | Buffer output, pre-commit and commit per checkpoint | Output partitions |
| Object storage | Checkpoint and savepoint files | Managed service |

## 7. Detailed design

<!-- Guidance: Go deep on two to four hard problems. For each, list the options, the choice and the reason, and show the failure behaviour. -->

### 7.1 Deep dive: time, watermarks and progress

Processing time is when a worker sees an event; event time is when it
happened. Mobile and retried events arrive minutes late, and a replay after an
outage delivers hours of events in minutes, so grouping by processing time
puts events in the wrong window and makes results depend on when the job ran.

| Option | How it works | Strength | Weakness |
|---|---|---|---|
| A. Processing-time windows | Window by arrival clock | Simple; no lateness logic | Wrong under replay and delay; not reproducible |
| B. Event time with perfect watermarks | Source guarantees order (for example a single ordered log) | Exact completeness | Rarely available for real sources |
| C. Event time with heuristic watermarks | Watermark = max event time seen minus a bound, per source partition | Works on real data; tunable | Some events arrive behind the watermark (late) |
| D. Event time, no watermarks, fire on processing-time triggers only | Periodic emissions of running totals | Fresh | Never knows when a window is complete |

**Choice:** C, with per-partition watermarks combined by taking the minimum
across inputs at every operator. **Reason:** it produces reproducible results
under replay and a clear point at which a window is considered complete,
while triggers (7.2) cover freshness before that point and corrections after.

Rules that keep watermarks honest:

- Generate watermarks per source partition, not per task, so a slow partition
  holds back the watermark instead of producing late data.
- Mark a partition idle after `idle_partition_timeout` so an empty partition
  does not stall all windows; un-idle it on the next event.
- Never let the watermark move backwards; on restore, start from the
  checkpointed watermark.
- Track watermark lag (processing time minus watermark) as the main
  freshness signal; it is the number a dashboard should show.

### 7.2 Deep dive: windows, triggers and late data

Windows decide where in event time results are computed; triggers decide when
in processing time results are emitted; accumulation decides how successive
results for the same window relate.

| Window type | Use | State cost |
|---|---|---|
| Tumbling (fixed) | Per-minute metrics | One open window per key plus lateness |
| Sliding (hopping) | Moving averages, for example 10 min every 1 min | One copy per overlapping window, or panes combined at firing |
| Session | User activity separated by a gap | Windows merge as events arrive; state per active session |
| Global with custom trigger | Count-based or per-key batching | Unbounded unless a trigger clears it |

| Accumulation mode | Each firing carries | Consumer must |
|---|---|---|
| Discarding | Only new data since last firing | Sum firings itself |
| Accumulating | Full value so far | Overwrite (upsert by window key) |
| Accumulating and retracting | New full value plus a retraction of the previous one | Apply retraction, then the new value (needed when results are regrouped downstream) |

**Options for late data:** drop silently; keep windows open forever; keep
windows for an allowed lateness then send later events to a side output.
**Choice:** allowed lateness of 10 minutes with accumulating firings and a
side output. **Reason:** state cost is bounded (3.3), dashboards stay correct
through ordinary device delays, and nothing is dropped silently: the side
output feeds the daily reconciliation that compares stream totals with the
batch recomputation (NFR-03).

```mermaid
stateDiagram-v2
  [*] --> Open: first event for key and window
  Open --> Open: event, early firing every 10 s
  Open --> Fired: watermark passes window end, on-time firing
  Fired --> Fired: late event, late firing
  Fired --> Purged: watermark passes end + allowed lateness
  Purged --> [*]
  note right of Purged
    Later events go to the side output
  end note
```

### 7.3 Deep dive: keyed state and checkpoints

| Option | How it works | Strength | Weakness |
|---|---|---|---|
| A. Write every state change to an external database | State lives remotely | Simple recovery | One remote write per event; 2 million writes/s |
| B. Periodic stop-the-world snapshot | Pause all tasks, snapshot, resume | Simple consistency | Pauses the whole job; long at 25 GB |
| C. Aligned barrier snapshots | Coordinator injects numbered barriers at sources; an operator snapshots when barriers from all inputs arrived | Consistent without global pause; small checkpoints with incremental upload | Alignment waits under backpressure |
| D. Unaligned barrier snapshots | Barriers overtake buffered records; in-flight buffers are stored in the checkpoint | Checkpoint time independent of backpressure | Larger checkpoints; more restore work |
| E. Changelog per state update | Every update appended to a changelog log | Fast recovery with standby replicas | Doubles write traffic |

**Choice:** C by default, switching a checkpoint to D when alignment exceeds
60 s. **Reason:** aligned barriers give a consistent cut of all operator state
and source offsets with no global pause and small checkpoints; the fallback
keeps checkpoints completing under sustained backpressure, which is exactly
when a recovery point matters most.

```mermaid
sequenceDiagram
  participant CK as Checkpoint coordinator
  participant S as Source task
  participant W as Window task
  participant K as Sink task
  participant O as Object storage
  CK->>S: trigger checkpoint 812
  S->>S: record source offsets
  S->>W: barrier 812 (after records before it)
  W->>W: wait for barrier 812 on all inputs
  W->>O: upload new LSM files (async)
  W->>K: barrier 812
  K->>K: pre-commit output transaction 812
  S-->>CK: ack 812 (offsets)
  W-->>CK: ack 812 (state handles)
  K-->>CK: ack 812 (pre-committed txn id)
  CK->>CK: write metadata, swap latest pointer
  CK->>K: checkpoint 812 complete
  K->>K: commit transaction 812, open 813
```

State snapshots are asynchronous: the task flushes its memtable, records the
list of immutable files and continues processing while files upload. Files
already uploaded by earlier checkpoints are referenced, not copied. Retain the
last 3 completed checkpoints and delete unreferenced files with a reference
count, never by age alone.

### 7.4 Deep dive: exactly-once state and output

Checkpoints make state exactly-once: after a failure, state and source
offsets rewind together, so each event affects state once. Output is a
separate problem because records emitted after the last checkpoint are emitted
again on replay.

| Option | How it works | Fits | Cost |
|---|---|---|---|
| A. At-least-once output | Emit immediately; duplicates on replay | Metrics consumers that tolerate duplicates | Consumers deduplicate |
| B. Idempotent upsert | Output key is deterministic (`campaign_id`, `country`, `window_start`) and the value is a full accumulating result | Key-value and analytical stores | Needs accumulating mode and deterministic keys |
| C. Two-phase commit sink | Write into a transaction per checkpoint, pre-commit at the barrier, commit when the checkpoint completes | Transactional logs read with read-committed isolation | Output visible only after commit, so latency is at least the checkpoint interval |
| D. Write-ahead then publish | Buffer output in state, publish after checkpoint | Sinks without transactions | Duplicates possible if publish fails mid-way |

**Choice:** B for databases and C for the output log. **Reason:** B gives
exactly-once effect with no extra latency because replays overwrite the same
rows with the same values; C is the only way to give downstream stream
consumers exactly-once records, and the 30 s interval fits the 60 s freshness
target (NFR-02). Record the pre-committed transaction IDs in the checkpoint so
a restarted sink can commit or abort them; set the transaction timeout well
above the longest expected recovery, or a commit after recovery fails and data
is lost.

### 7.5 Backpressure, hot keys and rescaling

- **Backpressure:** use credit-based flow control between tasks: a receiver
  grants buffer credits and a sender sends only with credit, so a slow sink
  slows its upstream tasks instead of growing unbounded queues. Expose the
  share of time each task is backpressured and the source lag.
- **Hot keys:** a campaign that gets 10% of all traffic overloads one task.
  Pre-aggregate before the shuffle (each source task combines events per key
  for a short interval), then merge partial results after it. For non-mergeable
  logic, salt the key and add a second merge stage.
- **Rescaling:** take a savepoint, restart with the new parallelism, assign
  key-group ranges to tasks and restore only those ranges. Source partitions
  are reassigned; the source offsets in the savepoint keep reads exact.
- **Upgrades:** give every stateful operator a stable ID; refuse an upgrade
  whose operator IDs or state schemas cannot be matched (error 409 in 4.4).

### 7.6 How real systems do it

> **How real systems do it (paraphrased, not copied).** Akidau and colleagues
> at Google described the Dataflow model in "The Dataflow Model: A Practical
> Approach to Balancing Correctness, Latency, and Cost in Massive-Scale,
> Unbounded, Out-of-Order Data Processing" (PVLDB vol. 8 no. 12, 2015). They
> argue that pipelines should stop assuming data will ever be complete and
> should instead let authors choose explicitly among correctness, latency and
> cost. The model separates what is computed (transformations), where in event
> time (windowing, including merging session windows), when results are
> materialised (watermarks and triggers) and how later results relate to
> earlier ones (discarding, accumulating, or accumulating with retractions).
> Separately, Carbone and colleagues described lightweight asynchronous
> barrier snapshots for dataflows (arXiv 1506.08603, 2015), the basis of the
> aligned checkpoint approach in 7.3. Read both papers for the reasoning; this
> template only reuses the concepts.

## 8. Scalability and performance

<!-- Guidance: Show how the design meets the numbers in section 3 and where it breaks first. -->

| Concern | Approach | Limit before redesign |
|---|---|---|
| Ingest | Source parallelism equals partitions | *Example:* 128 partitions per topic; add partitions with a planned rescale |
| Shuffle | Pre-aggregation before keyed exchange | Network per worker |
| State growth | TTL on dedup and session state; LSM compaction tuning | *Example:* about 50 GB per task before restore time breaks NFR-04 |
| Checkpoint duration | Incremental, asynchronous, unaligned fallback | Upload bandwidth to object storage |
| Many small jobs | Shared session cluster for small jobs, dedicated cluster for large | Coordinator memory per job |
| Hot keys | Two-stage aggregation | One key above one task's capacity after combining |

Latency budget for an on-time result (NFR-02, *Example*):

| Step | Budget |
|---|---|
| Producer to log | 2 s |
| Watermark delay (bounded out-of-orderness) | 30 s |
| Processing and shuffle | 2 s |
| Commit wait for transactional sinks (half an interval on average, up to 30 s) | 15 s typical |
| Headroom | 11 s |

## 9. Reliability, security and operations

<!-- Guidance: Name the failure modes, the threats and the signals operators watch. Each mitigation should be testable, and each failure needs a detection signal and a recovery time objective. -->

### 9.1 Failure-mode table

| Component | Failure | Detection | Blast radius | Mitigation | RTO |
|---|---|---|---|---|---|
| Worker | Process or host lost | Heartbeat timeout 10 s | Tasks of the jobs on that worker; whole job rewinds | Restart from last checkpoint; spare slots reserved | *Example:* under 3 min to fresh output |
| Job coordinator | Leader lost | Coordination-service lease expiry | Scheduling and checkpoints of one job | Standby takes leadership, reads latest pointer | *Example:* under 1 min |
| Object storage | Slow or unavailable | Checkpoint duration and failures | New checkpoints fail; processing continues | Tolerate N failed checkpoints, then alert; never delete the last good one | Depends on provider |
| Source partition | Stalled or idle | Per-partition watermark lag | Windows for all keys stall | Idle timeout; alert on stuck partition | *Example:* 60 s |
| Sink | Transaction commit fails after recovery | Commit error with expired transaction | Lost output for one interval | Transaction timeout well above recovery time; alert | Manual replay from savepoint |
| Late data surge | Device backlog arrives hours late | Side-output rate | Results for old windows | Reconciliation job; optional backfill job | Next reconciliation |
| Poison event | Deserialisation or logic error | Task restarts in a loop | Whole job | Dead-letter route for bad records; restart backoff | *Example:* 5 min |
| [Component] | [Failure] | [Signal] | [Scope] | [Mitigation] | [Time] |

### 9.2 Security

| Threat | Control |
|---|---|
| Job code runs with broad data access | Per-job service identity; source and sink permissions scoped per job |
| Sensitive fields in state and checkpoints | Encrypt object storage and local disks; drop or tokenise personal fields at the source operator |
| Tampered artifacts | Signed job artifacts verified at submission |
| Data retention in checkpoints | Checkpoint retention aligned with the data retention policy; savepoints catalogued and expired |
| [Threat] | [Control] |

### 9.3 Observability

| Signal | Metric | Alert threshold |
|---|---|---|
| Freshness | Watermark lag per job | *Example:* above 120 s for 5 min |
| Input health | Source lag in records and time | Growing for 10 min |
| Checkpoints | Duration, size, failures, alignment time | 3 consecutive failures |
| Backpressure | Share of time backpressured per task | Above 50% for 15 min |
| Late data | Side-output events/s | Above baseline x 5 |
| Correctness | Daily diff versus batch recomputation | Any window above 0.1% difference |

## 10. Trade-offs and alternatives

<!-- Guidance: For each major decision list the options, what each costs and why one was chosen. Include what changes at ten times and a tenth of the load. -->

| Decision | Options | Chosen | Why | What we give up |
|---|---|---|---|---|
| Time domain | Processing time; event time | Event time with heuristic watermarks | Reproducible results under replay and delay | Watermark delay adds latency |
| Late data | Drop; keep forever; allowed lateness + side output | Allowed lateness + side output | Bounded state, nothing dropped silently | Corrections after the on-time result |
| State location | Remote database; embedded LSM; heap | Embedded LSM on SSD | Large state, incremental checkpoints | Local disk management |
| Checkpoint | Stop the world; aligned; unaligned; changelog | Aligned with unaligned fallback | Consistent cut, small checkpoints | Complexity of two modes |
| Output guarantee | At-least-once; idempotent upsert; two-phase commit | Upsert for stores, 2PC for logs | Exactly-once effect where needed | Commit latency for log readers |
| Engine | Build; adopt an open-source engine; managed service | [Choice] | [Reason] | [Cost] |
| At 10x load | 20 million events/s, 250 GB state | Partition sources further, raise parallelism within 1 024 key groups, add a standby-replica option to cut restore time | Restore of larger state dominates recovery | More hardware; checkpoint bandwidth becomes the bottleneck |
| At 0.1x load | 200 000 events/s, 2.5 GB state | Run on a shared session cluster with 8 tasks; heap state may suffice | Lower operating cost | Less isolation between jobs |

## 11. Delivery plan and risks

<!-- Guidance: How the design reaches production safely, and what could stop it. -->

### 11.1 Milestones

| Milestone | Scope | Exit criterion | Target date |
|---|---|---|---|
| M1 | One job, event-time windows, upsert sink, shadow run next to the batch job | Daily diff below 0.1% for 14 days | YYYY-MM-DD |
| M2 | Checkpoints with failure drills | Kill-worker drill meets NFR-04 three times | YYYY-MM-DD |
| M3 | Transactional log sink; downstream consumers switch | Consumers read committed output only | YYYY-MM-DD |
| M4 | Rescaling and upgrades via savepoints | Rescale 64 to 128 at peak without data loss | YYYY-MM-DD |

### 11.2 Rollout and rollback

[*Example:* Run the stream job in parallel with the batch job for one month;
dashboards switch per team behind a flag. Rollback restores the previous job
version from its last savepoint and points dashboards back to batch results.]

### 11.3 Risks

| ID | Risk | Likelihood | Impact | Mitigation | Owner |
|---|---|---|---|---|---|
| R-01 | *Example:* Watermark bound too tight, many events late | Medium | Medium | Measure the lateness distribution; tune `max_delay` per source | [@name] |
| R-02 | *Example:* State grows without bound in session windows | Medium | High | Session gap maximum; state TTL; alert on state size | [@name] |
| R-03 | *Example:* Transaction timeouts lose output after long recovery | Low | High | Timeout above worst-case recovery; drill | [@name] |
| R-04 | *Example:* Teams misread early firings as final numbers | High | Medium | `pane.timing` in every record; dashboards label speculative values | [@name] |
| [R-nn] | [Risk] | [L / M / H] | [L / M / H] | [Mitigation] | [@name] |

## 12. Appendices

<!-- Guidance: Supporting detail, glossary, open questions and references. -->

### 12.1 Open questions

| ID | Question | Owner | Due |
|---|---|---|---|
| OQ-01 | *Example:* Do finance reports read early firings or only final values? | [@name] | YYYY-MM-DD |
| OQ-02 | *Example:* Is a standby state replica worth the doubled write traffic? | [@name] | YYYY-MM-DD |
| [OQ-nn] | [Question] | [@name] | YYYY-MM-DD |

### 12.2 Glossary

| Term | Meaning |
|---|---|
| Event time | When the event happened, carried in the record |
| Processing time | When the engine processes the event |
| Watermark | The engine's estimate that no more events older than a timestamp will arrive |
| Trigger | Rule that decides when a window emits a result |
| Pane | One emission of a window; early, on-time or late |
| Allowed lateness | How long a window's state is kept after the watermark passes its end |
| Key group | The unit of keyed state that moves between tasks on rescale |
| Barrier | Marker injected into streams that separates records of two checkpoints |
| Savepoint | A user-triggered, retained checkpoint used for upgrades and rescaling |
| [Term] | [Meaning] |

### 12.3 Related templates

- Message Queue System Design (`system-design/design-message-queue`): the partitioned logs this engine reads and writes.
- CDC Pipeline System Design (`system-design/design-cdc-pipeline`): database changes as an input stream.
- Distributed Lock and Coordination Service System Design (`system-design/design-distributed-lock-coordination`): coordinator leader election and checkpoint pointer.
- Data Pipeline Specification (`ai-data/data-pipeline-spec`): contracts, quality checks and ownership for each job.
- Event Sourcing and CQRS (`architecture/event-sourcing-cqrs`): read models fed by stream jobs.

### 12.4 References

| Reference | Use |
|---|---|
| Akidau T. et al., "The Dataflow Model: A Practical Approach to Balancing Correctness, Latency, and Cost in Massive-Scale, Unbounded, Out-of-Order Data Processing", PVLDB vol. 8 no. 12, 2015 | Windows, watermarks, triggers, accumulation (7.1, 7.2, 7.6) |
| Carbone P. et al., "Lightweight Asynchronous Snapshots for Distributed Dataflows", arXiv 1506.08603, 2015 | Barrier snapshots (7.3) |
| Kleppmann M., Riccomini C., *Designing Data-Intensive Applications*, 2nd ed., 2026 (inspiration only) | Stream processing background |
| RFC 9457 Problem Details for HTTP APIs | Error bodies (4.4) |

## Standard and licence note

Based on the publicly described Dataflow model and asynchronous barrier
snapshot technique cited in 12.4; no standard governs stream processing
engines. This is a reference design to adapt, with illustrative numbers marked
*Example*. Original text written for this collection; the cited papers and
books were used as inspiration only and no text, figures or diagrams were
copied. Check the current versions of the papers and of any engine you adopt.
Template licence: MIT.