Streaming Architecture — Flink + Kafka deep-dive.

Kafka as the spine, Flink as the engine.

Most modern streaming designs are some flavor of: events into Kafka, transformations in Flink, sink to a serving store. Every interview prompt — fraud detection, real-time recommendations, IoT telemetry, click-stream attribution — fits this skeleton. The contents change; the shape doesn't.

Why Kafka is non-negotiable for the spine

Why Flink usually beats the alternatives

The five questions to ask BEFORE drawing

1. What's the latency target? Sub-second changes the architecture; minutes is just batch with smaller windows.
2. What's the event volume? 10K/sec vs 10M/sec is a different system entirely.
3. Is state involved? Stateless filters are easy; stateful joins / windowed aggregations need RocksDB + checkpoints.
4. What's the ordering requirement? Per-key order is cheap (partition key); global order is expensive.
5. How do you replay if the job crashes? Kafka offsets + Flink savepoints define the answer.

Event-time vs ingest-time vs processing-time.

Every event has at least three timestamps. Knowing which one to use is the most-asked streaming question after "what's a watermark".

Why event-time is almost always right

Imagine a click on a banner ad at 3:00:00 PM. The mobile is offline, uploads at 3:30:00 PM. Three timestamps:

• Event-time: 3:00:00 (when the click happened)
• Ingest-time: 3:30:00 (when Kafka received it)
• Processing-time: 3:30:01 (when Flink processed it)

The "click happened at 3pm" is the truth users care about; the other two are infrastructure timestamps. Always partition / window by event-time.

The store-all-three discipline

// Every event row should carry all three timestamps:
{
  "user_id": "U_001",
  "event":   "click",
  "event_ts":     "2024-09-12T15:00:00Z",   // client-set
  "ingest_ts":    "2024-09-12T15:30:00Z",   // Kafka-set
  "processing_ts":"2024-09-12T15:30:01Z"    // Flink-set
}

Partition / window by event_ts. Alert / debug by processing_ts - event_ts latency. Audit by ingest_ts.

When event-time goes wrong

Watermarks — the late-event contract.

A watermark is a timestamp the engine commits to: "I've waited long enough; I do not expect any event with event-time earlier than this watermark to arrive from now on." It's the contract that lets windowed aggregations close.

The mental model

Watermark generation strategies

// Flink — bounded-out-of-orderness watermark, 10s tolerance
WatermarkStrategy
  .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
  .withTimestampAssigner((event, timestamp) -> event.getEventTs())
  .withIdleness(Duration.ofMinutes(1));   // idle-source escape

What happens when an event arrives AFTER its watermark

The event is late. Three options:

1. Drop — the simplest; you lose the event. Default in many engines.
2. Side-output — route late events to a separate stream for offline processing / alerting.
3. Allowed lateness + update — keep the window state for an extra grace period; emit an UPDATE result when a late event arrives.

// Flink — windowed aggregation with allowed lateness + late-event side output
DataStream<Result> results = events
  .keyBy(Event::getUserId)
  .window(TumblingEventTimeWindows.of(Time.minutes(5)))
  .allowedLateness(Time.minutes(2))                    // grace period
  .sideOutputLateData(LATE_EVENTS_TAG)                  // side output the late ones
  .aggregate(new MyAggregator());

DataStream<Event> lateEvents = results.getSideOutput(LATE_EVENTS_TAG);

The trade-off

The right answer is empirical: measure the 99th-percentile event-arrival lag for your pipeline; pick watermark lag = P99 + safety margin. Re-tune quarterly.

Exactly-once — the 4-link end-to-end chain.

"Exactly-once" is not a feature you turn on. It's a chain of four guarantees, and the chain is only as strong as the weakest link. If your sink isn't idempotent, the producer / consumer / engine-level guarantees don't matter.

The four links

Producer-side idempotency (link 1)

# Kafka producer config
enable.idempotence       = true
acks                     = all
max.in.flight.requests   = 5    # ≤ 5 with idempotence enabled
retries                  = MAX

# What this gives you:
# - Producer attaches (producer_id, sequence_number) to every batch
# - Broker rejects duplicates with the same (PID, seq) pair
# - Retries are safe — no double-write

Consumer + engine atomicity (links 2 + 4)

Flink's two-phase commit barrier:

1. Periodic checkpoint barrier flows through the operator graph alongside data.
2. Each operator snapshots its state (RocksDB → S3) when the barrier arrives.
3. Sink operator pre-commits — the next batch of output is staged but not visible to consumers.
4. When all operators acknowledge, the JobManager commits — pre-committed sink batches become visible.
5. Crash before commit → restart from last successful checkpoint, replay from Kafka offset stored in checkpoint.

Sink-side idempotency (link 3) — the hardest link

When at-least-once is OK

Exactly-once is expensive (latency cost of two-phase commit, throughput cost of transactions). Skip it when:

• Downstream is itself idempotent (counter increment via SUM is naturally tolerant of duplicates? No — but distinct count of events IS, with proper deduplication on event_id).
• The metric is approximate anyway (DAU, click-through-rate — small dup counts wash out).
• You can run a daily reconciliation batch that fixes any dup damage.

Backpressure, checkpointing & state backends.

The interview moves from semantics to operations. "What happens when the job lags?", "How do you handle 200GB of state?" — these are the questions that separate someone who's read the docs from someone who's run Flink in production.

Backpressure — what it is

Backpressure = downstream operator can't keep up with upstream's rate. Flink's transport layer signals upstream to slow down. Symptoms:

