Watermarks

Who decides that a minute is over

Module 4 left GatewayMeter billing the right minute: TenantCost runs over 1-minute tumbling event-time windows on RequestRecord.completed_ts, so a retry-buffered event delayed 30 s still lands in the minute it happened. It also left a question open. The window for [10:00:00, 10:01:00) has to close at some point, and nothing in the data says "there will be no more 10:00 events". Something has to decide.

The mechanism is the watermark, and its definition is worth keeping in upstream's exact words:

Watermarks flow as part of the data stream and carry a timestamp t. A Watermark(t) declares that event time has reached time t in that stream, meaning that there should be no more elements from the stream with a timestamp t' <= t (i.e. events with timestamps older or equal to the watermark).

Read "should be". A watermark is a claim, not a fact; upstream's training material hedges it as an assertion that the stream is (probably) now complete up through time t. An element with timestamp ≤ t can still show up after Watermark(t), and when it does it is late — defined against the watermark, never the wall clock:

Late elements are elements that arrive after the system's event time clock (as signaled by the watermarks) has already passed the time of the late element's timestamp.

Out-of-order is not late: an event can arrive behind a newer one and still be on time, as long as no watermark has passed it. Late is what happens when a claim turns out wrong.

So who makes the claim? The watermark generator you configured, from the data it has seen. Nobody else — not Kafka, not the gateway, not the window operator. Too optimistic a rule and events are late, billed to nothing; too pessimistic and tenant-cost lags the world. Upstream calls this the tradeoff between latency and completeness, and it is yours to set.

Where watermarks are generated

Upstream, on placement:

Watermarks are generated at, or directly after, source functions. Each parallel subtask of a source function usually generates its watermarks independently. These watermarks define the event time at that particular parallel source.

For GatewayMeter that means one generator per source subtask, each with its own opinion of what time it is. There is no global clock. A WatermarkStrategy goes either directly on the source or after a later operation via assign_timestamps_and_watermarks; upstream prefers the first because it allows sources to exploit knowledge about shards/partitions/splits in the watermarking logic.

For Kafka that matters. gateway-events has 8 partitions read by 4 source subtasks, so each subtask interleaves two partitions and destroys whatever per-partition order existed. With the strategy on the source, watermarks are generated inside the Kafka consumer, per Kafka partition, and the per-partition watermarks are merged in the same way as watermarks are merged on stream shuffles. Hold on to "merged" — the failure section is about what it means.

The bounded-out-of-orderness rule

Flink ships two periodic generators. for_monotonous_timestamps() assumes timestamps are ascending per parallel source task and uses the current timestamp as the watermark. for_bounded_out_of_orderness(...) keeps the watermark lagging behind the maximum (event-time) timestamp seen in the stream by a fixed amount of time.

The arithmetic is in the reference generator upstream prints: on every event, remember max(currentMaxTimestamp, eventTimestamp); on every periodic emit, output currentMaxTimestamp - maxOutOfOrderness - 1. The extra millisecond keeps an element exactly bound behind the maximum on time. The course writes watermark = max timestamp seen − bound; carry the −1 ms in your head.

"Periodic" is the other half. A periodic generator observes the incoming events via onEvent() and then emits a watermark when the framework calls onPeriodicEmit(); the interval is set via ExecutionConfig.setAutoWatermarkInterval(...) (Java setter; the configuration key is pipeline.auto-watermark-interval), and a new watermark will be emitted if the returned watermark is non-null and larger than the previous watermark. The default period is 200 ms as of 2026-09 — upstream states it on the SQL time-attributes page, which uses the same key; see https://nightlies.apache.org/flink/flink-docs-release-2.3/docs/deployment/config/. The other style, a punctuated generator emitting on marker events, is Java only: the Python tab for custom WatermarkGenerators reads "Still not supported in Python API." GatewayMeter uses the built-in strategies.

Here is the rule by hand for one partition, bound 20 s, timestamps as mm:ss in one hour. "Next watermark" is what the next periodic emit would send; lateness is judged against the watermark already emitted when the event arrives.

