Time in Apache Flink: watermarks, idle partitions, state TTL and replay

In stream processing, most of the hard-to-find bugs aren't in the computation but in time: events that arrive late, a partition that goes quiet, state expiring by the wrong clock, and whether a rerun produces the same result.

This post collects what I learned about time in Apache Flink from an event analytics pipeline I never finished: events from GitHub, Jira and deploy systems, grouped into windows to compute metrics like PR review time or deploy frequency. Where data can be lost or duplicated in the same pipeline is covered in exactly-once doesn't come for free.

Event time and watermarks

A PR opened at 10:00 may have its webhook arrive at 10:03. To count "PRs opened during the 10 o'clock hour" correctly, windows must use event time (when the thing happened), not when Flink received it.

The catch is that Flink can't know when it has seen every event for the 10 o'clock hour. A watermark is how it says "from now on, I won't wait for anything older than time T". The usual setup:

WatermarkStrategy
    .<Event>forBoundedOutOfOrderness(Duration.ofMinutes(5))
    .withTimestampAssigner((e, ts) -> e.eventTime().toEpochMilli())
    .withIdleness(Duration.ofSeconds(30));

The watermark trails the newest event by 5 minutes. The 10 o'clock window only closes once the watermark passes 11:00, that is, once events from 11:05 onward have been seen. The trade-off is plain: waiting longer misses fewer late events but delivers results later.

Events that are too late are dropped, silently

allowedLateness gives a window extra time after it closes: events arriving within it update the window's result. But what about events arriving after that?

By default, Flink drops them without a word. The project's design doc said "events that are too late go to a side output and then to the DLQ", but the code never had that line. Doing it right takes a few lines:

OutputTag<Event> lateTag = new OutputTag<>("late-events") {};

var result = stream
    .keyBy(Event::teamId)
    .window(TumblingEventTimeWindows.of(Time.hours(1)))
    .allowedLateness(Time.minutes(5))
    .sideOutputLateData(lateTag)
    .aggregate(new CountAgg());

result.getSideOutput(lateTag).sinkTo(lateEventsSink);   // and count them in a metric

You don't have to process late events right away, but you do have to count them. That number tells you whether 5 minutes is enough, instead of guessing.

One quiet partition can stall the whole job

With a multi-partition Kafka source, the source's watermark is the minimum across partitions. If one partition has no data, its watermark doesn't move, and it holds the whole job back: no window ever closes.

withIdleness(30s) marks a partition "idle" after 30 seconds of silence, and Flink leaves it out of the combined watermark.

The original design doc only applied withIdleness to live mode, while replay mode (rereading history) just used ascending timestamps, since historical data is already ordered. The code has a comment correcting that: during replay only a few of the many partitions have data, the empty ones hold the watermark at Long.MIN_VALUE, and no window ever fires. The job runs, reads everything, and produces nothing. It's the kind of bug you find by running the job, not by reading the design.

Know the downside of withIdleness too: when an idle partition comes back, its first events may already be behind the combined watermark and count as late. Set the idle timeout too short and ordinary delays turn into lost data.

State TTL: three details that are easy to get wrong

State such as the set of deduplicated keys needs to expire so it doesn't grow forever. The project's config:

StateTtlConfig.newBuilder(Time.hours(24))
    .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
    .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired)
    .cleanupInRocksdbCompactFilter(1000)
    .build();

1. OnCreateAndWrite, not OnReadAndWrite. For dedup state, every event reads the state to check it. If reads extended the TTL, a key that keeps getting duplicated would never expire. Extending only on write is the intent: the key lives 24 hours from when it was first seen.

2. The cleanupInRocksdbCompactFilter parameter isn't a sampling rate. A comment in the code says "1000 means check one in every 1000 keys during compaction". In fact every key is checked; the number is how many entries are processed before the filter asks Flink for the current time again. A smaller number cleans up more precisely but slows compaction, since each query is a JNI call. Read the API docs rather than trusting comments, your own included.

3. TTL runs on wall-clock time, not event time. Per Flink's documentation, state TTL currently only supports processing time. As a result, replaying 30 days of data in an hour, 24-hour state won't expire the way it does in production. Which leads to the next section.

"Deterministic" replay, and what quietly breaks it

One of event time's most attractive promises: same input, same output, whenever and however fast you run it. That's what lets you fix the logic and rerun history to recompute metrics. The project's design doc made that promise too, on the condition of "no processing-time-dependent logic".

Rereading the code turned up three things to watch, two of which quietly break that promise:

• Processing-time timers. The pipeline computing 30-day metrics sets a timer for "the start of the next day" after each event with registerProcessingTimeTimer. In production, the timer fires once the day is over, after all its events are in. During replay, that moment is already in the past by the wall clock, so the timer fires almost immediately, when only the day's first few events have arrived. The result depends on how fast the job reads. The fix is registerEventTimeTimer: the timer fires when the watermark passes it, so fast or slow runs yield the same sequence of results.
• Instant.now() in the output. Output records carry a field taken from the wall clock. Here it's only "computed at" metadata, so it's harmless, but it's a spot to watch: the moment such a value becomes part of a key or a version, every rerun produces a different row.
• The replay flag isn't part of the key. The ClickHouse tables use ReplacingMergeTree with an is_replay column to separate rerun data from live data. But is_replay isn't in the ORDER BY. With this engine, rows with the same sorting key count as one, so on merge a replay row can replace a live row. To keep two kinds of data apart in one table, the distinguishing column must be part of the key, or the data must go in a separate table.

The rule I took away: inside any operator, only use time that comes from the event or the watermark. Any call to the system clock needs a clear reason.

Bonus: after hours for whom?

The same pipeline computes an "after-hours work" signal in UTC:

int hour = e.eventTime().atZone(ZoneOffset.UTC).getHour();
return hour < 9 || hour >= 18;

For someone in Hanoi (UTC+7), 9:00–18:00 UTC is 16:00 to 1:00 at night. A whole normal morning of work counts as "after hours". Event time tells you when something happened; "office hours" belong to each person, and must be computed in their time zone. It's the same kind of trap as the Vietnamese lunar calendar using UTC+7: an instant is absolute, but "day" and "hour" are not.

Summary

• Window on event time, with a watermark trailing the newest event by just enough.
• Always sideOutputLateData and count late events; by default Flink drops them silently.
• withIdleness is needed in both live and replay modes, and shouldn't be too short.
• State TTL: extend on write for dedup state, read what cleanup parameters actually mean, and remember TTL runs on the wall clock.
• For deterministic replay: event-time timers, no system clock in the logic, and distinguishing flags belong in the key.

Time in stream processing has many more corners I haven't hit yet. If you have good experience to share, I'd love to compare notes. My email is on the home page.