DE Mastery GitHub

Apache Flink · FLK-03

Time & windows

From your world. late-arriving facts and the as-of grain in your SCD2 / event marts

← All exercises

A watermark strategy that survives an idle partition

A global rollup that runs fine in testing can stall in production on the first quiet night. One partition goes silent, the watermark minimum freezes, and every window stops firing. The fix is a watermark strategy that tolerates out-of-order events, reads event time from the data, and declares idleness.

Task

Implement de.flk.IdleWatermark.strategy(long outOfOrdernessMs, long idleMs) to return a WatermarkStrategy<Event> that does these three things:

  1. It tolerates bounded out-of-orderness of outOfOrdernessMs (WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofMillis(outOfOrdernessMs))).
  2. It assigns the event timestamp from Event.tsMs (.withTimestampAssigner((e, ts) -> e.tsMs)).
  3. It marks a partition idle after idleMs of no events (.withIdleness(Duration.ofMillis(idleMs))).

What the tests check

The tests drive the watermark generator of the strategy with a controllable clock (no sleep, no wall-clock):

  • After two events at event times 100 and 200, the generator emits a watermark of 200 − outOfOrdernessMs. This proves that the timestamp assigner reads Event.tsMs (a missing assigner would never produce 195).
  • After a silent gap longer than idleMs, the generator marks the partition idle. A strategy without withIdleness never does this.

Hints

  • withTimestampAssigner and withIdleness each return a new composed WatermarkStrategy. Chain them and return the final one.
  • forBoundedOutOfOrderness(d) emits (max event timestamp seen − d).
  • Idleness stops a quiet partition from holding the watermark minimum.

Starter code

src/main/java/de/flk/IdleWatermark.java
package de.flk;

import de.flk.harness.Event;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;

/**
 * FLK-03 starter — NOT implemented yet (the tests must fail: G1).
 *
 * Return a WatermarkStrategy for the event stream that tolerates bounded
 * out-of-orderness, assigns event time from Event.tsMs, and marks a quiet
 * partition idle after idleMs so it does not freeze the watermark minimum.
 */
public class IdleWatermark {

    public static WatermarkStrategy<Event> strategy(long outOfOrdernessMs, long idleMs) {
        // TODO: forBoundedOutOfOrderness(..).withTimestampAssigner(..).withIdleness(..)
        return WatermarkStrategy.noWatermarks();
    }
}

Hints

Hint 1

Start from WatermarkStrategy.forBoundedOutOfOrderness(Duration.ofMillis(outOfOrdernessMs)) and attach a timestamp assigner that returns event.tsMs.

Hint 2

Add .withIdleness(Duration.ofMillis(idleMs)) so a partition that produces no events for idleMs is marked idle and excluded from the watermark minimum.

Hint 3

The order matters: withTimestampAssigner then withIdleness both return a composed WatermarkStrategy; return the final one.

Run it locally
uv run de start FLK-03/02-idle-watermark
# edit workspace/FLK-03/02-idle-watermark/
uv run de test