Rows 3 and 5 are the frozen distinction in one table: out-of-order costs nothing; late is the generator's claim failing. The wall clock appears nowhere.

Propagation: the minimum rule

Watermarks travel with the data. Whenever an operator advances its event time, it generates a new watermark downstream for its successor operators, and it must completely process a given watermark before forwarding it: a window operator first evaluates all windows that should be fired, and only after producing all of the output triggered by the watermark will the watermark itself be sent downstream.

The rule that decides everything in this module is what happens where streams meet:

Some operators consume multiple input streams; a union, for example, or operators following a keyBy(...) or partition(...) function. Such an operator's current event time is the minimum of its input streams' event times. As its input streams update their event times, so does the operator.

Apply it to GatewayMeter. After key_by(request_id), each of the four assemble subtasks receives records from all eight source subtasks, one per partition (stage 5, below), so its watermark is the minimum over all eight partitions. The same again after key_by(tenant_id) into tenant-cost. A subtask handling only tenant acme's records still has its clock set by the slowest partition of any tenant. Keying spreads the data out; the minimum rule pulls the clock back to the laggard. It also means a subtask processes only its keys' events but inherits a claim computed at the sources from all the data.

The worked example

The timeline every learner in this course must be able to answer cold, as seen at one tenant-cost subtask. Wall-clock time runs along the line; each note carries the record's event time:

Is C late? Yes. Not because it arrived at 10:01:10, 15 s after its timestamp, and not because it is 6 s behind A. It is late because the operator's current watermark is 10:01:00 and 10:00:55 ≤ 10:01:00. B, at 10:00:57, was out of order relative to A and not late: no watermark had passed it.

Who decided? The watermark generator you configured, from the data it had seen. By 10:01:07 every source subtask's generator had seen enough timestamps to claim at least 10:01:00 (the minimum rule), and the claim propagated here — events this subtask never saw. C's arrival changed nothing; the decision predates it.

Is C discarded? That depends on the operator, not the watermark. Late elements are, by default, ignored when computing the result of the job for their corresponding window — a window operator drops C and counts it in numLateRecordsDropped, unless allowed lateness or a side output says otherwise (module 6). A KeyedProcessFunction such as RequestAssembler sees C in process_element like any record; ctx.timer_service().current_watermark() tells it C is behind, and it may do anything with that. The watermark is an input to the operator, not a filter in front of it.

Can C change an already-emitted result? Only if the operator re-emits. With allowed lateness a window re-fires and a second TenantCostMinute for the same (tenant_id, window_end) reaches the sink — module 6's failure. Without it, C's minute was billed at the watermark and C's tokens are never billed. Revenue lost, silently.

GatewayMeter stage 5

Upstream's advice is to put the strategy on the Kafka source, so that generation is per partition. In PyFlink that cannot read Event.ts: from_source hands the source only the Java half of a WatermarkStrategy, and a Python TimestampAssigner is dropped without an error, which leaves event time at the Kafka record timestamp, the moment the producer sent the record. So GatewayMeter reads Event.ts in assign_timestamps_and_watermarks, right after the source, and runs the source with one subtask per partition. Each generator still sees exactly one partition, and nothing is interleaved before the watermark is computed:

from pyflink.common import Duration, WatermarkStrategy
from pyflink.common.watermark_strategy import TimestampAssigner
# KafkaSource, KafkaOffsetsInitializer, SimpleStringSchema as in module 1


class EventTsAssigner(TimestampAssigner):
    """Event time is the gateway's `ts` (ms since epoch), never Kafka append time."""

    def extract_timestamp(self, value, record_timestamp) -> int:
        return value.ts                    # value is the parsed Event


watermarks = WatermarkStrategy \
    .for_bounded_out_of_orderness(Duration.of_seconds(20)) \
    .with_idleness(Duration.of_minutes(1)) \
    .with_timestamp_assigner(EventTsAssigner())    # last: with_idleness returns a strategy without it

