System Design Communication

Stream Processing

Instagram scroll traffic never stops: every tap, impression, and pause becomes an event. Fraud rules, live dashboards, and ranking features cannot wait for tonight's . They need answers while the session is still happening.

means consumers continuously read those events, keep , and write to sinks. The hard part is not "read forever". It is , , and proving your sink is safe under retries.

Clickstream journey

Events in a log

Clicks land in partitions with timestamps.

Key by entity

Group by user, post, or device.

Slice event time; enrich with tables.

Features, alerts, dashboards.

Try it: when a slow partition stalls the watermark

Stream windows close on event time, not when your laptop receives bytes. The watermark is roughly the minimum event time seen across partitions minus allowed lateness. One idle or slow partition keeps the watermark low, so event-time lag can look fine on offset charts while business windows stay open.

Allowed lateness: 5sTumbling window size: 10s

Partition 0 (healthy)

Event time: 0s

Partition 1 (slow or idle)

Event time: 0s

Wall processing time0s

Global watermark (event time)-5s

Next window can close through-1s

Open tumbling window ends at event time-1s

Watermark lag here is how far wall clock has run ahead of the watermark in seconds (0s max skew across partitions).

Watermark: -5s
Watermark lag (max): 0s
Windows closed through: -1s
Partition 0 event time: 0s

Watermarks estimate event-time progress. The global watermark is the minimum across partitions minus allowed lateness. Toggle partition 1 idle to watch windows stall.

Idle partition. Mark partition 1 idle and advance twenty seconds. Partition 0 races ahead, but the watermark sticks to partition 1's last event time minus lateness, delaying window closes for the whole job sharing that watermark.

Separate metrics. Low Kafka offset lag does not imply low event-time lag during backfill. Track watermark age and per-partition event-time skew alongside consumer lag.

Model scope: two partitions, fixed tumbling windows, and a single global watermark. Production engines add idleness detection and per-key timers.

Problem → naive → break

Concrete problem. Fraud and live ranking need answers while the session is still open. Waiting for tonight's partition job means the bad actor already scrolled away.

Naive system. Count clicks with GROUP BY minute using wall-clock on whatever worker happens to see the message. On crash, restart from latest offset and hope counts are "close enough."

Break. A phone was offline; its clicks arrive with old after your minute bucket already closed. Dashboards undercount. After a crash, replay doubles alert fires into an HTTP sink that is not idempotent.

Counterexample to "enable exactly-once and you are done." Engine usually means state and compatible sinks align with . External APIs still need : the label is not magic end-to-end.

Correctness invariant

What must never become false

Window results obey the product's + + lateness contract (drop, side-output, allow lateness, or retract), never silent processing-time buckets when late data matters. After restore, operator state and transactional sinks match the last successful checkpoint barrier. External side effects remain safe under replay via idempotent keys or upserts. Recovery time + oldest input age must stay below source retention, or the invariant becomes unrecoverable data loss.

What actually happens

The feed looks like one screen. Underneath, a pipeline treats data as never-ending: parse, key, aggregate, sink. Unlike , the engine never gets a clean "end of file." It keeps and advances a so can close without waiting forever.

A phone that was offline can emit a click with an old after has moved on. That is why late-data policy matters: drop, side-output, allow lateness, or emit retractions: each choice is a product promise, not a framework quirk.

When a worker crashes, the job restores a and replays. Retries can duplicate side effects, so plus an is the honest story, and "" only holds where state and sinks commit together.

Events in log

Parse and validate

Key by entity

Window or join

State store

Checkpoint

Replay after failure → re-key

Sink: alerts, features, dashboards

How to choose batch vs stream

1. How fresh must the answer be? Nightly revenue can stay . Fraud flags, surge pricing, or live counters usually need a stream.

2. Do you need unbounded state? Session counts, joins with changing users, and pattern detection force , watermarks, and recovery planning.

3. Can the sink absorb duplicates? If not, design an before you claim strong guarantees, or keep the path batch where simplicity wins.

Simple rule of thumb

Prefer when latency budgets allow it. Reach for a stream when continuous stateful pipelines earn their complexity. Model in , pick deliberately, and prove sink idempotency before trusting vendor "exactly-once" labels.

Batch vs stream

Batch bounds inputs (daily partitions); stream engines treat data as never-ending and manage state incrementally: complexity buys latency.

Batch

