Definition
- Late event – A record whose event-time timestamp falls into a window that has already fired because the watermark has moved past that window's end.
- Default behavior – Flink silently drops late events, which is efficient but loses data without any error or notification.
- Allowed lateness – A per-window grace period during which the window state is retained and each late event triggers a re-computation and re-emission of the window result.
- Side output – A mechanism to route events that arrive later than the allowed-lateness period to a separate typed stream for special handling such as reconciliation.
- Re-emission – During allowed lateness, the window function fires again for each late arrival, so downstream sinks must handle duplicate or corrected results.
The National Bank of India closes the books for the 10:00-to-10:01 minute once the manager announces that minute is over. If Rajesh's transaction that happened at 10:00:30 arrives a little after closing, the bank has three choices. By default, the clerk throws the slip away. With allowed lateness, the bank keeps that minute's ledger page open for an extra two minutes and, if the slip arrives within that grace period, reopens the page and reprints the corrected total. If the slip arrives even later, the bank drops it into a special late arrivals tray (side output) so a human can reconcile it manually.
Understanding the Concepts
What is a late event in Flink?
A window fires when the watermark crosses its end time. Any event whose timestamp belongs to that window but arrives after the watermark has passed is late. This happens frequently in real systems where mobile networks, retries, or batch uploads introduce delays.
package com.bank.streaming;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
public class LateEventDemo {
// Window [10:00 - 10:01) fires when watermark >= 10:01:00
// A transaction with POS time 10:00:30 arriving after
// watermark 10:01:00 is LATE
// stream.keyBy(Transaction::getAccountId)
// .window(TumblingEventTimeWindows.of(Time.minutes(1)))
// .sum("amount")
// Late events are silently dropped by default
}
Timeline:
10:00:00 Window [10:00 - 10:01) opens
10:00:30 Transaction T1 arrives (in window, processed)
10:00:45 Transaction T2 arrives (in window, processed)
10:01:00 Watermark reaches 10:01 --> window FIRES
10:01:15 Transaction T3 (POS time 10:00:35) arrives
T3 is LATE --> dropped by default
What is the default behavior for late events?
By default Flink silently drops late events without any error or notification. The window's state is purged after firing, so there is nowhere to put the late record. This is efficient but loses data, which is unacceptable for correctness-sensitive financial workloads.
package com.bank.streaming;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
public class DefaultDropDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.addSource(new KafkaSource<>("nbi-transactions"))
.assignTimestampsAndWatermarks(nbiWatermarkStrategy())
.keyBy(Transaction::getAccountId)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.sum("amount") // late events SILENTLY DROPPED
.print("Settlement");
// Late transaction T3 at 10:00:35 arriving at 10:01:15
// is dropped. Settlement total is WRONG.
env.execute("NBI Default Drop Demo");
}
}
How does allowed lateness work?
With allowed lateness, the window does not purge its state immediately after firing. It keeps the state for the extra lateness period. During that grace period, every late event triggers the window function again and emits an updated result.
package com.bank.streaming;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
public class AllowedLatenessDemo {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
env.addSource(new KafkaSource<>("nbi-transactions"))
.assignTimestampsAndWatermarks(nbiWatermarkStrategy())
.keyBy(Transaction::getAccountId)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.minutes(2)) // keep state 2 min extra
.sum("amount")
.print("Settlement");
// Window [10:00-10:01) fires at watermark 10:01:00
// Output: ACC-001 = Rs80,000 (first result)
// Late T3 arrives at 10:01:15 (within 2-min grace)
// Window re-fires: ACC-001 = Rs85,000 (corrected result)
env.execute("NBI Allowed Lateness Demo");
}
}
Output timeline:
[10:01:00] Settlement> ACC-001 = Rs80,000 (first fire)
[10:01:15] Settlement> ACC-001 = Rs85,000 (re-fire with late T3)
// Downstream sink sees TWO results for the same window!
// Sink must be idempotent or upsert-capable
How do side outputs capture very late events?
Events later than the allowed-lateness period are routed to a side output rather than being dropped. You define an OutputTag and attach it to the window, then retrieve the side stream for separate processing.
package com.bank.streaming;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;
public class SideOutputLateDemo {
// Tag for very-late events
private static final OutputTag<Transaction> LATE_TAG =
new OutputTag<Transaction>("late-transactions") {};
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
SingleOutputStreamOperator<String> mainStream =
env.addSource(new KafkaSource<>("nbi-transactions"))
.assignTimestampsAndWatermarks(nbiWatermarkStrategy())
.keyBy(Transaction::getAccountId)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.minutes(2))
.sideOutputLateData(LATE_TAG) // route very-late events
.sum("amount");
// Main output: normal + slightly-late corrected results
mainStream.addSink(new SettlementSink());
// Side output: very-late events for reconciliation
DataStream<Transaction> lateStream =
mainStream.getSideOutput(LATE_TAG);
lateStream.addSink(new ReconciliationSink());
env.execute("NBI Side Output Late Demo");
}
}
How do allowed lateness and side outputs work together?
The two tools combine in layers. A modest allowed lateness handles the common slightly-late case automatically with recomputation. A side output catches the rare very-late case for human or batch handling. This gives both low latency and no silent data loss.
Layered late-event handling:
Event arrives Within window? Within allowed Side output?
lateness?
────────────── ────────────── ────────────── ──────────
Before watermark YES: processed N/A N/A
After watermark, NO YES: re-fire N/A
within grace window
After watermark, NO NO YES: routed
beyond grace to side output
No side output NO NO DROPPED
configured
// Complete layered setup for NBI settlement
stream
.keyBy(Transaction::getAccountId)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.minutes(2)) // layer 1: recompute
.sideOutputLateData(LATE_TAG) // layer 2: capture
.sum("amount");
What are common mistakes with late event handling?
The most common mistake is not making the downstream sink idempotent when using allowed lateness. The window re-fires and emits a second result for the same window, so an append-only sink produces duplicates. Another mistake is setting allowed lateness too high, which keeps window state in memory for a long time and increases checkpoint size.
// WRONG: append-only sink with allowed lateness
// Window fires twice for the same key+window
// Sink appends both: Rs80,000 AND Rs85,000 --> double-counted!
stream.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.minutes(2))
.sum("amount")
.addSink(new AppendOnlyJdbcSink()); // WRONG: appends duplicates
// CORRECT: upsert sink that overwrites the previous result
stream.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.minutes(2))
.sum("amount")
.addSink(new UpsertJdbcSink()); // CORRECT: overwrites with correction
When should you configure allowed lateness?
Configure allowed lateness when you cannot afford to lose slightly-late data and your downstream sink can handle re-emissions. The right value is a business decision: how long do you expect common stragglers to arrive, balanced against how long you want to keep window state. For a banking settlement report, two minutes of allowed lateness covers mobile retries without excessive state retention.
package com.bank.streaming;
public class LatenessDecisionGuide {
// Decision guide for NBI pipeline:
//
// Settlement report (must be accurate):
// allowedLateness = 2 minutes (covers mobile retries)
// sideOutput = yes (reconciliation for hour-late merchants)
// sink = upsert (handles re-emissions)
//
// Ops dashboard (approximate is fine):
// allowedLateness = 0 (default, drop late events)
// sideOutput = no
// sink = append (no corrections needed)
//
// Regulatory audit feed:
// allowedLateness = 5 minutes (strict compliance)
// sideOutput = yes (manual review for extreme latecomers)
// sink = upsert with audit trail
}
What should a senior engineer know about late events?
A senior engineer should know that allowed lateness keeps window state in memory, increasing checkpoint size proportionally, that the re-emission pattern requires the entire downstream pipeline to be correction-aware, and that for extreme lateness (hours or days) a batch reconciliation job is more practical than keeping windows open.
package com.bank.streaming;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
public class SeniorLateEventConfig {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env =
StreamExecutionEnvironment.getExecutionEnvironment();
// Senior-level: monitor late event counts via metrics
// Flink exposes numLateRecordsDropped per operator
// Alert if late drops exceed threshold
env.addSource(new KafkaSource<>("nbi-transactions"))
.assignTimestampsAndWatermarks(nbiWatermarkStrategy())
.keyBy(Transaction::getAccountId)
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.minutes(2))
.sideOutputLateData(LATE_TAG)
.aggregate(new SettlementAggregator())
.addSink(new UpsertSettlementSink());
// Side output: very-late events go to reconciliation
// Batch job runs nightly to settle any remaining gaps
env.execute("NBI Senior Late Event Config");
}
}
Visual Overview
Late Event Handling in Flink
Timeline for Window [10:00 - 10:01):
10:00:00 ──▶ Window opens
10:00:30 ──▶ T1 arrives (normal, processed)
10:00:45 ──▶ T2 arrives (normal, processed)
10:01:00 ──▶ Watermark passes ──▶ Window FIRES (first result)
─ ─ ─ ─ ─ allowed lateness period (2 min) ─ ─ ─ ─ ─
10:01:15 ──▶ T3 arrives (late but within grace)
Window RE-FIRES (corrected result)
10:02:30 ──▶ T4 arrives (late but within grace)
Window RE-FIRES (corrected result)
10:03:00 ──▶ Watermark passes end + lateness
Window state PURGED
─ ─ ─ ─ ─ beyond allowed lateness ─ ─ ─ ─ ─ ─ ─ ─
10:05:00 ──▶ T5 arrives (very late, beyond grace)
Routed to SIDE OUTPUT for reconciliation
Layered Architecture:
┌──────────────┐ ┌──────────────────┐
│ Normal Events │────▶│ Window Function │──▶ Settlement Sink
└──────────────┘ │ (fires + re-fires│ (upsert)
┌──────────────┐ │ during grace) │
│ Slightly Late│────▶│ │
└──────────────┘ └──────────────────┘
┌──────────────┐ ┌──────────────────┐
│ Very Late │────▶│ Side Output │──▶ Reconciliation
└──────────────┘ └──────────────────┘ Topic
Key Takeaway
A late event in Flink is a record whose event-time timestamp belongs to a window that has already fired because the watermark has moved past that window's end. The default behavior is that Flink silently drops late events, which is efficient but means you lose data without any error. For correctness-sensitive work like banking settlements, that is unacceptable.
Flink gives you two layered tools to handle lateness. The first is allowed lateness, where you tell the window to keep its state for an extra grace period after it first fires. During that period, every late event triggers the window function again and re-emits an updated result, so the window corrects itself. The important consequence is that downstream consumers see the window result more than once, so the sink must be idempotent or upsert-capable. The second tool is side outputs: events that arrive later than the allowed-lateness period are routed to a separate stream for reconciliation rather than being dropped.
At the National Bank of India, our per-account per-minute totals needed to be accurate for the fraud engine. Most transactions arrived within seconds, but mobile retries meant a small fraction came a little late, so we set a two-minute allowed lateness and made the downstream sink an upsert. For the rare transaction that arrived an hour late because a merchant terminal had been offline, we attached a side output that routed those records to a reconciliation topic.
That layered approach, default drop replaced by a tuned allowed lateness for the common case and a side output for the extreme case, gives both low latency and no silent data loss.
Remember one sentence: Late events are dropped by default, kept and recomputed during allowed lateness, and captured in a side output for reconciliation beyond the grace period.
---