P07 — Stream-Processing System

Run it first. There is a companion page that builds this project's machinery as numbered, independently runnable blocks and then assembles them into one measured system: P07 hands-on — block by block (handson/h*.py). Every number on it was produced by running the code. Read it alongside the milestones below.

Medium · 88 hours · Weeks 76–83 · Stage 3 · Go

Table of Contents

• The Loop, Instantiated
• Why This Project Matters
• Prerequisites
• Duration and Size
• Central Technical Questions
• Architecture
• Showcase — Do This Before You Start
• Implementation Milestones
• Concepts To Study
• Primary-Source Readings
• Experiments
• Benchmarks and Metrics
• Correctness Tests
• Failure Tests
• Expected Difficulties
• Scope Boundaries
• Deliverables
• Exit Criteria
• Extension Ideas
• Connections
• References

The Loop, Instantiated

Why This Project Matters

Batch processing has an easy definition of correct: the output is a function of the input, and the input is finite. Streaming has no such luxury. When the input never ends, you must decide what "the answer" means before you can compute it, and every streaming system is a set of answers to that question.

A watermark is not a feature. It is a formal admission that you are giving up on completeness in exchange for the ability to emit a result at all. Once you have built one, you can never again read "exactly-once semantics" on a marketing page without asking exactly-once with respect to what?

You already operate Kinesis and Flink-class systems. This project turns your operational knowledge into design knowledge, and it does so on the specific question that matters most in a news-recommendation context: how fresh can the system be, and what does freshness cost in correctness?

Prerequisites

• P05 complete — the replicated log becomes the event log; the fault injector is reused
• P06 helpful — the scheduler and worker pool generalise
• P04 helpful — operator state is an LSM in every real system

Duration and Size

Medium, 88 hours, 8 weeks.

Central Technical Questions

1. What is the difference between event time and processing time, and what breaks if you use the wrong one? Give a concrete wrong answer your system would produce.
2. What is a watermark, formally? It is an assertion — state it as one, including what happens when the assertion is false.
3. Why is exactly-once delivery impossible, and what is achievable instead?
4. What does a checkpoint have to capture for recovery to be correct? The answer includes something people forget: input positions.
5. What is backpressure, and why is dropping data sometimes the correct response?
6. How large can operator state get, and what happens when it exceeds memory?

Event time vs processing time — the concrete failure

A user reads an article at 23:58 on a phone that is offline. The phone syncs at 00:07. Your "articles read per day" job, keyed on processing time, attributes that read to the wrong day. Run it on a month of data and every daily number is wrong by the size of the offline-sync population — a systematic bias, not noise, and one that correlates with exactly the users you care about.

Keyed on event time, the read lands in the right day, but now the 23:00–00:00 window cannot be closed at 00:00, because more events may still arrive. You have traded a wrong answer for a late answer. That trade is the whole subject of this project, and the watermark is the dial.

Architecture

Write your naive design first.

  producers ──► ┌──────────── log ─────────────┐
                │ partition 0: [0][1][2][3]... │   append-only, offset-addressed
                │ partition 1: [0][1][2]...    │   retention by time or size
                │ partition 2: [0][1][2][3]... │
                └──────────────┬───────────────┘
                               │  consumer group: one partition → one consumer
                               ▼
        ┌──────────────── operator ─────────────────┐
        │  deserialize ─► assign event time         │
        │  ─► watermark tracker (min over partitions)│
        │  ─► window assigner (tumbling/sliding/session)
        │  ─► state store (keyed, LSM-backed)       │
        │  ─► trigger on watermark ─► emit          │
        │  ─► allowed lateness ─► late-firing / drop│
        └───────────────┬───────────────────────────┘
                        ▼
                 sink (idempotent by (window,key) OR transactional)
                        │
        checkpoint: {operator state, input offsets, watermarks} ──► durable store

Watermarks, stated precisely

A watermark \(W(t)\) emitted at processing time \(t\) is the assertion:

No event with event time \(\le W(t)\) will arrive after this point.

Three consequences follow immediately, and they are the entire design space:

1. The assertion can be wrong. An event arriving with event time below the current watermark is late. The system must have a policy: drop it, fire the window again with a correction, or route it to a side output. There is no fourth option and no option that is free.
2. A watermark is a heuristic in any real system. Perfect watermarks require knowing the maximum possible delay, which you do not. Typical implementations use \(W(t) = \max(\text{observed event time}) - \delta\) for a chosen \(\delta\), and \(\delta\) is a completeness/latency dial with no correct setting.
3. Watermarks must be the minimum across all inputs. One idle partition holds the watermark back forever and every window stalls. This is the single most common operational failure in streaming systems, and it has a standard fix (idle-partition detection with a timeout) that itself weakens the guarantee.

Your E3 measures the \(\delta\) frontier: completeness (fraction of events included in their correct window) against latency (time from window end to result emission).

Exactly-once, disassembled

The phrase means three different things and only two are achievable:

The mechanism for the middle row: a checkpoint must contain operator state and the input offsets that produced it, written atomically. Recovering means restoring state and rewinding the input to the checkpointed offsets. If you checkpoint state and offsets separately, you get either duplicates or gaps depending on the order — which is a bug you should deliberately introduce and observe in E9.

This is the same idea as P06's atomic rename and P03's atomic segment flush. Three projects, one pattern: make the state transition and the position advance atomic. Say that in the report.

Showcase — Do This Before You Start

W4 · walkthroughs/w4_watermarks.py · ~45 minutes

A working miniature of this project: the same events counted three ways, and the completeness/latency frontier with its dead zone.

cd walkthroughs && python3 w4_watermarks.py

It is 80-ish lines and it surfaces this project's central surprise in an evening rather than in week six. Run it before committing the weeks.

Implementation Milestones

Concepts To Study

• Log abstraction: append-only, offset-addressed, retention; why a log is the right primitive for both messaging and state
• Partitions and keys: partitioning determines parallelism and ordering guarantees; ordering is per-partition only
• Consumer groups and rebalancing; the stop-the-world rebalance problem
• Event time, processing time, ingestion time — three clocks, all different
• Watermarks: the assertion, heuristic generation, the min-across-inputs rule, idle sources
• Window types: tumbling, sliding, session; and why session windows need merging
• Triggers and allowed lateness; the Dataflow model's separation of what, where, when, how
• State backends: in-memory vs LSM-backed; keyed vs operator state
• Checkpointing: stop-the-world vs Chandy–Lamport barriers; aligned vs unaligned
• Delivery semantics: at-most-once, at-least-once, exactly-once effects
• Backpressure: credit-based flow control vs blocking; why unbounded queues are the enemy
• Consumer lag as the primary operational metric

Primary-Source Readings

Budget: 12 hours.

Experiments

E3 is the project's headline. Plot completeness (fraction of events counted in their true window) against emission latency, one point per δ. That curve is the streaming version of P02's recall/QPS curve, and it makes the same point: you choose an operating point on a frontier; there is no correct answer, only a stated one.

E12 is the trap worth falling into on purpose. Session windows with no timeout accumulate state for every key ever seen. Watch memory grow, then implement state TTL and watch it stop. Unbounded state is the production failure of streaming systems.

Benchmarks and Metrics

Correctness Tests

1. Batch equivalence. For any finite prefix of the stream, streaming output with a sufficiently large watermark delay equals the batch job's output. This is the oracle; everything else is a special case of it.
2. Idempotent recovery: kill and restart at 20 random points; final output unchanged.
3. No lost events at any watermark setting — every event either lands in a window, fires late, or is explicitly counted as dropped. The three must sum to the input.
4. Offset/state atomicity: after recovery, the state matches exactly the events before the committed offset.
5. Window boundary correctness: an event exactly on a boundary lands in exactly one window. Test both boundaries.
6. Session merge: overlapping sessions merge correctly when a late event bridges them — the hardest window semantics to get right.
7. Watermark monotonicity: the watermark never goes backwards.
8. Idle-partition handling: watermark advances when one partition is silent.
9. Rebalance safety: no event processed twice, none skipped, across a group rebalance.
10. Bounded memory under sustained backpressure.

Failure Tests

Expected Difficulties