Finite datasets and schedules
Simpler correctness story
Minutes-to-hours latency typical

Stream

Unbounded pipelines
Stateful operators + watermarks
Milliseconds-to-seconds latency targets

What to watch

Input lag

How far the job is behind source partitions; it is the early warning before retention loss.

Checkpoint duration

Long checkpoints predict painful recovery and can create their own backpressure.

State size and TTL

Unbounded state makes restarts slow and storage expensive; every keyed store needs cleanup rules.

Capacity math: stream pipelines

Stream jobs need enough input partitions, operator parallelism, state storage, and sink capacity to stay ahead of while retaining enough history for replay.

Analytics clicks

Ingest partitions

Windowed operators

Keyed state

Checkpoint storage

Aggregates and alerts

Idempotent sink writes

Metric	Formula	Analytics clicks example
Ingress events/sec	users × events/user/day ÷ 86,400 × peak factor	1M × 100 clicks/day × 6× peak ≈ 6.9k/s
Parallelism	ceil(peak events/sec ÷ safe operator events/sec)	6.9k/s ÷ 500/s ≈ 14 workers before headroom
State size	active keys × bytes/key × windows retained	1M users × 256 B × 4 windows ≈ 1 GB before overhead
Backlog drain time	lag events ÷ (processing rate - input rate)	2M lag ÷ (10k/s - 7k/s) ≈ 11 minutes
Retention storage	events/sec × payload bytes × seconds retained × replicas	7k/s × 600 B × 7 days × 3 ≈ 7.6 TB raw

How to choose

Backpressure math in stream interviews

Prove drain capacity: Show the fleet can process faster than peak input after a deploy pause, not just during steady state.
Respect partitions: Adding workers only helps up to partition and key parallelism; hot keys still serialize.
Protect retention: Alert on lag age relative to retention so a stuck job does not lose replayable input.

Stream bottleneck map

Source partitions cap read parallelism.
Key skew caps per-key ordering and state ownership.
Stateful operators cap on state input/output and time.
Sinks cap on write throughput, transaction latency, and dedupe cost.

Recovery worksheet

Restore time = checkpoint bytes / restore bandwidth + replay lag / processing headroom. Example: 600 GB checkpoint at 300 MB/sec takes about 34 minutes before replay. Add catch-up time and verify oldest input age plus recovery time is below retention.

Processing guarantees

End-to-end links sources, state, and sinks with transactional commits: always read what your version actually promises versus best-effort sinks.

At-most-once

With at-most-once handling, progress can be recorded before the corresponding output is durable. Suppose the consumer saves offset 101 and then crashes before writing the result for event 101. Restarting at 102 avoids a duplicate but permanently skips that result. This can suit disposable telemetry; it cannot support a count whose missing events must eventually be recovered.

At-least-once

duplicate outputs: or dedup keys required.

Exactly-once effect

A checkpoint records a consistent recovery point for source positions and operator state. After a crash, the engine restores that state and rereads later input. A compatible transactional sink coordinates its visible output with checkpoint completion, preventing partial results from appearing twice. An ordinary HTTP call is outside that protocol; it needs a stable operation key or another explicit way to make retries harmless.

python

pipeline.enable_checkpointing(every_ms=10_000)
pipeline.set_commit_mode("checkpoint_aligned")
sink.use_idempotent_writes(key="event_id")

How to choose

Exactly-once caveat for stream processing

Scope: Exactly-once usually means state updates and compatible sinks commit with checkpoints; external APIs still need idempotency.
Trade-off: Stronger guarantees can raise latency, require transactional sinks, and make recovery behavior more complex.
Answer style: Say what is exactly-once within the engine, then state how the sink handles retries and duplicates.

Sink guarantee matrix

Append-only sink

Duplicates must be filtered later or accepted by analytics semantics.

Upsert sink

Use stable keys so retries replace the same row.

Transactional sink

Commit output and checkpoint progress in one compatible boundary.

External side effect

Wrap calls with idempotency keys and store requested effects before retrying.

Windowing

Windows slice into buckets: for disjoint intervals, for overlaps, for gap-based bursts.

Tumbling

A tumbling window divides time into non-overlapping intervals. For five-minute windows, an event at 10:03 belongs to the 10:00–10:05 interval, while an event at exactly 10:05 belongs to the next interval under a start-inclusive, end-exclusive convention. Each event contributes to one bucket, making this useful for per-period totals. The watermark and late-data policy determine when that total becomes final.

