Flink event-time checklist: stalled watermarks, idle Kafka partitions, late data and watermark alignment

The job is running, checkpoints succeed, records flow in, and no window results ever come out. Or results come out hours late, or a backfill fills state until TaskManagers die. In event-time Flink jobs these usually have one cause: the watermark is not where you think it is.

Use this checklist when building a new event-time job and when triaging one that has stopped producing output.

#Know how the watermark moves

• Each source subtask generates a watermark from the timestamps it has seen, minus the out-of-orderness bound you configured.
• Every operator with multiple inputs takes the minimum watermark across its inputs. One slow or silent input holds back the whole job.
• Windows and event-time timers fire only when the watermark passes their end time.
• Periodic watermarks are emitted every pipeline.auto-watermark-interval (default 200 ms). Setting it to 0 disables periodic emission.

#1. See the watermark before changing anything

• Open the job in the Web UI, select each operator and check the Watermarks tab per subtask. A subtask showing "No Watermark" or a value far behind the others is your blocker.
• Pull the same data from the REST API so you can script it:

JM=http://flink-jobmanager:8081
JOB=$(curl -s $JM/jobs | jq -r '.jobs[] | select(.status=="RUNNING") | .id' | head -1)
for V in $(curl -s $JM/jobs/$JOB | jq -r '.vertices[].id'); do
  echo "== $V"
  curl -s $JM/jobs/$JOB/vertices/$V/watermarks | jq -r '.[] | "\(.id) \(.value)"'
done

• Convert values to timestamps. A watermark of -9223372036854775808 (Long.MIN_VALUE) means that subtask has never emitted one.
• Check operator metrics currentInputWatermark and currentOutputWatermark in your metrics system and alert on event-time lag: wall clock minus watermark.

#2. Timestamps and the out-of-orderness bound

• Timestamps are extracted from the event, not from Kafka record time, unless you mean to use broker time.
• Units are right. Seconds treated as milliseconds puts the watermark in January 1970 and nothing fires; milliseconds treated as seconds puts it in the far future and everything is late.
• The bound in forBoundedOutOfOrderness matches measured disorder. Too small and records are dropped as late; too large and every result is delayed by the bound.
• One bad record with a far-future timestamp can jump the watermark forward and make everything after it late. Validate or clamp timestamps at the source.

WatermarkStrategy<Order> wm = WatermarkStrategy
    .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(30))
    .withTimestampAssigner((o, kafkaTs) -> o.getEventTimeMillis())
    .withIdleness(Duration.ofMinutes(2));

DataStream<Order> orders =
    env.fromSource(kafkaSource, wm, "orders-kafka");

Assign the strategy on the source with fromSource, not with assignTimestampsAndWatermarks downstream. Source-level watermarks are tracked per Kafka partition, so ordering within a partition is respected and partitions do not mix before the watermark is computed.

#3. Idle partitions and idle subtasks

The most common reason windows never fire.

• Some Kafka partitions receive no data (low-volume keys, a producer that stopped, a topic with more partitions than active keys). The minimum across partitions never advances.
• Source parallelism is higher than the number of partitions, leaving subtasks with no partitions. Depending on connector version these may hold back the watermark. Keep source parallelism at or below the partition count.
• withIdleness(...) is set so silent partitions are excluded after the timeout.
• In Flink SQL the equivalent is a table option:

SET 'table.exec.source.idle-timeout' = '2 min';

CREATE TABLE orders (
  order_id STRING,
  amount   DECIMAL(18,2),
  ts       TIMESTAMP_LTZ(3),
  WATERMARK FOR ts AS ts - INTERVAL '30' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'broker1:9092',
  'properties.group.id' = 'flink-orders',
  'scan.startup.mode' = 'group-offsets',
  'format' = 'json'
);

Warning

Idleness is a trade-off, not a fix. When an idle partition wakes up, its records may already be behind the watermark and get dropped as late. If a partition is idle because a producer broke, fix the producer; do not hide it with a short idle timeout.

• If you combine idleness with watermark alignment on a FLIP-27 source, test the backfill path. There are reports of all splits being marked idle in this combination on some versions; check the open Flink issues for your release.

#4. Late data handling

• Decide explicitly what happens to late records. By default window operators drop them silently.
• Use allowedLateness to keep window state a little longer and emit updated results, and sideOutputLateData to capture what is still too late:

OutputTag<Order> late = new OutputTag<Order>("late-orders"){};

SingleOutputStreamOperator<Summary> result = orders
    .keyBy(Order::getStoreId)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(5)))
    .allowedLateness(Duration.ofMinutes(10))
    .sideOutputLateData(late)
    .aggregate(new SumAmount());

result.getSideOutput(late).sinkTo(lateSink);

Method signatures taking Duration versus Time depend on Flink version; older releases use Time.minutes(5).

• Sinks downstream of allowedLateness must accept updates for the same window (upsert by window key), or you get duplicates.
• Monitor the operator metric numLateRecordsDropped and the side-output volume.

#5. Backfills and replays: watermark alignment

When a job restarts from an old offset, partitions catch up at different speeds. Fast partitions push their part of the stream far ahead while the global watermark is held by the slow ones, so window state for the "future" accumulates until memory or RocksDB disk runs out.

• Enable alignment so sources pause partitions or subtasks that run too far ahead:

WatermarkStrategy<Order> wm = WatermarkStrategy
    .<Order>forBoundedOutOfOrderness(Duration.ofSeconds(30))
    .withTimestampAssigner((o, ts) -> o.getEventTimeMillis())
    .withWatermarkAlignment("orders-group",
                            Duration.ofMinutes(5),    // max drift
                            Duration.ofSeconds(1));   // update interval

• Split-level alignment (per partition inside one subtask) needs Flink 1.17+ and a source that supports it; KafkaSource does. Check your connector release notes.
• Sources in the same alignment group across different topics stay within the drift of each other, which also helps joins between streams.
• During a large replay, watch checkpoint size and state backend disk. If they grow without bound, alignment is not working or the drift is too large.

#6. Joins, unions and multi-input operators

• Interval joins and two-input operators use the minimum watermark of both inputs. A low-volume side (dimension updates, a control stream) with no events holds the join. Apply idleness to that source too, or use a processing-time pattern for it.
• After union, the watermark is the minimum across all unioned inputs.
• Changing parallelism between operators (rebalance, keyBy) does not lose watermarks, but each downstream subtask still waits for all upstream subtasks.

#7. Before go-live

• Run a replay test from an old offset with production-like volume and measure state size, checkpoint duration and time to catch up.
• Alert on event-time lag per job, not just consumer lag. Consumer lag can be zero while the watermark is stuck.
• Document the out-of-orderness bound, idle timeout and allowed lateness with the reasoning, so the next person does not "fix" them blind.

#Symptom map

#Prevent it next time

Put the watermark checks into your job template: strategy on the source, idleness and alignment configured, late-data side output wired to a sink, and an event-time lag alert created with the job.