A tumbling one-minute window closes when the watermark passes its end, not when the clock does. Seven events stamped 10:00:58 that arrive at 10:01:12 have exactly two possible fates: they correct a result you already published, or they increment a counter nobody reads. Neither one throws an exception.
That is the whole bug. Late data is not an error condition, so nothing in your dashboards turns red while the number is wrong.
Why "late" has no error path
Take a pipeline that reports purchases per region in one-minute tumbling windows. Mobile clients buffer events while they are offline and flush a batch after a reconnect, so the order events arrive on the wire is only loosely related to the timestamps inside them.
The 10:00 window is emitted at 10:01:05. A batch lands at 10:01:12 carrying seven events stamped 10:00:58 and a few stamped 10:01:03.
If the job runs with the default allowed lateness (zero, in Flink), those seven events are gone. Not delayed, not retried, not written to a dead-letter queue: dropped, because the window's state was purged a second after it fired. The metric for 10:00 is now 7 purchases short, and a pipeline that drops 3% of events has the same 5xx rate as one that drops 0%.
That is why this drill is worth an evening: the failure mode is not a crash you can grep for.
Two knobs everyone conflates
Almost everyone I have run this with reaches for one number first: how much out-of-orderness to tolerate. In Flink's vocabulary there are two, and they do different jobs.
| Knob | Decides | Costs |
|---|---|---|
| Out-of-orderness | How far the watermark trails the newest timestamp seen | Reporting latency on every window, including the ones that were never late |
| Allowed lateness | How long window state survives after the first result goes out | State size: open windows × keys × retention |
Out-of-orderness sets when the first result goes out. Allowed lateness sets whether a late event becomes a correction or a drop. Raising the first does not remove drops. It moves the boundary and delays everything you report.
Flink's window lifecycle is explicit about the second knob: a window is "completely removed when the time (event or processing time) passes its end timestamp plus the user-specified allowed lateness". Their example: with 5-minute tumbling windows and 1 minute of allowed lateness, the window for 12:00–12:05 is removed when the watermark passes 12:06. Internally the condition is maxTimestamp + allowedLateness, where maxTimestamp is the window end minus one millisecond.
Beam calls the same knob withAllowedLateness; Kafka Streams calls it the grace period on a windowed aggregation. The name changes, the contract does not.
Build the aggregator
Roughly 55 lines, no dependencies, no cluster. outOfOrdernessMs and allowedLatenessMs are the two knobs above, onEmit is your sink, and onDrop is the metric you should have had all along.
function createWindowAggregator({
windowMs,
outOfOrdernessMs = 0,
allowedLatenessMs = 0,
onEmit = () => {},
onDrop = () => {},
}) {
let maxEventTs = -Infinity;
let watermark = -Infinity;
let pushed = 0;
let dropped = 0;
const windows = new Map(); // windowStart -> { fired, revision, keys: Map<key, total> }
const windowStartOf = (ts) => Math.floor(ts / windowMs) * windowMs;
function fire() {
const due = [...windows.entries()]
.filter(([start, w]) => !w.fired && start + windowMs <= watermark)
.sort((a, b) => a[0] - b[0]);
for (const [start, w] of due) {
w.fired = true;
for (const [key, total] of w.keys) {
onEmit({ windowStart: start, windowEnd: start + windowMs, key, total, revision: 0 });
}
}
}
function purge() {
// Flink removes a window when the watermark passes maxTimestamp + allowedLateness,
// where maxTimestamp is windowEnd - 1. For whole-millisecond windows that is windowEnd + lateness.
for (const [start] of windows) {
if (start + windowMs + allowedLatenessMs <= watermark) windows.delete(start);
}
}
return {
push(event) {
pushed += 1;
maxEventTs = Math.max(maxEventTs, event.ts); // monotonic: a skewed event cannot pull it back
watermark = maxEventTs - outOfOrdernessMs;
const start = windowStartOf(event.ts);
const expired = start + windowMs + allowedLatenessMs <= watermark;
if (expired) {
dropped += 1;
onDrop(event);
} else {
let w = windows.get(start);
if (!w) {
w = { fired: false, revision: 0, keys: new Map() };
windows.set(start, w);
}
w.keys.set(event.key, (w.keys.get(event.key) ?? 0) + event.value);
if (w.fired) {
// This window already went out. Same (window, key), new total: a correction.
w.revision += 1;
onEmit({ windowStart: start, windowEnd: start + windowMs, key: event.key, total: w.keys.get(event.key), revision: w.revision });
}
}
fire();
purge();
},
advanceTo(ts) {
// A source heartbeat: "nothing older than ts can still arrive". Idle partitions need this,
// otherwise the global watermark stops moving.
if (ts > maxEventTs) {
maxEventTs = ts;
watermark = maxEventTs - outOfOrdernessMs;
}
fire();
purge();
},
watermark: () => watermark,
stats: () => ({ pushed, dropped, openWindows: windows.size }),
};
}
Three choices in there are the ones an interviewer will poke at.
The watermark only moves forward. maxEventTs is a running maximum, so a skewed client clock cannot resurrect purged state.
Firing and purging are separate steps. A window can fire and still be open for corrections — that is what allowed lateness buys.
Every drop is reported. Not logged as an error, because it is not one: counted, so you can chart it and alert on it.
Five contracts to prove
A correct-looking aggregator and a correct one differ in the boundary cases, and those are the drill. Five assertions, with node:assert/strict:
1. A window fires once, when the watermark crosses its end. Two events land in window [0, 60000) and nothing fires yet. Then an event at ts: 70000, with a 5-second out-of-orderness bound, moves the watermark to 65000 — and exactly one emission comes out: { windowStart: 0, key: 'us', total: 7, revision: 0 }.
2. An event inside allowed lateness is a correction, not a new row. Add an event stamped 59,500 after that emission:
w.push({ key: 'us', ts: 59_500, value: 5 });
assert.equal(out.length, 2);
assert.deepEqual(out[1], {
windowStart: 0, windowEnd: 60_000, key: 'us', total: 12, revision: 1,
});
Same windowStart, same key, higher total, revision incremented. If your sink appends, that window now has two rows — 7 and 12 — and a downstream SUM reports 19 where the truth is 12.
3. An event past windowEnd + allowedLateness is dropped, never re-homed. This is the assertion that catches the bug people actually ship: the old window is missing from the state map, so the late event gets counted into the current one. Push an event far enough ahead to move the watermark past the purge point, then push one stamped 30,000 into that purged window. The dropped list grows by one, the emitted totals do not move, and nothing lands in the current window.
4. The watermark never goes backwards. With the same 5-second bound, push ts: 200000 (watermark 195,000), then ts: 1000. The watermark stays at 195,000 and the skewed event is dropped rather than reopening a purged window.
5. Conservation: nothing is unaccounted for. After any stream, the sum of the final totals you emitted plus the value of the events you dropped must equal the value you pushed in.
const finalTotals = (out) => {
const t = new Map();
for (const e of out) t.set(`${e.windowStart}:${e.key}`, e.total); // last emission per (window, key) wins
return t;
};
const counted = [...finalTotals(out).values()].reduce((a, b) => a + b, 0);
const lostValue = dropped.reduce((a, e) => a + e.value, 0);
assert.equal(counted + lostValue, pushedValue); // pushedValue = sum of every event.value you pushed
If that identity fails, you have a bug you cannot see from any dashboard.
The naive fix, measured
The tempting response to all of this is to raise the out-of-orderness bound until nothing is late. So I ran the same 500 events through the same code with one constant changed. Event timestamps are uniformly random across a 5-minute range, replayed in a slightly shuffled arrival order (a seeded PRNG, so these numbers reproduce exactly), then flushed with a heartbeat:
| Setting | Events dropped | Value never counted | Emissions before the flush |
|---|---|---|---|
| 5s out-of-orderness, 30s lateness | 384 of 500 | 1,874 of 2,429 (77%) | 4 |
| 600s out-of-orderness, 30s lateness | 0 | 0 | 0 |
Both runs satisfy the conservation identity. The second one looks perfect and costs you ten minutes of reporting delay on every window — and it still drops anything later than ten minutes. A bound is a promise about your input, not a property you can configure away. When the input does not honor the promise, the drop counter is the only thing standing between you and a number you cannot defend in a review.
What the interviewer asks next
The global watermark is a minimum, not a maximum
With multiple input partitions, the global watermark is the minimum of the per-partition watermarks. One partition whose newest event is five minutes old holds back every window downstream, no matter how far ahead the others are — and a partition that goes idle holds it back forever. That is why real jobs configure an idle timeout, and why the code above has advanceTo: a source has to be able to say "nothing older than this is coming" even when it has no events to send.
If you have only run single-partition jobs, this is the follow-up that separates "read the Flink docs" from "operated this".
How to pick the two numbers
Measure, do not guess. Take arrival_ts - event_ts per source from your ingestion log, read the p99, set allowed lateness around twice that, and alert on the drop counter. Pick the out-of-orderness bound from how much reporting delay the consumer tolerates — that is a product decision, not a data one.
The sink has to upsert
Corrections mean the same (windowStart, key) can be emitted more than once, so the sink has to upsert on that pair. Append-only inserts double-count after every correction, and delete-then-insert is not idempotent when the emit is at-least-once, which it is. "We dedupe later" only works if the dedupe key is that same pair — in which case you have built the upsert with extra steps.
Replay determinism
Because events are assigned to windows by event time and the watermark is a pure function of the maximum timestamp seen, replaying the same stream through the same code produces the same final totals. Processing-time windows do not have that property, which is why a backfill of last Tuesday disagrees with what the dashboard showed on Tuesday.
FAQ
Is the timestamp in the payload the event time?
Only if you trust the producer's clock. Client clocks drift, and a device set a day forward pushes the watermark past every open window. Validate skew at ingestion and prefer a broker timestamp when the client cannot be trusted.
Does a large allowed lateness cost anything beyond memory?
State lifetime. Every retained window keeps its keys and accumulators until the watermark passes windowEnd + allowedLateness, so lateness multiplies key space by retention. It also extends how long a correction can arrive, which downstream consumers have to be ready for either way.
What if I cannot change the pipeline?
Recompute the affected windows from the raw events — for a bounded window, a batch correction is cheaper than a streaming rewrite — publish it alongside the streaming number, and label which one is provisional. The mistake is publishing one number and letting the drop counter be the only place the difference shows up.
Where to rehearse the follow-ups
Those two questions arrive as a conversation, not as code, and that is the part a static list of questions cannot rehearse. If you want a live partner for them, aceround.app — AI interview assistant — runs mock sessions where the interviewer follows up on what you just said instead of reading the next item off a list, and its data analyst interview guide walks through the SQL, case-study, and AI-usage rounds a data loop actually puts in front of you. The drill above needs nothing but Node.
Before your next data interview
Write the aggregator above from memory, then run the five assertions until they pass without looking.
Break it on purpose: set allowed lateness to zero and watch contract 2 fail; set it to an hour and watch contract 3 stop firing. Both failures answer "why not just make the bound bigger".
In under a minute, out loud: why the global watermark is a minimum, and what an idle partition does to an otherwise ready window.
Say your sink contract in one sentence: "upsert on
(window_start, key), because emissions are corrections and delivery is at-least-once."
Drafting was AI-assisted. The window lifecycle and allowedLateness semantics were checked against Flink's Windows documentation; the drop and correction numbers come from the script in this article, which prints all contracts hold on Node 24. Beam documents the same setting as withAllowedLateness in its programming guide.
Top comments (0)