env.set_parallelism(8)                     # one source subtask per partition of `gateway-events`
events = env.from_source(
    source=gateway_events_source,          # KafkaSource over `gateway-events`, 8 partitions
    watermark_strategy=WatermarkStrategy.no_watermarks(),
    source_name="gateway-events") \
    .map(parse_event).uid("parse") \
    .assign_timestamps_and_watermarks(watermarks)
env.set_parallelism(4)                     # the rest of the job

Three decisions are encoded there. The bound is 20 s: module 4 measured the retry buffer at up to 30 s, and 20 s is where the course sits on the latency/completeness tradeoff. The assigner reads Event.ts rather than the Kafka record timestamp the source would use by default, because that is when the gateway published, not when the thing happened. with_idleness is for the failure below.

Two PyFlink limits, stated plainly. The Kafka connector page marks defining your own WatermarkStrategy for the source as "(Not supported in PyFlink)". And a built-in strategy passed to from_source keeps only its Java half, so on the source it can only use the Kafka record timestamp; on the course cluster (2026-09-19), a trace stamped in 2027 read that way produced watermarks that followed the wall clock. The layout above costs slots: from stage 4 the source runs eight subtasks, so a stage-4 job needs all eight slots of the compose cluster, which module 7 will feel. The environment's parallelism is set around the source because assign_timestamps_and_watermarks adds operators that take the default parallelism, and each must stay one-to-one with its partition.

The failure: one partition goes quiet

Partition 5 of gateway-events stops receiving records — the gateway node producing to it stalled, or the traffic that happened to hash there dried up. A tenant going quiet is not sufficient: the Kafka key is request_id, so other tenants can still route requests to p5. Lab 5.2 uses --stall-partition 5 to omit only request IDs that the console producer's default keyed partitioner would send there, after a warm-up that gives p5 a watermark first. Everything else is healthy. Wall clock 10:05:00, source parallelism 8, one partition per subtask:

Every tenant-cost subtask takes the minimum over S0–S7: 10:01:12. So every tenant's windows [10:01, 10:02), [10:02, 10:03), [10:03, 10:04) are open, none can fire, and the last TenantCostMinute anyone received had window_end = 10:01:00. Not one tenant sent traffic to p5 in particular; the minimum rule does not care. Upstream names the condition:

If one of the input splits/partitions/shards does not carry events for a while this means that the WatermarkGenerator also does not get any new information on which to base a watermark. We call this an idle input or an idle source. This is a problem because it can happen that some of your partitions do still carry events. In that case, the watermark will be held back, because it is computed as the minimum over all the different parallel watermarks.