• Kafka consumer lag rising — you're consuming slower than producing.
• Operator CPU pinned at 100%, but throughput flat.
• End-to-end event latency growing without bound.

Backpressure causes — the four classics

Async I/O — the #1 fix for slow external calls

// BAD — synchronous; one slow call blocks the whole subtask
DataStream<Enriched> enriched = events.map(event -> {
  Profile profile = http.get("/profile/" + event.userId);  // blocks!
  return enrich(event, profile);
});

// GOOD — async; many in-flight requests, bounded by capacity
DataStream<Enriched> enriched = AsyncDataStream.unorderedWait(
  events,
  new ProfileLookup(),
  /*timeout*/ 1, TimeUnit.SECONDS,
  /*capacity*/ 100   // max in-flight async requests
);

Checkpointing — what to know

Aligned vs unaligned checkpoints

State backends — the three flavors

The "checkpoint storm" anti-pattern

Schema evolution in a streaming world.

Schema changes break streaming jobs catastrophically. A producer adds a field; the consumer's deserializer fails; the entire pipeline stalls. The schema registry pattern (covered in the System Design page) is how you survive change.

The four schema-evolution scenarios

The dual-write deprecation pattern

Rename username → display_name:

1. Phase 1 — Add new field: producer writes BOTH username and display_name with same value.
2. Phase 2 — Migrate consumers: each consumer team switches to read display_name; mark username as deprecated in the schema.
3. Phase 3 — Drop the old field: after deprecation window (typically 90 days), producer stops writing username; schema removes it.

Stream-Table joins with evolving dimensions

If your stream joins to a slowly-changing dim table, schema changes on the dim ripple through:

// Flink SQL — streaming fact joined to versioned dim (event-time temporal join)
SELECT
  e.user_id,
  e.event_ts,
  d.tier,
  d.region
FROM events e
JOIN dim_user FOR SYSTEM_TIME AS OF e.event_ts AS d
  ON e.user_id = d.user_id;
-- Joins against the dim VERSION current at event time, not now.
-- Schema-evolves cleanly: as long as new fields have defaults.

CDC streams — schema evolution at the source

Most streaming pipelines today have CDC (Debezium → Kafka) as a source. The OLTP database adds a column → Debezium emits the change with the new schema. Three things to set up:

1. Schema registry per CDC topic with backward compatibility
2. Tolerant Flink deserializer that skips unknown fields rather than crashing
3. Alert on schema-version mismatch — when consumer's expected version drifts > 1 from latest, page the team

The 8-step play-event walkthrough (Netflix-flavored).

Putting it all together — every concept above shows up in a single canonical walkthrough. The interview prompt: "design Netflix's playback-event pipeline.". The 8-step answer:

Step 1 — Producer: client SDK emits event

// On every play-start, pause, resume, complete:
{
  "event_id":     "evt_8f3a...",   // client-generated UUID — idempotency key
  "user_id":      "U_001",
  "profile_id":   "P_005",
  "title_id":     "T_strangerthings_S5E1",
  "event_type":   "play_start",
  "event_ts":     "2024-09-12T19:00:00Z",   // client clock
  "device":       "tv",
  "country":      "US"
}

The event_id is the idempotency key — every retry uses the same id; the broker dedupes.

Step 2 — Kafka: durable spine

Topic play_events, partitioned by profile_id (so one profile's events stay in order), 30 partitions, 7-day retention (shorter than typical because we Iceberg-archive everything).

Step 3 — Flink job: enrich + sessionize

// Pseudo-Flink:
events
  .keyBy(Event::getProfileId)
  .map(new ProfileEnrichment())               // async I/O lookup to dim_profile
  .process(new SessionAssigner())              // 30-min idle → new session_id
  .keyBy(e -> e.userId + "|" + e.titleId)
  .window(SessionWindow.withGap(Time.minutes(30)))
  .aggregate(new SessionMetrics())            // total watch_minutes per session
  .addSink(new IcebergSink());                // exactly-once via 2PC + Iceberg

Step 4 — Watermarks: 60-second lateness budget

Mobile uploads can lag (subway tunnel = 2-minute spike). Set watermark = max_event_ts − 60s. P99 lateness in production is 8s, so 60s is comfortable. Side-output anything later for the daily reconciliation job.

Step 5 — State: per-profile session window

State backend is RocksDB with 30-second incremental checkpoints to S3. Session state is small (a few hundred bytes per profile) but multiplied by 250M profiles = 50GB working set. RocksDB handles this comfortably; HashMap would OOM.

Step 6 — Sink: Iceberg upsert by event_id

Iceberg fct_play_event table partitioned by DATE(event_ts). Sink writes use Iceberg's two-phase commit; uniqueness on event_id means retries are no-ops. Downstream Flink re-streams and dbt batch jobs both read from this Iceberg table.

Step 7 — Serving: low-latency analytical store

Per-title rolling counts (last 1h, 24h, 7d) shipped to Pinot for sub-100ms queries. The TopN-by-country dashboard reads Pinot directly; the historical reports read the Iceberg table via Spark or Trino.

Step 8 — Operations: the 4 SLOs

The 90-second articulation script.

Three sentences that signal seniority — in any streaming round

1. "Watermarks are a contract, not a clock — the engine commits to not seeing earlier events, and downstream windows close on that commitment."
2. "Exactly-once is end-to-end; if the sink isn't idempotent, the upstream guarantees don't matter."
3. "Backpressure is rarely about throughput — it's almost always skew, sync I/O, state size, or sink bottleneck. Diagnose by exclusion."