Watermarks in Apache Flink: Handling Late and Out-of-Order Events

In an ideal world, events would arrive at the stream processor in the exact order they occurred. In reality, distributed networks, cellular handoffs, multi-partition Kafka topics, and server GC pauses cause events to arrive out of order and with arbitrary lateness.

To compute correct time-window aggregations without waiting indefinitely, stream engines like Apache Flink and Kafka Streams use Watermarks.

---

1. What is a Watermark?

A Watermark is a special control element that flows through the data stream alongside regular data records. It acts as a temporal progress metric:

A Watermark with timestamp $W(t)$ is a formal assertion by the stream engine that all records with an Event Time $t_{\text{event}}\le W(t)$ have already been observed.

Key Properties of Watermarks:

1. Monotonically Increasing: Watermarks only move forward in time. A watermark timestamp never decreases.
2. Trigger for Window Evaluation: When a watermark $W(t)\ge T_{\text{window\_end}}$ reaches an operator, the engine knows the window is complete. It executes the aggregation function (e.g. SUM or COUNT), emits the output downstream, and purges the window state from memory.

---

2. Bounded-Out-of-Orderness Watermarks

The most widely used watermark generation strategy is Bounded Out-Of-Orderness, which allows a configurable maximum delay $\Delta$:

$W(t)={\max }_i(t_{\text{event\_seen}})-\Delta$

---

3. Handling Extremely Late Data: Allowed Lateness and Side Outputs

What happens if an event arrives after the watermark has already advanced past its window boundary? (e.g. an event with $t=00:08$ arrives when $W=00:11$).

Flink provides three tiers of late data handling:

1. Default (Zero Lateness): Late events arriving after the watermark closes the window are silently dropped.
2. allowedLateness(Duration): Keeps the window state in memory (RocksDB) for an additional grace period. If late events arrive during this window, Flink updates the running total and emits an updated result.
3. sideOutputLateData(OutputTag): Diverts unprocessable late events to a secondary Kafka topic (Dead-Letter Stream) for offline audit and reconciliation without dropping data.

---

4. The Idle Partition Pathology & How to Fix It

A subtle but catastrophic production failure occurs in multi-partition topics when traffic is uneven:

The Root Cause:

The unified watermark across multiple partitions is the minimum watermark of all inputs: $W_{\text{unified}}=\min (W_{P_0},W_{P_1},\dots ,W_{P_n})$ If a single partition sits idle without new records, its watermark does not advance. Consequently, the entire downstream streaming application hangs indefinitely, producing zero output!

The Solution: withIdleness()

Configure an idleness timeout (e.g. Duration.ofMinutes(1)). If a partition receives no records for 1 minute, Flink temporarily ignores it when computing $W_{\text{unified}}$, allowing downstream windows to evaluate normally.

---

5. Production Implementation (Java & Go)

A. Java (Apache Flink with Side Outputs and Idleness)

---

6. Summary Checklist

• Always set a realistic bounded out-of-orderness threshold ($\Delta =5\text{s}-30\text{s}$) based on actual network latency distributions.
• Always configure withIdleness() on Kafka sources to prevent silent application stalls on low-traffic partitions.
• Attach sideOutputLateData() to mission-critical financial streams to audit and capture delayed records without dropping data.