Apache Flink: Watermarks and Exactly-Once Streams

It picks the wrong winner. A leaderboard that counts each event by the time it reaches the server puts late uploads in the wrong time slot. Phones with weak signal upload late all the time. Flink watermarks are how a Flink job decides how long to wait for them.

Picture a common setup. A running app holds a city-wide challenge every weekend morning. Each hour from 6 am to 10 am, the runner with the most kilometres in that hour wins a badge and a voucher. The setup is made up, but the problem is real.

Each time a runner finishes a kilometre, the phone sends a small event: runner id, distance and the time on the phone. About 22,000 runners take part on a busy Saturday (made up).

The first version was simple. A worker read events from Kafka, a system that stores streams of events in order, and added each kilometre to the hour shown on the server clock. It worked in testing.

Then real weekends happened. Routes through parks and hills lose mobile signal, so phones store their events and upload them in a burst when signal returns, sometimes 15 minutes later.

• A kilometre run at 6:56 am reached the server at 7:09 am. It counted toward the 7 am hour, so the 6 am badge went to someone who had run less.
• One morning the worker crashed and restarted. It read some events from Kafka a second time, and a few runners got kilometres they never ran.

The team rebuilds the job on Apache Flink, an open-source engine that processes streams of events as they arrive. Watermarks fix the late events. Checkpoints, covered in the engineer half, fix the double counting.

Event time is when an event really happened on the device that made it. Processing time is the clock of the machine handling the event right now. The gap between them can be milliseconds or many minutes.

The Flink documentation calls processing time the simplest choice, with the best performance and lowest latency. It also warns that results then depend on how fast records arrive and move through the job, and on outages. Run the same data twice and you can get two different leaderboards.

Event time reads a timestamp written inside each event, here the moment the phone recorded the kilometre. Once all the data has arrived, the result depends only on the data, not on the server clock. Replaying Saturday's events gives the same totals, with one exception. Which events count as too late can change, because that depends on how fast and in what order Flink reads them.

Think of a pigeon race. Each bird is clocked the moment it lands at its home loft, far from the other lofts. The race committee collects those landing times later, some quickly and some much later. Here is how the picture maps:

• The landing time on the loft clock is event time, like the time stamped on the phone.
• The moment the committee receives that time is processing time.
• The committee's rule for when to stop waiting and declare results is the watermark.
• A landing time that reaches the committee after results are out is a late event.

Where the picture breaks: a race committee can wait days for every clock. A leaderboard cannot, and some phones never upload at all. So Flink has to guess when to stop waiting.

That guess is the real problem. The 6 am hour has no natural end in event time. At 7:00 on the server clock, events from 6:58 may still be on a phone in the hills. So when should Flink announce the 6 am winner?

Here is the pipeline. Follow the arrows from the phones to the screen, and notice that the phone, not the server, stamps the time.

Flink watermarks are markers that flow inside the stream and carry a timestamp. A watermark with time t tells Flink that event time has reached t, so no more events at or before t should arrive.

In other words, a watermark is Flink's clock for event time. Each operator, meaning one processing step in the job, moves its clock forward when a newer watermark arrives. Windows read that clock to decide when they are complete.

A window is a slice of the stream grouped by time. This job uses tumbling windows, fixed one-hour slots that do not overlap: 6:00 to 7:00, 7:00 to 8:00, and so on. The stream is keyed by runner id, meaning grouped per runner, so each runner gets a window per hour.

By default, an event-time window fires once the watermark passes the end of the window. Firing means the window runs its function, here adding up kilometres, and sends the result on.

Watermarks are a guess, and the guess can be wrong

A watermark is a promise Flink cannot fully keep, because it cannot see events still waiting on phones. Flink's documentation says Flink implements many techniques from the Dataflow Model paper (VLDB 2015). That paper names both ways the guess fails.

• Too fast: the watermark moves on while some events are still on their way. Those events arrive behind the watermark, which makes them late.
• Too slow: the watermark waits longer than needed. Results are correct but reach the screen late. The paper adds that a single slow piece of data can hold back the whole pipeline.

