Late Data Is Dropped by Default, and Nothing Tells You

A watermark sounds like a mechanism that handles late data. It is closer to a deadline you picked. Flink's own definition says a watermark declares "there should be no more elements" before that time. The docs then admit some elements violate it anyway.

When they do, the default is not an error or a retry. Flink discards the record.

Key takeaways#

• A watermark asserts what should arrive, and the docs concede records can violate it.
• allowedLateness defaults to 0, so records behind the watermark are discarded.
• Nothing throws. The window already fired and downstream already saw a number.
• The out-of-orderness bound is a constant you chose, so your loss rate is a config value.
• An idle partition holds the whole pipeline's watermark back until you set withIdleness.

What do watermarks actually promise?#

They promise an assertion about progress, not completeness. A watermark carries a timestamp saying event time has reached that point, which lets windows decide they can close and emit. It is the signal that unblocks output. It is not a check that every record for that period has actually turned up.

Flink states the definition precisely, and one word in it carries the weight.

The word the definition turns on#

Here is the definition from the docs, with the qualifier intact.

A Watermark(t) declares that event time has reached time t in that stream, meaning that there should be no more elements from the stream with a timestamp t' <= t.

"Should be", not "will be". A watermark is a claim the source makes about its own future behaviour, and claims can be wrong.

The docs say so directly in the next breath.

It is possible that certain elements will violate the watermark condition

So the system's own documentation tells you the invariant is breakable. Everything else in this article follows from that sentence.

What happens to late data that arrives anyway#

This is the part worth knowing before you ship an event-time pipeline.

By default, the allowed lateness is set to 0. That is, elements that arrive behind the watermark will be dropped.

And again, for windows specifically:

By default, late elements are dropped when the watermark is past the end of the window.

Read that as a data-loss default. Records that reached your system, parsed correctly and belonged to a real window get discarded because they missed a deadline.

Flink gives you a way to keep them, and it is opt-in. You call sideOutputLateData(OutputTag) on the windowed stream, then read that side output separately. Nothing enables it for you.

The bound is a number you picked#

Where does the deadline come from? Usually from one line of your own code.

The common strategy is forBoundedOutOfOrderness, and the docs show it as forBoundedOutOfOrderness(Duration.ofSeconds(20)). The generator then works exactly as you would guess, emitting "the watermark as current highest timestamp minus the out-of-orderness bound".

So the watermark trails the newest event you have seen by a fixed gap. Pick 20 seconds and you tolerate 20 seconds of reordering. Anything later is late data.

That makes the tradeoff explicit and yours. A larger bound waits longer, so windows fire later and dashboards lag. A smaller bound fires sooner and discards more. There is no setting that gives you both, and the default for the discarded records is to forget them.

A worked example#

Take a mobile analytics pipeline. Tumbling one-minute windows, forBoundedOutOfOrderness(20s), and the default allowedLateness of zero.

A phone goes into a tunnel at 12:01 and reconnects at 12:04. The client flushes the events it buffered, all stamped with their true event times around 12:01.

Those events are roughly 180 seconds late against a 20-second bound. The 12:01 window closed nearly three minutes ago, and its count already went to your dashboard. Flink discards the buffered events on arrival.

Now size it. Suppose half a percent of your events come from clients that buffer longer than the bound, which is unremarkable for mobile. At 100 million events a day that is 500,000 records a day deleted without a log line.

The number that reaches your dashboard is not wrong by a random amount either. It skews one way. The events you lose come from poor connectivity, so you systematically undercount your worst-served users.

A table for choosing the bound#

The right bound depends on why your data is late, so here is how I would choose.

Source	Bound	Reasoning
Single ordered partition, one datacentre	Near zero, or monotonous timestamps	Reordering barely happens
Kafka topic, several partitions	Seconds	Interleaving between partitions only
Mobile or IoT clients that buffer offline	Minutes, plus a side output	Lateness has no natural ceiling
Billing or regulatory counts	Any bound, always a side output	You must account for every record
Live dashboard, freshness first	Small bound, accept the loss	A late correction is worse than a stale tile

Row three is the one that catches teams. Offline buffering has no natural ceiling, so no bound is large enough. Capture the stragglers instead of widening the window forever.

Row four is a stance. Where you have to account for every record, a side output stops being optional. "We dropped some" does not satisfy an auditor.

What allowedLateness costs#

Raising allowedLateness is the obvious fix, so it deserves its price tag. There are two costs, and the second surprises people.

The first is state. The docs are direct about the mechanism.

