Allowed Lateness and Side Outputs

Analogy categoryAnalogies

Why Windows Need a Grace Period

A window normally fires and its state is purged the moment the watermark passes the window's end timestamp, since the watermark is a declaration that no earlier data remains in flight. In practice, though, that declaration is a heuristic bound, not an absolute guarantee — a mobile device can reconnect after being offline, a batched upstream system can flush a backlog, or a straggling network path can deliver a record minutes after the watermark already advanced past it. allowedLateness(Time) tells Flink to keep a window's state around for an additional grace period after the watermark passes its end, so that late-arriving elements within that grace period still trigger a window recomputation instead of being silently dropped.

🏏

Cricket analogy: Allowed lateness is like an official scorer keeping an over's tally open for a short grace period even after declaring it complete, in case a delayed DRS review overturns a decision and the runs need to be recounted before the scorecard is truly locked.

Firing Behavior During the Grace Period

With allowedLateness(Time.minutes(2)) configured, a window still fires its first result the moment the watermark crosses its end boundary, exactly as it would without lateness handling — the grace period doesn't delay the initial output. What changes is that the window's state and per-window timers are retained instead of being immediately purged; each subsequent late element that arrives before the grace period expires triggers the window function to fire again with an updated result, producing what the default trigger calls a 'late firing.' Only after the watermark advances past window-end-plus-allowed-lateness does Flink finally purge the window's state, and downstream consumers must be prepared to handle multiple result records for the same window rather than assuming exactly one output per window.

🏏

Cricket analogy: This is like a broadcaster publishing the over's final score the moment the over ends, but still quietly issuing a corrected graphic if a DRS overturn lands within the next two minutes, rather than holding the initial graphic back to wait for every possible review.

Capturing Truly Late Data with Side Outputs

Once an element arrives after even the allowed lateness grace period has expired — meaning the watermark has already advanced past window-end-plus-lateness and the window's state has been purged — Flink drops that element from the windowed computation by default, silently. To avoid losing this data entirely, you attach an OutputTag<T> to the windowed stream via sideOutputLateData(lateOutputTag), which redirects any element that misses even the grace period into a separate DataStream retrievable from the main result via getSideOutput(lateOutputTag). This side stream is commonly written to a 'late records' sink (a dead-letter topic, a raw archive table, or a correction pipeline) so that truly late data is preserved for auditing or manual reconciliation instead of vanishing without a trace.

🏏

Cricket analogy: Side-outputting late data is like routing a DRS review that arrives long after the match has been officially archived into a separate 'post-match corrections' ledger, rather than letting it silently vanish and leave the historical scorecard subtly wrong forever.

java

OutputTag<UserEvent> lateTag = new OutputTag<UserEvent>("late-events") {};

SingleOutputStreamOperator<WindowResult> result = events
    .keyBy(UserEvent::getUserId)
    .window(TumblingEventTimeWindows.of(Time.minutes(10)))
    .allowedLateness(Time.minutes(2))
    .sideOutputLateData(lateTag)
    .aggregate(new CountAggregate());

// Main stream: window results, including any late-firing corrections
result.print();

// Side stream: elements that missed even the 2-minute grace period
DataStream<UserEvent> lateEvents = result.getSideOutput(lateTag);
lateEvents.sinkTo(deadLetterSink);

allowedLateness only applies to event-time windows; processing-time windows fire once and purge their state immediately, since there is no notion of 'late' data when time is defined by the local clock.

Setting allowedLateness too generously (e.g., hours) keeps window state alive far longer, multiplying state backend size and checkpoint duration — always pair a nonzero allowed lateness with a realistic upper bound informed by your actual data's tail latency.

allowedLateness(Time) keeps a fired window's state alive for a grace period, allowing late elements to trigger recomputation.
The window still fires its first result exactly when the watermark crosses window-end; the grace period only affects subsequent late firings.
Downstream consumers must handle multiple result records per window when allowed lateness is configured.
Elements arriving after the allowed lateness grace period expires are dropped by default, silently.
sideOutputLateData(OutputTag) redirects truly late elements to a separate retrievable DataStream instead of dropping them.
allowedLateness only applies to event-time windows, not processing-time windows.
Overly generous allowed lateness values inflate state size and checkpoint duration, so tune it to real tail latency.