Measure how late your events really arrive, then pick a delay that covers most of them without making results feel slow. In Flink this delay is the bound of a bounded out-of-orderness watermark strategy.

Bounded out-of-orderness means you assume no event arrives more than a fixed time behind the newest one. Flink tracks the largest event time it has seen. The watermark is that largest time minus your bound, minus 1 millisecond.

The team logs one weekend of gaps between phone time and arrival time in Kafka. In this made-up data, 97 percent of events arrive within 2 minutes and 99.5 percent within 20 minutes. A few take hours.

They set the bound to 2 minutes. The 3 percent that arrive later are what allowed lateness, in the next section, is for.

Notice that the watermark is computed from events, not from the wall clock. The 6 am window fires once Flink has seen an event stamped 7:02 or later, whatever time the server clock shows.

By default, Flink drops an event that arrives after the watermark has passed the end of its window. Allowed lateness lets the window update its result instead, and a side output catches anything later still.

Allowed lateness is extra event time after a window's end during which Flink keeps the window's data. The Flink documentation says its default is 0. A late event inside that time is added, and the window fires again with a corrected result.

A side output is a second stream coming out of the same step. With sideOutputLateData, or side_output_late_data in Python, events that miss even the allowed lateness go there instead of being thrown away.

The team sets allowed lateness to 20 minutes. The 6 am window first fires when the watermark reaches the hour's last millisecond, 6:59:59.999. Late events keep correcting it until the watermark reaches 7:19:59.999. Then Flink deletes the window's state, the data it remembers for that window, such as each runner's running total.

Here is that rule as plain Python you can run, with no Flink needed. Times are milliseconds after midnight, and Flink also counts time in milliseconds.

HOUR_MS = 60 * 60 * 1000
BOUND_MS = 2 * 60 * 1000          # wait 2 minutes for stragglers
LATENESS_MS = 20 * 60 * 1000      # then accept corrections for 20 more

def at(h, m, s=0, ms=0):
    return ((h * 60 + m) * 60 + s) * 1000 + ms

class BoundedWatermark:
    def __init__(self):
        self.max_seen = None

    def on_event(self, event_ms):
        if self.max_seen is None or event_ms > self.max_seen:
            self.max_seen = event_ms

    def current(self):
        return self.max_seen - BOUND_MS - 1