In order to make this work, Flink keeps the state of windows until their allowed lateness expires. Once this happens, Flink removes the window and deletes its state.

So an hour of allowed lateness means every window stays resident for an hour past its end. Multiply that by your key cardinality and it becomes the dominant term in your state size, which shows up as checkpoint duration long before it shows up as a memory alert.

The second cost changes your sink contract.

Depending on the trigger used, a late but not dropped element may cause the window to fire again. This is the case for the EventTimeTrigger.

Read that carefully. The window emits a second result for the same window, with a corrected value. Downstream now receives two rows for one logical window.

That is fine if your sink upserts on a window key. It is silently wrong if your sink appends, because the corrected count lands beside the original and your totals double-count. Make the sink idempotent before you widen the bound, for the same reasons as exactly-once semantics in Kafka.

The idle partition that stalls everything#

A second failure runs the opposite way. Instead of losing data, the pipeline stops producing output at all.

The cause lies in how Flink computes a parallel watermark. If one input goes quiet, as the docs put it, "the WatermarkGenerator also does not get any new information", and then "the watermark will be held back, because it is computed as the minimum over all the different parallel watermarks".

One silent partition therefore freezes event time for the whole job. Windows never close, output stops, and every other partition is working perfectly.

The fix is withIdleness(), which marks a quiet input as idle so the remaining sources can advance the watermark. Set it on any source whose partitions can go empty. That covers most of them once traffic runs uneven across keys or regions.

The failure mode, and how to detect it#

Both failures are silent, and they need different alarms.

For late data, the good news is that Flink counts it. Watch the numLateRecordsDropped metric on your window operators, and treat any sustained non-zero value as data loss rather than noise. It is the one number that turns this from invisible to obvious.

Better still, route the stragglers somewhere. Attach sideOutputLateData and write that stream to cheap storage, even if nothing consumes it, because a side output turns a deletion into a decision you can revisit.

For the stalled pipeline, watch the watermark itself rather than throughput. Flink exposes currentInputWatermark, documented as "the last watermark this operator/tasks has received (in milliseconds)". Compare it against wall-clock time per operator and alert when the gap grows. Throughput looks healthy right up until you notice no window has fired for an hour.

One habit catches both. Replay a day of production events into a staging job and compare its output totals against the live job's. Any gap is the data you have been dropping, quantified.

Run that replay with a deliberately huge bound, an hour rather than twenty seconds, since the staging job has no latency budget to protect. The difference between the two totals is your real lateness distribution, measured rather than guessed.

Then set the production bound from that curve. Most teams find the tail is thinner than they feared, and a few find it is much fatter, and both answers beat the number someone typed in a year ago.

What batch processing gets right#

Batch has a real advantage here, and it is worth naming before recommending streams.

A batch job over a fixed window waits for wall-clock time to pass the end of that window, then reads whatever landed. It does not guess when the data is complete, it simply starts late enough that the question rarely arises. Schedulers rest on the same data interval idea.

Batch also reruns cleanly. Tomorrow's job picks up whatever arrived late, or a backfill of yesterday's does. The correction then costs one re-execution rather than a bespoke recovery path.

What it gives up is latency, obviously. You wait a whole interval for any answer at all. Streaming trades that wait for an estimate, and this article is about the size of the estimate's error, not an argument against making the trade.

Where this advice is wrong#

Most pipelines lose almost nothing. If your events come from servers in one region through ordered partitions, a few seconds of bound covers essentially all reordering, and the dropped count stays at zero for months. Do not build a lateness recovery path for a problem you can measure at zero.

Side outputs are not free. Every late record you keep is storage you pay for and a second code path someone must maintain, and reconciling it into already-published aggregates is genuinely hard. Sometimes accepting a known, measured loss is the correct engineering choice.

Widening the bound often makes things worse. Raising the bound from 20 seconds to five minutes delays every window for every user to rescue a small tail. Where lateness is heavy-tailed, a small bound plus a side output beats a large bound, because you keep both the latency and the records.

The short version#

A watermark asserts that late data should not arrive, and Flink's docs concede some will. When it does, allowedLateness of zero discards it with no error. The window has already fired, and your dashboard has already published a number that skews against your worst-connected users. The bound producing that outcome is a constant you chose. So watch numLateRecordsDropped, attach a side output before you need one, set withIdleness on sources that can go quiet, and measure the loss with a replay rather than assuming it is zero.

Sources: Flink: timely stream processing, Flink: windows and allowed lateness, Flink: generating watermarks