Phase 2: Time Semantics, Watermarks & Windowing Mechanics¶

Audience: Principal Distributed Systems Architect & Stream Processing Engineer
Source References: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/watermark/, flink-streaming-java/src/main/java/org/apache/flink/streaming/api/windowing/

1. The Three Time Semantics¶

In real-time distributed stream processing, the physical arrival time of an event at a server rarely corresponds to the time the event actually occurred in the physical world.

┌────────────────────────────────────────────────────────────────────────┐
│ 1. Event Time                                                          │
├────────────────────────────────────────────────────────────────────────┤
│ - The timestamp embedded inside the event payload by the originating   │
│   device (e.g. smartphone sensor, IoT device, POS terminal).          │
│ - Completely deterministic: Replaying historical events yields the     │
│   exact same calculation regardless of processing speed!               │
└────────────────────────────────────────────────────────────────────────┘

┌────────────────────────────────────────────────────────────────────────┐
│ 2. Ingestion Time                                                      │
├────────────────────────────────────────────────────────────────────────┤
│ - The timestamp assigned to the event when it enters Flink's source    │
│   operator. Independent of device clocks.                              │
└────────────────────────────────────────────────────────────────────────┘

┌────────────────────────────────────────────────────────────────────────┐
│ 3. Processing Time                                                     │
├────────────────────────────────────────────────────────────────────────┤
│ - The physical system clock of the TaskManager machine executing the   │
│   transformation. Non-deterministic; varies with network jitter.      │
└────────────────────────────────────────────────────────────────────────┘

2. Watermarks: The Mathematical Assertion of Time¶

When processing out-of-order event streams across distributed networks, how does an operator know when all events for a given time window have arrived?

A Watermark is an in-band control record carrying timestamp $T$.
It represents the mathematical assertion:
$$\text{Assertion: No future records with timestamp } t < T \text{ will arrive.}$$

Out-of-Order Stream with Watermarks:
[Event: 09:01] ──► [Event: 09:03] ──► [Event: 09:02] ──► [Watermark: 09:02] ──► [Event: 09:04]
                                                               │
                                                               ▼
                                       Windows ending at or before 09:02
                                       are guaranteed COMPLETE and trigger!

2.1 Bounded-Out-Of-Orderness Watermark Generator¶

WatermarkStrategy<TransactionEvent> strategy = WatermarkStrategy
    .<TransactionEvent>forBoundedOutOfOrderness(Duration.ofSeconds(10))
    .withTimestampAssigner((event, timestamp) -> event.getEventTimestamp())
    .withIdleness(Duration.ofMinutes(1)); // Prevents idle partition deadlock!

Tracks maximum observed timestamp: $T_{\max} = \max(T_{\max}, t_{\text{event}})$.
Emits watermark: $W = T_{\max} - 10\text{ seconds}$.
Idleness Invariant: If one Kafka partition stops receiving data, its watermark stops advancing. Because downstream operators track the minimum watermark across all input channels, a single idle partition freezes the entire application! Calling .withIdleness() marks stalled partitions idle so the engine ignores them when computing downstream watermarks.

3. Window Cellular Lifecycle: Tumbling, Sliding & Session¶

Window Types:
Tumbling (Fixed, Non-overlapping):
[ 09:00 - 09:10 ) [ 09:10 - 09:20 ) [ 09:20 - 09:30 )

Sliding (Overlapping Intervals):
[ 09:00 --------- 09:10 )
       [ 09:05 --------- 09:15 )
              [ 09:10 --------- 09:20 )

Session (Inactivity Gap Driven):
[ Event A, Event B ] ──(Inactivity > 5 min)──► Window Closes!

3.1 Anatomical Components of a Window Operator¶

A window operator in Flink coordinates three distinct cellular hooks: 1. WindowAssigner: Maps an incoming record to one or more logical windows (e.g., TumblingEventTimeWindows.of(Time.minutes(5))). 2. Trigger: Determines when the window computation should execute (default: triggers when watermark $\ge \text{window.maxTimestamp()}$). 3. Evictor: Optional modifier that prunes elements from the window before or after the trigger fires.

4. Late Data Handling Strategies¶

What happens when an event arrives with timestamp $t < \text{current Watermark}$? By default, Flink drops late records silently. In production financial pipelines, dropping late data is unacceptable.

OutputTag<TransactionEvent> lateDataTag = new OutputTag<TransactionEvent>("late-transactions"){};

SingleOutputStreamOperator<AggregatedResult> windowedStream = sourceStream
    .keyBy(TransactionEvent::getAccountId)
    .window(TumblingEventTimeWindows.of(Time.minutes(10)))
    .allowedLateness(Time.minutes(2)) // Keep window state alive for 2 extra minutes!
    .sideOutputLateData(lateDataTag)  // Spill critically late events to side output!
    .aggregate(new TransactionSumAggregate());

// Route critically late events to dead-letter queue or alerting topic
DataStream<TransactionEvent> lateEvents = windowedStream.getSideOutput(lateDataTag);

Late Data Processing Flow:
1. Event arrives before Watermark: Ingested into active window state.
2. Event arrives after Watermark but within allowedLateness:
   - Window re-triggers with updated computation.
3. Event arrives after allowedLateness:
   - Emitted to Side Output dead-letter stream for audit logging. Zero data loss!