def classify(event_ms, watermark_ms):
    window_last_ms = (event_ms // HOUR_MS + 1) * HOUR_MS - 1
    if watermark_ms < window_last_ms:
        return "on time"
    if watermark_ms < window_last_ms + LATENESS_MS:
        return "late, result updated"
    return "too late, side output"

wm = BoundedWatermark()
wm.on_event(at(7, 5))                            # newest event seen: 7:05
print(wm.current() == at(7, 2, 59, 999))         # True
print(classify(at(6, 56), at(6, 59, 59, 998)))   # on time
print(classify(at(6, 56), wm.current()))         # late, result updated
print(classify(at(6, 56), at(7, 19, 59, 999)))   # too late, side output

Expected output, one line each: True, on time, late, result updated, and too late, side output. The 6:56 kilometre is late but still counted.

What just happened? The current method subtracts the bound and 1 more millisecond, as Flink's built-in generator does in its code. A window's last millisecond is its end minus 1, because the end is exclusive. Flink 2.3.0 gave the same three answers when we replayed these events through a real job.

Here is the full path of one event, as a decision. Start at the top and follow the labelled arrows.

The middle row has a catch. The Flink documentation warns that late firings put several results for the same window into your stream. Whatever reads Flink's output must treat the newest one as correct.

So the sink, the step that writes results out of Flink, stores a full total per runner and hour, for example runner 4471, 6 am, 11.2 km. A later firing simply overwrites it. Nothing is added twice.

Allowed lateness has a cost. Flink keeps each window's state until the allowed lateness ends. Here that means 20 extra minutes for about 22,000 small totals. Twenty hours of lateness would keep up to about 21 hourly windows per runner alive at once.

The product team adds one rule on top. The screen marks an hour provisional until its allowed lateness ends, then final. Prizes go out on Monday, after someone checks the late corrections table for big misses.

Use event time and watermarks when the answer depends on when things happened, and events can arrive late or out of order. Use processing time when you only care about what the system sees right now.

Event time fits this challenge, because a prize depends on when each kilometre was run. It also fits billing by the minute, fraud checks on card swipes and sensor readings from places with poor signal.

Processing time is fine for a live chart of server load, where an event landing in the next second does not matter. It is also simpler, with no watermarks to tune.

Sometimes a stream job is not needed at all. If results can wait until the next morning, a nightly batch job over the full day's events sees every upload that has arrived by then and needs no watermarks.

An event-time window fires when the watermark reaches its last millisecond, the end minus 1. With the built-in bounded out-of-orderness strategy, that happens once Flink has seen an event stamped at least the end plus the bound.

We tested this on Flink 2.3.0, the latest release on PyPI as of 19 October 2026, with a 2-minute bound and hourly windows. A newest event stamped 7:01:59.999 did not fire the 6 am window. One stamped 7:02:00.000 fired it, with the watermark at 6:59:59.999.

Two more details shape the timing. Built-in strategies emit watermarks periodically, every 200 milliseconds by default, set by pipeline.auto-watermark-interval. Flink emits a new watermark only if it is larger than the previous one, so the clock never moves backward.

What the real job looks like in PyFlink

Here is the same job in Flink's Python API, PyFlink 2.3.0. A small list stands in for the Kafka source, so it runs on a laptop. PyFlink runs on Java underneath, so a Java runtime must be installed; we used OpenJDK 17.

import sys
from pyflink.common import Duration, Time, Types, WatermarkStrategy
from pyflink.common.watermark_strategy import TimestampAssigner
from pyflink.datastream import OutputTag, StreamExecutionEnvironment
from pyflink.datastream.window import TumblingEventTimeWindows

class PhoneTime(TimestampAssigner):
    def extract_timestamp(self, value, record_timestamp):
        return value[2]                      # when the km was run, per the phone

def km_type():                               # (runner id, km, phone time in ms)
    return Types.TUPLE([Types.STRING(), Types.FLOAT(), Types.LONG()])

strategy = (WatermarkStrategy
            .for_bounded_out_of_orderness(Duration.of_minutes(2))
            .with_idleness(Duration.of_minutes(1))     # before the assigner
            .with_timestamp_assigner(PhoneTime()))

env = StreamExecutionEnvironment.get_execution_environment()
env.set_python_executable(sys.executable)  # workers use this same Python
env.set_parallelism(1)
env.enable_checkpointing(30 * 1000)          # a checkpoint every 30 seconds

MIN = 60 * 1000
# Stand-in for the Kafka source: three events, phone time in ms after midnight
events = env.from_collection(
    [("r4471", 1.0, 390 * MIN), ("r4471", 1.0, 416 * MIN), ("r9", 1.0, 425 * MIN)],
    type_info=km_type()).assign_timestamps_and_watermarks(strategy)

late_tag = OutputTag("late-km", km_type())
hourly = (events
          .key_by(lambda e: e[0], key_type=Types.STRING())
          .window(TumblingEventTimeWindows.of(Time.hours(1)))
          .allowed_lateness(20 * MIN)
          .side_output_late_data(late_tag)
          .reduce(lambda a, b: (a[0], a[1] + b[1], max(a[2], b[2])),
                  output_type=km_type()))

hourly.print()
hourly.get_side_output(late_tag).print()
env.execute("hourly-km")

Expected output: (r4471,2.0,24960000) and then (r9,1.0,25500000). The first is runner r4471's 6 am total of 2 km; the second is r9's 7 am total. Without the set_python_executable line, Flink starts its Python workers with the python on your PATH. In our test that Python lacked PyFlink, and after 150 seconds the job had printed nothing.

Nothing is late here, and that is expected. When a finite list ends, Flink sends a final watermark that fires every window. Lateness only shows up when events arrive after the watermark has moved, as in a live stream.

A PyFlink trap: your timestamp assigner can be ignored

The timestamp assigner is the small class that tells Flink which field holds event time, here PhoneTime. In PyFlink 2.3.0 it is honoured in only one place, and two common moves lose it without any warning.

• Calling with_idleness or with_watermark_alignment after with_timestamp_assigner returns a new strategy without your assigner. Call them first, as the code above does.
• Passing the strategy to env.from_source hands only the Java part to the source, so a Python assigner there is ignored. Attach it with assign_timestamps_and_watermarks after the source instead.

We ran both mistakes. With a list as input, the window step failed because records had no timestamp. With a Kafka source, Flink's documentation says the Kafka record's own timestamp is used when no assigner is given, so the job would not fail.

That timestamp is not the phone time. Unless the intake service sets one, Kafka's producer stamps each record with its own current time, so the leaderboard would quietly count by upload time again.

This is not in the Flink docs as of 19 October 2026. The with_idleness case is open Apache issue FLINK-32040, reported for PyFlink 1.16 and 1.17, and our run shows it unchanged in 2.3.0. We traced the with_watermark_alignment and from_source cases in the PyFlink source files. Pin your PyFlink version and re-run this check after upgrades.

Attaching the strategy after the source has a cost. The Flink documentation says a strategy on the source can track each Kafka partition separately and is more accurate. If that matters, write this step in Java. Or have the intake service set phone time as the Kafka record timestamp, on a topic left on Kafka's CreateTime setting so the broker keeps it.

Usually because one input has gone quiet. A step that reads several inputs sets its clock to the minimum of their watermarks, so the slowest or emptiest input holds everything back.

The Kafka topic has 8 partitions, which are separate ordered logs inside one topic. With the strategy set on the Kafka source, Flink tracks a watermark per partition and merges them by taking the minimum. The window step then takes the minimum again across all source tasks.

The PyFlink code above sets its strategy after the source instead. Then each source task makes one watermark from its partitions' events mixed together, and the window step still takes the minimum across tasks.

The diagram below shows the strategy set on the Kafka source. Look at where the minimum is taken: one quiet partition decides the clock for everything after it.

Early on Sunday few people run, and some partitions get no events for a while. The fix is an idleness timeout, here 1 minute. With the strategy on the Kafka source, a partition that sends nothing for that long is marked idle and left out of the minimum.

With the strategy after the source, as in the PyFlink code above, a quiet partition does not hold the clock back, but its events risk arriving late. There, with_idleness marks a whole source task idle only once the task sends nothing for 1 minute.

An idleness timeout can turn a quiet input's events into late data

We tested this on Flink 2.3.0 with two inputs: a busy one and a quiet one that sends one event every 4 seconds. Without an idleness timeout, no hourly window fired in 20 seconds, though the busy input had raced hours ahead.

With a 1-second idleness timeout, the windows fired. But the quiet input's next events, stamped 6:02, 6:03 and 6:04, arrived after the watermark had passed 7:00. All three went to the side output as late.

So an idleness timeout is a bet that a quiet input stays quiet. Set it longer than the normal gaps between events, and keep a side output to catch what the bet misses.

One Kafka detail: the Kafka connector documentation says its source does not go idle by itself when there are more source tasks than partitions. Lower the parallelism, meaning the number of source tasks, or add an idleness timeout.

The opposite problem also exists. One input can race far ahead, for example while catching up after downtime, and the window step must hold its events as state. Watermark alignment pauses reading from any source that gets too far ahead. It works only when the strategy is set on the source.

How to check the watermark in a live job

Open the window task in Flink's web interface and read its currentInputWatermark metric. It is the lowest watermark the task has received, in milliseconds since 1 January 1970 UTC. To read one as Indian time, run this:

python3 -c "import datetime as d; print(d.datetime.fromtimestamp(1792200600000 / 1000, d.timezone(d.timedelta(hours=5, minutes=30))))"

Expected output: 2026-10-17 07:00:00+05:30, which is 7 am on Saturday. If that number stops growing while events still flow, look for an idle input.

The numLateRecordsDropped metric counts late events the window step threw away. In Flink 2.3.0's window code, an event sent to a side output is not counted there, so watch the side output itself too.

Flink takes regular checkpoints, which save every step's state together with the source positions. After a crash it restores the last complete checkpoint and replays events from the saved positions, so its state counts each event once.

Exactly-once here means each event affects Flink's state once, even if it is read twice. The wider idea of delivery guarantees, from at-most-once to exactly-once, has its own guide. This section covers only how Flink does it.

A checkpoint barrier is a small marker that Flink injects into the stream at the sources. Barriers flow in line with events and never overtake them. Everything before barrier n belongs to checkpoint n.

1. 1

   The source records its position

   Flink's Kafka source notes the offset, its read position, in each partition. Then it sends barrier 41 downstream after the last event it read.

2. 2

   Steps wait for every barrier

   A step with several inputs stops reading an input once barrier 41 arrives on it. It waits until barrier 41 has arrived on all inputs. This is barrier alignment.

3. 3

   Each step saves its state

   The window step saves its open windows, such as runner 4471 at 9.4 km for the 6 am hour. Then it forwards the barrier.

4. 4

   The checkpoint completes

   When every sink has received barrier 41 and reported it, checkpoint 41 is complete. Offsets and state now describe the same moment in the stream.

5. 5

   After a crash, rewind and replay

   Flink restores every step from checkpoint 41 and sets the Kafka source back to the saved offsets. Events after that point are read again, but the restored state never saw them.

The trick is that offsets and state roll back together. Runner 4471's total goes back to 9.4 km, and the source rereads exactly the kilometres that came after it. No kilometre is counted twice in Flink's state.

This needs a source that can rewind. Kafka can, as long as it still holds the events. Checkpointing is also off by default, so the job turns it on with a 30-second interval, as enable_checkpointing does in the code above.

Alignment can slow a job under backpressure, when a slow step makes the steps before it wait. Unaligned checkpoints let barriers overtake buffered events and save those events in the checkpoint. Flink allows them only in exactly-once mode with one checkpoint at a time.

Only if the sink cooperates. Checkpoints make Flink's own state exactly-once, but the sink must either commit writes in step with checkpoints, or write in a way that a replay cannot harm.

Here is the gap. The sink writes the 6 am result, then Flink crashes before the next checkpoint. After the restore, the window fires again and the sink writes the same result a second time.

Transactional writes to Kafka

Flink's Kafka sink has an exactly-once mode. It writes results inside a Kafka transaction that commits on a checkpoint. Readers set to read only committed data, through Kafka's isolation.level setting, see no duplicates after a restart.

The team uses this for a results topic that the rewards service reads. The Kafka connector documentation adds three rules:

• Results become visible only after a checkpoint, so readers see them up to one checkpoint interval later.
• Each application needs its own transactional id prefix on the Kafka cluster, so running jobs do not interfere with each other's transactions.
• Set Kafka's transaction.timeout.ms well above the longest checkpoint plus the longest restart. Otherwise Kafka can expire an open transaction and data can be lost.

Idempotent writes to the leaderboard

An idempotent write gives the same result however many times you repeat it. Setting runner 4471's 6 am total to 11.2 km is idempotent. Adding 1 km to it is not.

The leaderboard store uses idempotent writes, keyed by runner and hour. A replayed result overwrites the same entry with the same number. The same design already handles late firings, as you saw earlier.

Explain them as Flink's clock for event time, then show the trade-off between waiting and being correct with one concrete example. Follow these steps.

1. 1

   Name the time you count by

   Say that the leaderboard counts by event time, stamped on the phone. Give one example of an event that arrives 13 minutes late.

2. 2

   Define the watermark

   A watermark with time t means no more events at or before t are expected. Windows fire when the watermark reaches their last millisecond.

3. 3

   Pick the bound from data

   Say you would measure arrival lag first, then set the out-of-orderness bound to cover most events, such as 2 minutes.

4. 4

   Handle the stragglers

   Add allowed lateness to correct results, a side output for anything later, and a provisional label on the screen until results are final.

5. 5

   Cover stuck watermarks

   Mention that a step takes the minimum watermark of its inputs, so idle partitions need an idleness timeout, and that the timeout can make late data.

6. 6

   Finish with exactly-once

   Explain checkpoints and barriers, then the sink: a Kafka transaction that commits on checkpoint, or idempotent writes keyed by runner and hour.