Watermarks

Analogy categoryAnalogies

What a Watermark Represents

A watermark is a special record that flows through a Flink stream alongside normal data, carrying a single timestamp value that asserts: 'no more events with a timestamp earlier than this will arrive from now on.' It is Flink's mechanism for reasoning about event-time progress in an otherwise unbounded, out-of-order stream. When an operator such as a windowed aggregation sees a watermark with timestamp T, it treats all windows ending at or before T as complete and safe to fire, because the watermark is a promise that nothing older is still in flight.

🏏

Cricket analogy: A watermark is like the third umpire announcing 'no more reviews will be requested for overs 1 through 10' — once that announcement is made, the scorecard for those overs can be finalized even if some replay footage for over 9 is still uploading.

Bounded Out-of-Orderness Watermarks

The most common strategy, WatermarkStrategy.forBoundedOutOfOrderness(Duration), generates a watermark equal to the maximum event timestamp seen so far minus a fixed slack duration. If you configure a 5-second bound, an event timestamped 12:00:10 causes the watermark to advance to 12:00:05, meaning the operator now believes nothing earlier than 12:00:05 will arrive. Choosing this bound is a direct latency-versus-completeness trade-off: a small bound closes windows quickly but risks marking genuinely late data as 'too late,' while a large bound tolerates more disorder at the cost of longer wait before results are emitted.

🏏

Cricket analogy: Setting a 5-second bound is like a broadcaster deciding to wait only a fixed few seconds after a boundary before confirming the score graphic — wait too little and a delayed DRS override arrives after the graphic is locked, wait too long and viewers get an annoyingly delayed scoreboard.

Idleness and Per-Partition Watermarking

In a Kafka source with multiple partitions, Flink computes the watermark for the whole operator as the minimum of the watermarks across all its input partitions, because the operator can only be as confident as its slowest, most-behind source of data. If one partition goes idle — for example, a low-traffic IoT device that stops sending readings — that partition's watermark stalls and drags the entire operator's watermark down with it, stalling all downstream windows indefinitely. WatermarkStrategy.withIdleness(Duration) solves this by marking a source subtask as idle after a period of inactivity, excluding it from the minimum computation until it produces data again.

🏏

Cricket analogy: If a scoring system takes the earliest of all ground correspondents' clocks and one correspondent's feed goes silent during a rain delay, the whole match's live score freezes at that correspondent's last update until you explicitly mark that feed as 'currently idle.'

Punctuated and Custom Watermark Generators

Beyond the periodic, bounded-out-of-orderness generator, Flink lets you implement WatermarkGenerator directly for cases where progress should be inferred from the data itself rather than a fixed time bound — for example, a source that occasionally embeds an explicit 'checkpoint' record signaling that all prior data has been sent (a punctuated watermark). Custom generators implement onEvent (called per record, where you may or may not emit a watermark) and onPeriodicEmit (called on a timer, typically every 200ms by default, where periodic generators advance the watermark). This flexibility matters when out-of-orderness isn't a fixed duration but instead correlates with something domain-specific, like a batch boundary or a device's own sync marker.

🏏

Cricket analogy: A punctuated watermark is like a scorer who only updates the 'confirmed' scoreboard when the umpire explicitly signals 'over complete,' rather than guessing based on elapsed time, since overs can run long during a DRS review.

java

WatermarkStrategy<Event> strategy = WatermarkStrategy
    .<Event>forBoundedOutOfOrderness(Duration.ofSeconds(10))
    .withTimestampAssigner((event, ts) -> event.getTimestamp())
    .withIdleness(Duration.ofMinutes(1)); // ignore stalled partitions after 1 min

DataStream<Event> stream = env.fromSource(source, strategy, "events");

// Custom punctuated generator example
class BatchEndWatermarkGenerator implements WatermarkGenerator<Event> {
    private long maxTimestamp = Long.MIN_VALUE;

    @Override
    public void onEvent(Event event, long eventTimestamp, WatermarkOutput output) {
        maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
        if (event.isBatchEndMarker()) {
            output.emitWatermark(new Watermark(maxTimestamp));
        }
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        // no periodic emission; purely punctuated
    }
}

By default, Flink checks periodic watermark generators every 200ms via pipeline.auto-watermark-interval — you can tune this if you need finer-grained watermark advancement for low-latency use cases.

Forgetting withIdleness() on a source with sparse or bursty partitions is one of the most common causes of a Flink job whose windows never fire — the watermark silently stalls at the slowest partition's last-seen timestamp.

A watermark asserts that no event earlier than its timestamp will arrive again, letting operators safely close windows.
forBoundedOutOfOrderness sets the watermark to max-seen-timestamp minus a fixed slack, trading latency against completeness.
The operator's overall watermark is the minimum across all input partitions, so one stalled partition can block everything.
withIdleness(Duration) excludes inactive partitions from the minimum computation, preventing indefinite stalls.
Custom WatermarkGenerator implementations support punctuated watermarks driven by domain-specific signals rather than fixed durations.
Periodic generators are polled roughly every 200ms by default via pipeline.auto-watermark-interval.
Choosing a too-small out-of-orderness bound risks discarding late data; too-large a bound adds unnecessary latency.