1. Session windows are the hardest semantics in the project. A late event can bridge two existing sessions, requiring a merge and a retraction of previously emitted results. Budget real time for milestone 7.
2. Watermark propagation across operators is subtle: each operator's output watermark is a function of its input watermarks and its own buffering. Get it wrong and windows fire early — silently.
3. "Exactly once" will tempt you into over-claiming. Be precise in the report about which of the three claims you implemented and under what sink assumptions.
4. Testing streaming is harder than testing batch because time is an input. Make the clock injectable from milestone 1 — every test drives time explicitly. Retrofitting this is a rewrite.
5. State growth is silent until it is fatal. Plot state size in every experiment from the start.
6. Backpressure that "works" by buffering is not backpressure. Test bounded memory explicitly (E10), not throughput.

Scope Boundaries

In scope: single-node or few-process, a partitioned log, one operator DAG of modest depth, event time and watermarks, three window types, checkpointing, backpressure, exactly-once effects.

Out of scope: a distributed scheduler with dynamic rescaling; SQL over streams; a query optimiser; multi-DAG multi-tenancy; a replicated log with consensus (P05 already did that — reuse or simulate); machine learning on streams; a web UI.

Deliverables

1. streamproc/ — Go, with the log, runtime, windowing, and checkpointing
2. REPORT.md centred on the E3 completeness/latency frontier
3. The batch-equivalence test harness — reusable and genuinely valuable
4. Notebook entries for E3, E5, E9, E12
5. A state-size-over-time plot for every experiment (the leak detector)

Exit Criteria

• Batch equivalence holds for all window types at sufficient watermark delay
• Recovery from ≥20 random kill points produces batch-identical output
• E3 complete: the completeness/latency frontier plotted across five δ values
• E5 complete: checkpoint-interval optimum identified and explained
• E9 complete: exactly-once effects verified with zero duplicates under duplicate delivery
• E10 complete: memory bounded under sustained backpressure, plotted
• E12 complete: unbounded state observed, then fixed with TTL, both measured
• Late-event policy implemented in all three variants and counted
• REPORT.md written with a falsified prediction

Extension Ideas

• Chandy–Lamport barrier snapshots across a multi-operator DAG, compared against stop-the-world. Measure the throughput impact of alignment — and of unaligned checkpoints under backpressure, which is the modern Flink answer.
• Retractions: emit corrections when late data changes a previously emitted result, and handle the downstream consequences.
• Watermark-delay auto-tuning from the observed lateness distribution. A research direction.
• Streaming joins with two watermarks and state expiry on both sides.

Connections

Backward: P05's replicated log becomes the event log; the fault injector is reused. P04 backs the state store. P06 supplies the worker/scheduler patterns.

Forward:

• → P08/P09: real-time interaction ingestion; the EMA user profile is a stateful streaming operator, and computing it here rather than in a batch job changes the freshness/complexity trade materially
• → P15: the ingestion layer. The research question "how do storage and indexing choices affect recommendation freshness?" is answered largely here

References

• 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. VLDB 8(12), 2015.
• Akidau, T. Streaming 101: The world beyond batch and Streaming 102. O'Reilly Radar, 2015.
• Carbone, P., Fóra, G., Ewen, S., Haridi, S., Tzoumas, K. Lightweight Asynchronous Snapshots for Distributed Dataflows. arXiv:1506.08603, 2015.
• Chandy, K. M., Lamport, L. Distributed Snapshots: Determining Global States of Distributed Systems. ACM TOCS 3(1), 1985.
• Kreps, J., Narkhede, N., Rao, J. Kafka: a Distributed Messaging System for Log Processing. NetDB 2011.
• Kreps, J. The Log: What every software engineer should know about real-time data's unifying abstraction. LinkedIn Engineering, 2013.
• Zaharia, M., Das, T., Li, H., Hunter, T., Shenker, S., Stoica, I. Discretized Streams: Fault-Tolerant Streaming Computation at Scale. SOSP 2013.
• Carbone, P. et al. Apache Flink: Stream and Batch Processing in a Single Engine. IEEE Data Engineering Bulletin 38(4), 2015.
• Abadi, D. J. et al. The Design of the Borealis Stream Processing Engine. CIDR 2005.
• Kleppmann, M. Designing Data-Intensive Applications, ch. 11. O'Reilly, 2017.