Definition
- Forwarding exchange – the cheap, local data transfer between chained operators where subtask n feeds subtask n with no network shuffle.
- Redistributing exchange – the network shuffle caused by keyBy, which hashes the key and routes records across the network to the owning subtask.
- Parallelism – the number of parallel subtasks per operator, typically matched to Kafka partition count for the source.
- Tumbling window – fixed, non-overlapping time buckets (e.g., 30 min) that group events by event time.
- Checkpoint barrier – a marker injected into the stream that triggers each operator to snapshot its state.
- Idempotent sink – a JDBC upsert keyed by user and window start so replayed events do not duplicate rows.
Picture the National Bank of India tracking physical debit card deliveries. Events about each card's journey stream in through Kafka, already sorted into lanes by customer. A first station stamps each event with extra details (the map enrichment), and because it does not change lanes the parcels flow straight through (forwarding). Then parcels are re-sorted so everything for one customer goes to the same clerk (the keyBy shuffle), and each clerk tallies that customer's deliveries over each half hour (the tumbling window). The tallies are filed into the central ledger (the PostgreSQL sink). If the power blips, the bank does not lose count because it periodically photographs the whole line's progress (the checkpoint).
Understanding the Concepts
How does the Kafka source work?
The source reads a Kafka topic partitioned by user id. You match source parallelism to partition count so each subtask owns a share of partitions. The source assigns timestamps and watermarks from event time so downstream windows work correctly.
package com.bank.delivery;
public class KafkaSourceDesign {
public static void main(String[] args) {
System.out.println("Kafka topic: delivery-events");
System.out.println("Partitions: 12 (partitioned by user_id)");
System.out.println("Source parallelism: 12");
System.out.println(" -> 1 partition per subtask");
System.out.println();
System.out.println("Watermark strategy:");
System.out.println(" BoundedOutOfOrderness(5 seconds)");
System.out.println(" Idle partition timeout: 30 seconds");
}
}
Kafka topic: delivery-events
Partitions: 12 (partitioned by user_id)
Source parallelism: 12
-> 1 partition per subtask
Watermark strategy:
BoundedOutOfOrderness(5 seconds)
Idle partition timeout: 30 seconds
What is a forwarding exchange?
A forwarding exchange is the cheap, local data transfer between chained operators. Because the map enrichment does not change partitioning, Flink connects source to map with a forward: subtask n feeds subtask n with no network shuffle. The two chain into one task for efficiency.
package com.bank.delivery;
public class ForwardingExchange {
public static void main(String[] args) {
System.out.println("Source subtask 0 -> Map subtask 0");
System.out.println("Source subtask 1 -> Map subtask 1");
System.out.println("Source subtask 2 -> Map subtask 2");
System.out.println("... (no network shuffle)");
System.out.println();
System.out.println("Forwarding: local, cheap, chained.");
System.out.println("Both operators run in one task slot.");
}
}
Source subtask 0 -> Map subtask 0
Source subtask 1 -> Map subtask 1
Source subtask 2 -> Map subtask 2
... (no network shuffle)
Forwarding: local, cheap, chained.
Both operators run in one task slot.
What is a redistributing exchange (keyBy)?
keyBy performs a redistributing exchange: Flink hashes the user id to a key group and routes each record across the network to the subtask responsible for that key group. This is the main network cost, but it is what makes keyed state and per-user windows possible.
package com.bank.delivery;
public class RedistributingExchange {
public static void main(String[] args) {
System.out.println("keyBy(user_id):");
System.out.println(" hash('user-A') % maxParallelism -> subtask 3");
System.out.println(" hash('user-B') % maxParallelism -> subtask 7");
System.out.println(" hash('user-C') % maxParallelism -> subtask 1");
System.out.println();
System.out.println("Redistributing: network shuffle.");
System.out.println("Records cross the network to reach");
System.out.println("the subtask owning their key group.");
System.out.println("This is the expensive step.");
}
}
keyBy(user_id):
hash('user-A') % maxParallelism -> subtask 3
hash('user-B') % maxParallelism -> subtask 7
hash('user-C') % maxParallelism -> subtask 1
Redistributing: network shuffle.
Records cross the network to reach
the subtask owning their key group.
This is the expensive step.
How does the tumbling window aggregate?
A 30-minute tumbling window groups each user's events into fixed, non-overlapping buckets by event time. An aggregate function computes per-user stats like event count, completed deliveries, and average scan interval. The window fires when the watermark passes the window's end.
package com.bank.delivery;
public class TumblingWindowAggregate {
public static void main(String[] args) {
System.out.println("Tumbling window: 30 minutes");
System.out.println(" [09:00-09:30) [09:30-10:00) ...");
System.out.println();
System.out.println("Per-user aggregate at window close:");
System.out.println(" user_id: user-A");
System.out.println(" window_start: 09:00");
System.out.println(" event_count: 8");
System.out.println(" completed: 3");
System.out.println(" avg_scan_interval_sec: 120");
}
}
Tumbling window: 30 minutes
[09:00-09:30) [09:30-10:00) ...
Per-user aggregate at window close:
user_id: user-A
window_start: 09:00
event_count: 8
completed: 3
avg_scan_interval_sec: 120
How does the JDBC sink work?
The JDBC sink writes one row per user per window to PostgreSQL. You configure batching and set sink parallelism lower than source parallelism to respect database connection limits. The sink uses an upsert keyed by user and window start for idempotency.
-- JDBC sink: upsert per user per window
INSERT INTO delivery_stats
(user_id, window_start, event_count, completed, avg_interval)
VALUES (?, ?, ?, ?, ?)
ON CONFLICT (user_id, window_start)
DO UPDATE SET event_count = EXCLUDED.event_count,
completed = EXCLUDED.completed,
avg_interval = EXCLUDED.avg_interval;
package com.bank.delivery;
public class JdbcSinkDesign {
public static void main(String[] args) {
System.out.println("JDBC sink config:");
System.out.println(" Parallelism: 4 (lower than source 12)");
System.out.println(" Batch size: 500 rows per flush");
System.out.println(" Upsert key: (user_id, window_start)");
System.out.println(" Idempotent: replay-safe");
}
}
JDBC sink config:
Parallelism: 4 (lower than source 12)
Batch size: 500 rows per flush
Upsert key: (user_id, window_start)
Idempotent: replay-safe
How do parallelism settings work?
Parallelism is set per operator. Match source parallelism to partitions, give the keyed window enough parallelism to spread user keys, and keep the sink parallelism lower to respect database connection pools. Flink chains forwarding operators into tasks for efficiency.
package com.bank.delivery;
public class ParallelismSettings {
public static void main(String[] args) {
System.out.println("Operator parallelism:");
System.out.println(" Source (Kafka): 12 (= partitions)");
System.out.println(" Map (enrich): 12 (forwarded/chained)");
System.out.println(" keyBy + Window: 12 (spread keys)");
System.out.println(" JDBC Sink: 4 (connection limit)");
System.out.println();
System.out.println("Task slots needed: 12");
System.out.println(" Source+Map chained in 12 slots");
System.out.println(" Window in 12 slots (after shuffle)");
System.out.println(" Sink in 4 slots");
}
}
Operator parallelism:
Source (Kafka): 12 (= partitions)
Map (enrich): 12 (forwarded/chained)
keyBy + Window: 12 (spread keys)
JDBC Sink: 4 (connection limit)
Task slots needed: 12
Source+Map chained in 12 slots
Window in 12 slots (after shuffle)
Sink in 4 slots
How does checkpoint-based fault tolerance work?
Flink injects barriers into the stream. When a barrier reaches an operator, it snapshots state (Kafka offsets at the source, open window contents at the window) to durable storage. On failure, Flink restores the last checkpoint, resets Kafka to saved offsets, and replays. With an idempotent upsert sink, exactly-once is achieved.
package com.bank.delivery;
public class CheckpointFaultTolerance {
public static void main(String[] args) {
System.out.println("Checkpoint flow:");
System.out.println(" 1. Barrier injected at source");
System.out.println(" 2. Flows through map (forwarded)");
System.out.println(" 3. Crosses keyBy (redistributed)");
System.out.println(" 4. Window snapshots open windows");
System.out.println(" 5. Sink acknowledges");
System.out.println(" 6. Checkpoint complete on S3/HDFS");
System.out.println();
System.out.println("On failure:");
System.out.println(" Restore checkpoint -> replay Kafka");
System.out.println(" -> window state correct");
System.out.println(" -> upsert sink: no duplicates");
System.out.println(" -> exactly-once end-to-end");
}
}
Checkpoint flow:
1. Barrier injected at source
2. Flows through map (forwarded)
3. Crosses keyBy (redistributed)
4. Window snapshots open windows
5. Sink acknowledges
6. Checkpoint complete on S3/HDFS
On failure:
Restore checkpoint -> replay Kafka
-> window state correct
-> upsert sink: no duplicates
-> exactly-once end-to-end
What are common interview follow-up questions?
- How do you handle a stalled Kafka partition that holds back watermarks?
- What is the difference between forwarding and hash exchange in Flink?
- How do you size the checkpoint interval against state size?
- When would you use incremental checkpoints with RocksDB?
- How do you test a Flink pipeline end to end?
Visual Overview
Flink Delivery Tracking Pipeline
────────────────────────────────
Kafka Source (12 partitions) Map (enrich) keyBy(user_id)
┌──────────────────┐ FORWARD ┌──────────────┐ REDISTRIBUTE ┌──────────────┐
│ subtask 0: p0 │─────────▶│ subtask 0 │──────────────▶│ subtask 0 │
│ subtask 1: p1 │─────────▶│ subtask 1 │──────────────▶│ subtask 1 │
│ ... │─────────▶│ ... │──────────────▶│ ... │
│ subtask 11: p11 │─────────▶│ subtask 11 │──────────────▶│ subtask 11 │
└──────────────────┘ └──────────────┘ └──────┬───────┘
(chained: one task) │
│ Tumbling
│ Window
│ (30 min)
▼
┌──────────────┐
│ Aggregate │
│ per user │
│ per window │
└──────┬───────┘
│
▼
┌──────────────┐
│ JDBC Sink │
│ (PostgreSQL) │
│ parallelism=4│
│ upsert key: │
│ (user,window)│
└──────────────┘
Key Takeaway
The Flink delivery-tracking pipeline has five stages and three design concerns: parallelism, forwarding versus redistributing, and fault tolerance. The source reads Kafka partitioned by user id with parallelism matching the partition count. The stateless map enrichment uses a forwarding exchange, so subtask n feeds subtask n with no network shuffle, and the two chain into one task. The keyBy on user id is a redistributing exchange that shuffles records over the network to the subtask owning each key group, which is what makes per-user keyed state and windows possible. The 30-minute tumbling window buckets events by event time and aggregates per-user stats, firing when the watermark passes the window end. The JDBC sink writes one row per user per window to PostgreSQL with batching and limited parallelism.
The interviewer probes forwarding versus redistributing, parallelism matching, and exactly-once. Forwarding is cheap and local, used when partitioning is preserved. Redistributing is the network shuffle caused by keyBy, and it is the main network cost. Match source parallelism to partitions, give the window enough parallelism for the key spread, and keep the sink parallelism lower for the database.
Fault tolerance comes from checkpointing: Flink injects barriers that flow through the dataflow, each operator snapshots its state to durable storage, and on failure Flink restores the last checkpoint, rewinds Kafka, and replays. With an upsert JDBC sink keyed by user and window start, exactly-once is achieved end to end.
In a banking context at the National Bank of India, this was exactly how card delivery tracking worked, with aligned parallelism, partition-idleness detection, forwarding for enrichment, the keyBy shuffle for per-user aggregation, and RocksDB incremental checkpoints for large window state.
Remember one sentence: forward stateless enrichment cheaply, redistribute via keyBy for per-key windows, match parallelism to partitions, and checkpoint for exactly-once fault tolerance.
---