How you see it. On the tenant-cost vertex, currentInputWatermark — the last watermark this operator/tasks has received (in milliseconds) — is frozen at 10:01:12 while numRecordsIn keeps climbing. That pair is the signature: data flowing, time stopped. Walk it upstream: on the operator that assigns event time right after the source, currentOutputWatermark (the last watermark this operator has emitted) is frozen on subtask 5 only, the one reading p5. (With the strategy on the source, as in Java, the source's watermarkLag and the split-level watermark.currentWatermark would show the same thing.) Money: nothing is billed wrong yet, but three minutes of every tenant's revenue sits in window state, growing, and the billing system reads nothing.

The fix. with_idleness(Duration.of_minutes(1)): if no records flow in a partition of a stream for that amount of time, then that partition is considered "idle" and will not hold back the progress of watermarks in downstream operators. With it configured, one minute after p5's last record (10:02:35 wall) S5 is marked idle and drops out of the minimum, and the stall never builds: at 10:05:00 the minimum would be 10:04:36, with the windows ending 10:02, 10:03 and 10:04 fired on time instead of held. Lab 5.3 shortens this timeout to 15 seconds so you can observe the change without a long wait. Note upstream's Kafka warning for module 10: the Kafka Source does not go automatically in an idle state if the parallelism is higher than the number of partitions — 32 slots over 8 partitions leaves 24 subtasks with nothing to say unless idleness is configured.

Its cost. Idleness changes who the claim covers. When p5 wakes at 10:06 carrying events stamped 10:01:40 onward, the downstream watermark is past 10:05 and every one of them is late. The tenant-cost window drops them (numLateRecordsDropped), unless module 6's two-minute allowed lateness still covers their window (with the watermark near 10:05:40, only those stamped 10:03:00 or later) and the same minute is re-emitted. Idleness trades a stall for lateness; it does not make the events on time.

Slow is not dead. A slow partition still produces; its watermark advances, just behind the others. Idleness never triggers, and the minimum rule keeps downstream waiting while the fast partitions' records pile into window state — upstream's concern, uncontrollable growth of the operator's state. Watermark alignment addresses that:

WatermarkStrategy \
    .for_bounded_out_of_orderness(Duration.of_seconds(20)) \
    .with_watermark_alignment("alignment-group-1", Duration.of_seconds(20), Duration.of_seconds(1))

The three parameters, per upstream: a label that binds together all sources that share it; the maximal drift from the current minimal watermarks across all sources belonging to that group; and how often the current maximal watermark should be updated. In order to achieve the alignment Flink will pause consuming from the source/task, which generated watermark that is too far into the future. Since Flink 1.17 the pausing works per split, so p4 can be paused while p5 catches up inside the same subtask (setting pipeline.watermark-alignment.allow-unaligned-source-splits to true turns that off); it works only for FLIP-27 sources with the strategy on the source, not after assign_timestamps_and_watermarks. watermarkAlignmentDrift and watermark.pausedTimeMsPerSecond show it working.

Alignment is a throttle, not a fix. It bounds how far ahead the healthy partitions may run and does nothing to move the slow one; against a dead partition it is harmful — the frozen watermark pins the group minimum, every healthy partition drifts past the cap, and Flink pauses them all. Diagnose first: watermark.currentWatermark moving slowly means slow, align; not moving with sourceIdleTime climbing means dead, idleness. Lab 5 tests the idleness case; the slow-partition alignment comparison remains a reading exercise. The capstone keeps idleness and leaves alignment off, because against its dead partition alignment would pause every healthy one.

One dated sentence on DataStream API v2 (experimental as of 2026-09): its "watermark" is a different thing — a user-defined event propagated along the stream with a chosen combine function — and the course stays on the classic API.

If a TaskManager dies right now

Which numbers survive, which are recomputed, and which could be emitted twice?

Watermarks are not state. They are derived from data the generator has seen, and after a restart it has seen nothing. With checkpointing off — the module 1 default — the job restarts from the source's current position with empty state (or fails, without a restart strategy) and each watermark starts over from the first timestamp read minus 20 s. With checkpointing on (module 7), sources rewind to the checkpointed offsets and the generators re-derive watermarks from the replayed records: from outside, currentInputWatermark on tenant-cost goes backwards to about the checkpoint's position minus the bound, then climbs again. Event time is monotone within one run of a generator, not across recoveries.

Windows are state and survive: restored from the checkpoint with partial sums intact (module 7). What could be emitted twice are the TenantCostMinutes fired between the checkpoint and the crash — the replayed watermark passes those window ends again, and windows still in the restored state fire again. Whether the duplicate reaches the billing system is module 8.

Try it

Go to labs/05-watermarks/. The stage-5 gatewaymeter/ runs against the compose cluster with the strategy above.

• 5.1 Run the lab's deterministic watermark test and explain why the minimum of active inputs 100000, 99000, 20000 is 20000, then 99000 when the third input becomes idle.
• 5.2 Disable idleness and use the producer's --stall-partition 5 mode after its warm-up period. Confirm from Kafka offsets that p5 stopped while other partitions kept receiving records; then record source and downstream currentInputWatermark and the last emitted window_end.
• 5.3 Submit the normal stage with idleness enabled. Record when its downstream watermark advances and which windows fire; the lab uses a 15-second idleness timeout so the change is visible quickly.
• 5.4 Resume p5 traffic with older timestamps and compare each event's timestamp to the operator's current watermark. Decide which events are late, not merely out of order.

Lab page: lab-05-watermarks.html — the exercises, every source file, and the run-it-locally steps.