Measuring Disorder to Size a Lateness Budget

Two numbers govern an event-time pipeline: how far the watermark trails the data, and how long window state is retained after a window closes. Both are usually chosen by intuition, and both are derivable from one measurable quantity — the distribution of arrival delay, meaning the difference between when a record enters the system and the event time it carries. Treating that distribution as data rather than as a nuisance turns a guess into a calculation with a stated error rate.

The quantity to measure

Arrival delay is arrival_time - event_time, computed per record. It is not latency in the request-response sense; it is the lag between an event happening and the pipeline seeing it, and it can be negative if a producer’s clock is ahead of the ingesting host’s.

The distribution is almost always heavy-tailed and bimodal. The bulk of records arrive within a small delay determined by the transport: a few milliseconds to a few seconds for a well-behaved producer, tens of seconds for a batch producer that flushes on a timer. A second mode sits much further out and comes from a different mechanism entirely — a mobile client uploading a buffer on reconnect, a producer retrying a failed publish, a mirroring job replaying a remote cluster’s backlog. The two modes have different causes, so a single percentile over the combined distribution describes neither.

Measuring this requires the event timestamp to be recorded next to the ingestion timestamp for a sample of records, which is the only reliable way to get it. Sampling one partition over a representative period and computing the delay histogram by hand is enough; the number does not change quickly, and it changes for identifiable reasons.

From a histogram to a setting

Suppose the delay histogram over a week shows that 99.9 percent of records arrive within 45 seconds and the remaining 0.1 percent spread from a minute to several hours. The watermark delay is not 45 seconds, because the watermark delay and the allowed lateness do different jobs.

The watermark delay sets output latency. A delay of d means every window result appears at least d after the last event in the window. Choosing d at the 99th percentile of delay means roughly one record in a hundred arrives after its window has closed, and each of those is a late record requiring a correction. Choosing d at the median means half the records are late, which is not a correction mechanism but a redefinition of the pipeline.

Allowed lateness sets how long the window’s state survives to absorb those corrections. It must cover the far tail of the distribution, and it is bounded from below by the watermark delay: lateness below the delay is meaningless because those records are not late. A configuration of d = 45s and l = 2h says that results are published 45 seconds after the window’s end and revised for up to two hours afterward.

The two numbers also have very different costs. Raising d costs latency on every window, uniformly. Raising l costs memory on every window that is still retained, which is the number of keys multiplied by the windows per key multiplied by the retained duration. A pipeline with a one-minute tumbling window, a hundred thousand keys, and a two-hour lateness allowance retains up to twelve million window states, each of which holds a partial aggregate. The aggregate size is what decides whether that is tens of megabytes or tens of gigabytes.

def percentile(sorted_delays, p):
    if not sorted_delays:
        raise ValueError("no samples")
    # nearest-rank: the smallest delay below which p percent of samples fall
    idx = min(int(p / 100.0 * len(sorted_delays)), len(sorted_delays) - 1)
    return sorted_delays[idx]

def budget(delays, watermark_pct=99.0, lateness_pct=99.9):
    d = sorted(delays)
    w = percentile(d, watermark_pct)
    l = percentile(d, lateness_pct)
    return w, max(l - w, 0)   # lateness beyond the watermark delay is what state must cover

The subtraction in the last line is the part that gets missed. If a record arrives within the watermark delay it is never late, so the retention window only has to cover the excess.

Negative delays and clock skew

A delay can be negative, and a negative delay is not a fast record — it is evidence that the producer’s clock is ahead of the ingester’s. Those records carry event times in the future relative to the receiving system’s clock, and they interact badly with a periodic watermark generator that uses the operator’s clock, because the record’s timestamp exceeds the clock reading and the generator’s max(seen) - d jumps forward.

The effect is a watermark that leaps ahead of the data and then closes windows prematurely, and the leap is caused by one record with a wrong clock. Generators that clamp the watermark to the current processing time prevent the leap but do not fix the record: the record still has an event time in the future, and when real time catches up with it, it is not late, it is early, and it was already counted in a window that has closed.

The measurement step should therefore report the negative tail explicitly. A pipeline with more than a negligible fraction of negative delays has a clock synchronization problem in its producers, and adjusting the watermark will not fix it. The measurement is the diagnostic; the fix is upstream.

What a late record actually costs

When a record arrives below the watermark and the window’s state has been released, the options are dropping it or routing it elsewhere. The routing option is a side output: a second stream carrying the records the main operator could not place, tagged with the window they would have belonged to. It is cheap, it preserves the record, and it makes the loss visible — the side output’s rate is the direct measurement of how wrong the budget is.

The side output is also the right place to put records whose event time is unparseable or absurd. A single stream carrying all rejected records, with a reason field, gives a reconciliation job one input to work from, and gives the pipeline’s operators one place to report what they could not handle. A pipeline without such a channel has no way to distinguish “no records were late” from “late records were silently discarded”, which is the difference between a budget that works and one that was never tested.

The correction path, where late records update a window that already emitted, is not free either. It requires the window’s state to be retained for the lateness allowance, and it requires every downstream consumer to understand that a result it already received may be replaced. Retractions and Updating Results covers the downstream contract; the point here is that the decision to enable corrections propagates all the way to the sink.

Budgets that survive a source change

A lateness budget is calibrated against a source’s behaviour, and source behaviour changes without announcement. A producer that switches from per-event publishing to batching on a one-minute timer changes the delay distribution’s first mode from milliseconds to a minute. Nothing in the pipeline signals the change; the only evidence is that windows start closing before their records arrive.

The countermeasure is to measure continuously rather than once. A delay histogram emitted as a metric — the count of records per delay bucket, aggregated over a sliding interval — makes the shift visible within one interval, and comparing the current 99th percentile against the configured watermark delay turns the budget into a monitored quantity rather than a constant. This is the same discipline as any other capacity parameter: the setting is only as good as the observation that it still matches reality.

The residual limitation is that no percentile bounds the maximum. A budget at the 99.9th percentile leaves one record in a thousand to the correction path, and the thousandth record can be arbitrarily late. Pipelines that must account for every record need the side output and a reconciliation step that joins it back against the published results, because the correction path handles the common tail and the side output handles everything else.