[0–5)[5–10)

Sliding

A sliding window has a duration and a separate advance interval. A five-minute window advancing every minute produces overlapping views such as 10:00–10:05 and 10:01–10:06. An event at 10:03 contributes to several views, giving a smoother moving average but increasing state and update work. State the duration and slide together; saying only “five-minute sliding window” leaves the calculation ambiguous.

|----w1----|
   |----w2----|

Sessions

A session window groups activity separated by less than a chosen inactivity gap. With a ten-minute gap, clicks at 10:00 and 10:07 belong together, whereas a click at 10:30 starts another session. Late events can bridge two previously separate sessions, so the engine may need to merge state and revise outputs. This is why session results need a clear policy for lateness and finalization.

Watermarks and lateness

≠ ; estimate how far event-time has progressed so operators can close windows while tolerating skew.

Diagram text: the exact Mermaid definition below preserves the nodes, labels, and connections. Enable JavaScript to view the rendered diagram.

Read the diagram definition

sequenceDiagram
    participant P0 as Partition 0
    participant P1 as Partition 1
    participant W as Watermark
    P0->>W: events up to 10:05
    P1->>W: idle at 10:01
    Note over W: Window cannot close past slow partition
    P1->>W: heartbeat or idleness marker
    Note over W: Watermark advances safely

Failure timeline: idle partition stalls the watermark

T0  P0 events at event-time 10:05; P1 idle since 10:01
T1  Global watermark stuck near 10:01: windows past 10:01 cannot close
T2  Dashboards look live but aggregates freeze; operators blame CPU
T3  Idle heartbeat / watermark timeout advances past P1; late policy decides fate of stragglers

Diagram text: the exact Mermaid definition below preserves the nodes, labels, and connections. Enable JavaScript to view the rendered diagram.

Read the diagram definition

sequenceDiagram
    autonumber
    participant S as Source
    participant J as Job
    participant C as Checkpoint store
    participant K as HTTP sink

    S->>J: events ... barrier
    J->>C: snapshot state + offsets
    Note over J: Worker crash after sink call, before next barrier
    J->>C: restore last checkpoint
    S->>J: replay from barrier
    J->>K: may call again: need idempotent sink

T0  Checkpoint barrier commits state + source offsets
T1  Alert HTTP call succeeds; worker dies before next barrier
T2  Restore + replay re-emits the same alert key
T3  Idempotent sink dedupes; naive POST doubles the page

Definitions

Event time: when it happened upstream
Processing time: when the engine observes it

Watermark

A watermark is the pipeline’s estimate of progress through event time, not proof that an older event can never arrive. If the watermark reaches 10:05, the operator may publish a result for a window ending then. A delayed phone upload can still contain a 10:03 event. Decide whether to revise the earlier result, route that event for correction, or discard it, and expose that choice in the meaning of the reported metric.

python

stream.with_event_time(
    timestamp_field="occurred_at",
    allowed_lateness_seconds=5
)

Late data options

Drop
Side output
Allowed lateness window
Emit retractions / updates

Lag can mean different things

Input lag: unread events by source partition.
Event-time lag: watermark is behind because one partition is idle or late.
Processing-time delay: operators are slow even if watermarks move.
Sink commit lag: results are computed but not durably visible.

Stream–table duality

A table stores the latest value for each key; a change stream describes the sequence of updates that produced it. Applying each product-price change to a keyed table maintains the current catalog. Going the other way, CDC exposes table changes as a stream. A join must state whether it wants the latest price now or the price that was valid when an order happened, because those questions can produce different totals.

python

orders = stream("orders").key_by("order_id")
users = table("users").key_by("user_id")
orders.join(users, enrich_order)

Interval joins correlate events within time bounds, for example attribution windows between impressions and conversions.

Stateful execution

Aggregations and joins keep partitioned state: back it with durable changelog topics or sized to your Service Level Agreement (SLA) for recovery time.

Library on brokers

Local embedded state with changelog replication: great when ingress and egress stay on the same log.

Clustered engine

Distributed snapshots to object storage; scales complex DAGs.

Capacity

State grows with active keys, the retained history per key, and the number of overlapping windows or pending joins. A time-to-live policy removes inactive state, but an expiry that is too short can lose information needed by a legitimate late event. Estimate bytes per active key and checkpoint size, then test restore time. Adding workers helps only if the key distribution also spreads the work.

