Watermarks as Completeness Signals

An event-time aggregate cannot emit a final result from the data alone. Nothing in the records says whether a record with an older timestamp is still in flight, so the operator either waits forever or guesses. A watermark is that guess made explicit: a value carried alongside the data asserting that the stream will not contain events older than it. The assertion is usually false at the level of strict logic and true often enough to be useful, and understanding which is which is most of the design work.

The assertion and its form

A watermark is a timestamp W attached to a stream, meaning “all future records in this stream have event time greater than or equal to W”. A record with event time below the watermark of its stream is late by definition. The definition is per-stream and per-partition; it says nothing about other streams, which is why a join must reconcile the watermarks of its inputs before it can make any completeness claim about the result.

The watermark value is not a physical time and is not compared against the wall clock. It is a position on the event-time axis. An operator that receives W = 12:00:30 concludes that its window ending at 12:00:30 is complete, even if the wall clock reads 12:05 and the record stream is still moving.

Monotonicity is required: a watermark may stay put or move forward, never backward. If a source emits a lower value than the one before, either the source is inconsistent or the stream has been merged and the operator is looking at a partition that was previously ahead. Merging makes this concrete — the watermark of the union of two streams is the minimum of theirs, so adding a partition can pull the combined watermark backward relative to one of the inputs. Implementations handle that by only ever publishing a non-decreasing value, which means a single slow partition holds back the whole stream.

Three ways to produce one

The periodic generator reads the operator’s clock and subtracts a fixed delay. If the clock is now T and the delay is d, the emitted watermark is T - d. Records whose event time is more than d behind the clock are late on arrival. This scheme is cheap and bounded in latency, and it silently drops the connection between the watermark and the data: on a source that has been idle for a minute, the watermark still advances and windows still close, which may be right or may be premature depending on whether the source resumes.

The punctuated generator takes its value from markers in the data rather than from a timer. The producer embeds a punctuation record carrying a watermark value, and the operator emits that value when it sees the punctuation. This is the scheme for sources that know their own completeness — a partitioned log where each partition’s end-of-batch marker is a natural watermark — and it puts the assertion in the hands of whoever produced the data, which is usually the party with the best information.

The third approach is a bounded-out-of-orderness watermark, which derives the value from the data without requiring explicit markers. The operator tracks the maximum event time seen and emits max_seen - d on a timer. The bound is a statement about the data: records are expected to arrive within d of their event time. Choosing d is choosing how much lateness to absorb. A d smaller than the actual disorder produces late records; a d larger than the actual disorder delays every output by the surplus. The value is a property of the source, not of the processor, and it is measured rather than guessed: the distribution of (arrival time − event time) over a representative window gives the number directly. On a stalled source the watermark stalls with it, so nothing closes while nothing arrives — which keeps the assertion honest and also means an idle stream emits nothing at all.

# Bounded-out-of-orderness watermark: maximum event time seen, minus the bound.
on_event(e):
    max_seen = max(max_seen, e.event_time)

on_timer():
    emit_watermark(max_seen - allowed_disorder)

The only state is the maximum, and the only parameter is the bound. Everything else in the mechanism is propagation.

Propagation through an operator graph

A watermark advances through the graph with the data, and each operator has to decide what its output watermark is. The rule for a narrow transformation — a map, a filter, a flat-map — is that the output watermark equals the input watermark, because the transformation does not reorder events in time. A filter that drops most records does not advance the watermark faster; the watermark describes the events that could arrive, not the ones that did.

A stateful or keyed operator is more careful. Its output watermark is the minimum of its input watermarks, and for an operator with multiple inputs it is the minimum across all of them. A join of a fast stream and a slow one holds the fast side back to the slow side’s watermark, which is correct and is also why a single lagging partition can stall an entire job. Interval Joins Over Event Time depends on this minimum, since a join cannot declare an interval unmatched until both sides are known to be complete past it.

A watermark that cannot advance is the common failure. The graph is a chain of minimums, so one source whose watermark is stuck pins every downstream watermark at that value, and every window in the pipeline stops closing. Diagnosing this from the outside looks like a hang with no error: the job is running, records are being consumed, and no output appears. Instrumenting the watermark value at each operator is the only way to see where it stops moving.

Watermarks also have to survive checkpointing. They are part of the operator state, so a restored job resumes with the watermark values it had at the checkpoint rather than restarting from zero. Restarting from zero would re-close windows that already emitted, so the watermark’s position is one of the things a checkpoint must capture. Checkpoint Barriers and Aligned Snapshots describes how the barrier carries the position that makes this consistent.

When the assertion is false

Every real source violates its own watermark eventually. A producer retries a failed send an hour later; a mobile client uploads a batched backlog when it reconnects; a clock on one device is behind by minutes and stamps its records accordingly. The record arrives with an event time below the watermark that already passed, and the processor has three options.

It can drop the record, which is what an operator does when no lateness allowance is configured. The output stays consistent with the watermark and is missing data, and the loss is silent unless the operator counts it.

It can accept the record and update the result it already emitted, which is what an allowed-lateness setting enables. The window’s state is retained for l beyond its end, and a late record inside that window produces a new emission for the window. Downstream consumers then see two records for the same window and must know whether to replace or to add. The retraction mechanism that makes this safe is the subject of Retractions and Updating Results.

It can route the record to a side output, which preserves it without corrupting the main result and leaves the decision about what to do with it to a separate job.

The choice is a policy, and the policy has to be stated per pipeline rather than inherited from a default. The failure mode of an unstated policy is the worst of the three: records vanish, nobody counts them, and the totals are quietly short by an amount that grows with the source’s disorder.

Watermarks and the accuracy-latency dial

The watermark position is the main control over output latency in an event-time pipeline. Trailing the data by less means results appear sooner and more records are late; trailing by more means results appear later and fewer records are late. The dial has two ends that are both legitimate: a real-time dashboard can run with a short trail and accept that a small fraction of records will be corrected by later updates, while a nightly settlement job can run with a trail of hours and emit each window once.

The dial does not change correctness, only when the answer is available and how often it is revised. A pipeline with a correct late-record policy is correct at any watermark setting; a pipeline without one is correct only at the setting where no record is ever late, which is not a setting that can be found in advance.

One consequence is worth stating plainly: a watermark is not a guarantee about the data, and no amount of engineering makes it one. It is a claim the system makes so that it can proceed, and the engineering goes into detecting when the claim is violated and handling the records that result.