Producer rate

Input lag

Operator saturated?

CPU bound

Scale parallelism

State I/O bound

Tune state backend and checkpointing

Sink slow

Throttle source or buffer

Recover lag before retention expires

State storage worksheet

active keys * bytes/key * windows retained estimates hot state.
checkpoint size * checkpoint replicas estimates recovery storage.
changelog bytes/sec * retention seconds estimates replayable state history.
Recovery is safe only when oldest input age + restore time + catch-up time stays below stream retention.

How to choose

Backpressure and consumer lag in a streaming interview

Name the symptom: Lag rising while input rate exceeds processing rate, or checkpoints slow enough to stall progress.
Find the bottleneck: Check partition skew, hot keys, state IO, sink latency, and rebalance churn before blindly adding workers.
Mitigate safely: Increase parallelism only within partition/key constraints, reduce expensive joins, throttle sources, or degrade non-critical outputs.

Engine flavors

Embedded stream library

Library embedded in apps
Best when ingress/egress stay on the same log

Clustered streaming engine

Low-latency event-time features
CEP and advanced state

Micro-batch stream engine

Micro-batch trade-offs
Unified with batch jobs / ML plumbing

Complex event processing

Detect multi-step patterns (three failures within five minutes, geo-hop fraud) declaratively instead of hand-rolled counters.

python

pattern("three_failures")
    .where(event_type="Failure")
    .times(3)
    .within(minutes=5)

Real-time personalization (illustrative)

Clickstream and playback signals feed ranking features continuously; separate serving latency from offline training bulk jobs: often hybrid + stream.

High QPS

Edge + ingest

Features

Windowed counts

Serving

Low-latency APIs

Ops

Lag alerts

Numbers vary wildly by product: optimize instrumentation before chasing competitor benchmarks.

Break the improved design

Event-time windows and checkpoints fix naive counts, then fail differently: one idle partition freezes the watermark for everyone; allowed lateness grows until restore exceeds retention; hot keys serialize despite more workers; long duration creates its own .

A transactional sink helps inside the engine, but a third-party webhook still double-fires on replay without an idempotency key. Micro-batch engines trade simpler recovery for higher latency: pick deliberately when fraud SLOs are sub-second.

Trade-offs

Batch vs stream

Batch keeps a finite correctness story; stream buys freshness at the cost of watermarks, state, and recovery math.

Lateness policy

Drop is simple and wrong for money; retract/update is correct and harder for downstream caches to consume.

Stronger sink guarantees

Checkpoint-aligned commits raise latency and require compatible sinks; external effects still need idempotency.

Exercises

1. Allowed lateness as a product promise

A 5-minute revenue window allows 2 minutes of lateness. State the user-visible consistency contract in one sentence, and decide whether late events drop, side-output, or retract: given finance cannot accept silent undercounts.

2. Checkpoint vs retention race

900 GB keyed state restores at 300 MB/s (~51 min) before replay. Source retention is 60 minutes. What do you change first: state TTL, checkpoint frequency, retention, or parallelism, and how do you prove recovery is safe?

3. Exactly-once claim in review

A doc enables engine exactly-once and POSTs alerts to PagerDuty. Write the precise scope of the guarantee and the sink change you require before merge.

Going deeper

Production deep dive: checkpoints, time, and state

A stateful stream processor is a continuously advancing distributed database. Correct recovery must restore source offsets, operator state, timers, and sink effects to one coherent checkpoint.

What must stay true

After recovery, every event before the restored checkpoint is reflected exactly once in recoverable state, and every later event remains available for replay.

How the pieces fit

Sources

offsets + barriers

Aligned checkpoint

barrier at every input

State snapshot

keys + timers

Transactional sink

pre-commit output

Checkpoint complete

commit offsets + sink

Input

120k events/s

Peak source arrival rate.

Processing

100k events/s

Current sustainable operator rate.

Lag growth

1.2M/min

(120k − 100k) × 60; backpressure alone cannot fix this.

Aligned checkpoints

A barrier flows through each source partition. A multi-input operator pauses channels whose barrier arrived and keeps processing the others until all barriers align, then snapshots state. Alignment preserves a clean cut but can amplify backpressure when one input is slow; unaligned checkpoints trade larger snapshots for faster progress.

State recovery

Restore the latest completed snapshot, reset sources to recorded offsets, recreate event-time timers, and replay the tail. Measure restore bytes/second and retained-source headroom; CPU recovery is irrelevant if 900 GB of state takes an hour to download.

Backpressure

Bounded buffers propagate a slow sink upstream so memory does not grow without limit. Watch busy time, backpressured time, queue depth, and watermark delay. Sustained input above service rate requires more parallelism, less work, or admission control, not a bigger buffer.

Rescaling and rebalancing

Checkpoint first, assign key groups to the new parallel tasks, transfer only their state, then resume each input partition from the checkpoint. Expect a temporary throughput dip and cache cold start; canary the new parallelism on representative skew.

Sink guarantees

A transactional sink pre-commits under the checkpoint ID and commits only when the coordinator completes it. Otherwise use deterministic keys/upserts or an idempotency ledger. “Exactly once” ends at the weakest source, operator, or sink boundary.

Worked example: Window and watermark timeline

1Window W covers event time [10:00, 10:05); allowed lateness is two minutes.
2Event A (10:01) arrives at 10:01. Event B (10:04) arrives at 10:06. When watermark reaches 10:05, W emits an initial result.
3B arrives before the cleanup watermark 10:07, so W updates and emits a correction. Event C (10:03) arriving at 10:08 goes to a late-event side output or is dropped by policy.

The dashboard is explicitly provisional until watermark 10:07; its product contract must permit corrections during that interval.

Recovery failure matrix

Scenario	Consequence	How to contain it
Worker dies during snapshot	Partial checkpoint cannot describe a consistent cut.	Discard it and restore the previous completed checkpoint.
Sink commits before checkpoint	Replay duplicates external output.	Use two-phase sink commit or deterministic idempotent writes.
Hot key owns 40% of traffic	Adding workers does not move that single key faster.	Pre-aggregate, split key with controlled merge, or redesign partitioning.

What to watch in production

Track input/output rate, per-partition lag, watermark skew, checkpoint duration/failures, state bytes, restart time, and late-event rate.
Load-test a rescale with production key skew and verify checkpoint storage bandwidth.
Keep source retention longer than worst restore + replay + incident response time.

A scenario to reason through

Checkpoints now take 70 seconds but run every 60 seconds, and watermark lag is rising. What do you change first?

One solid approach

Find the slow operator/channel, verify state-store and checkpoint bandwidth, then reduce snapshot churn or use incremental/unaligned checkpoints. Do not merely lengthen retention while overlapping checkpoints keep consuming capacity.

Takeaway

Model in , pick deliberately, and prove when you claim strong guarantees. Prefer embedded libraries when everything stays on one log; reach for a clustered engine when DAG complexity or pattern detection dominates. Watch duration and state size. They predict recovery pain earlier than CPU graphs. Watch before retention expires.

Topic-specific references

The Dataflow Model (Akidau et al.) : event time, watermarks, windows, and triggers as first-class concepts.
Apache Flink: Fault Tolerance / Checkpoints : what exactly-once means for state and compatible sinks.
Kafka Streams processing guarantees : at-least-once vs exactly-once within the Streams + Kafka transactional model.

Practice and further reading

Use these exercises to connect the lesson to a design decision. State your assumptions, trace what happens to one request, and explain the rule that must remain true even during a failure. That rule is the design's correctness invariant.

Focus for this lesson: Reason about event time, watermarks, windows, state, checkpoints, backpressure, rescaling, late data, and sink guarantees.

Explain the decision

What does a watermark mean?

Read the reasoning

It is the processor’s estimate that event time has advanced; events older than it are late according to policy, not physically impossible.

Apply it to a production scenario

A sink slows, checkpoint duration rises, and lag grows. What do you inspect?

Read one approach

Backpressure propagation, in-flight buffers, state size, checkpoint alignment, sink batching, partition skew, and whether admission must reduce.

Work through the numbers

A keyed operator keeps 200 bytes for 30 million active keys with two copies. Estimate state.

Read the worked calculation

12 GB raw, before serializer, index, checkpoint, and incremental-history overhead.

Reason through a failure

What must exactly-once processing actually guarantee?

Read the recovery reasoning

Recovered operator state and sink commits agree on one checkpoint epoch; external side effects need transactional or idempotent integration.

Sources and further study

Continue with Async vs Sync Communication.