Definition
- WatermarkStrategy interface – A factory that supplies two collaborators: a
TimestampAssignerto extract event time from each record and aWatermarkGeneratorto advance the watermark. - TimestampAssigner – Pulls the event-time millis from each record, e.g.
(event, ts) -> event.getEventTimeMillis(). - Periodic generator – Emits watermarks on a fixed interval via
onPeriodicEmit, using max-seen minus a delay; the built-inforBoundedOutOfOrdernessis a periodic generator. - Punctuated generator – Emits watermarks in response to specific marker events inside
onEvent, useful when the source has explicit batch-completion markers. - Out-of-orderness trade-off – Too high adds latency and holds more state; too low drops legitimate late events, understating results.
The National Bank of India needs a rule for when to declare a business minute closed. The timestamp assigner is the clerk who reads the true time off each transaction slip. The watermark generator is the policy that says "announce the minute closed once we are confident all slips for it have arrived." If the policy waits a very long time, the report is late. If it closes too eagerly, slips that were just a few seconds behind get thrown away. A custom strategy lets the bank write its own closing rule.
Understanding the Concepts
What is a custom watermark strategy?
A WatermarkStrategy in Flink is a factory for two collaborators: a TimestampAssigner that extracts event time from each record, and a WatermarkGenerator that decides when to advance the watermark. You implement a custom one when the built-in strategies do not fit your data.
import org.apache.flink.api.common.eventtime.*;
import java.time.Duration;
// Built-in strategy (most common)
WatermarkStrategy<Transaction> builtIn =
WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(
Duration.ofSeconds(5))
.withTimestampAssigner(
(txn, ts) -> txn.getEventTimeMillis());
// Custom strategy when built-in is not enough
WatermarkStrategy<Transaction> custom =
new WatermarkStrategy<Transaction>() {
@Override
public TimestampAssigner<Transaction> createTimestampAssigner(
TimestampAssignerSupplier.Context ctx) {
return (txn, ts) -> txn.getEventTimeMillis();
}
@Override
public WatermarkGenerator<Transaction> createWatermarkGenerator(
WatermarkGeneratorSupplier.Context ctx) {
return new CustomWatermarkGenerator();
}
};
How does a periodic watermark generator work?
A periodic generator emits watermarks on a fixed interval controlled by env.getConfig().setAutoWatermarkInterval(...). The onEvent callback tracks state, and onPeriodicEmit emits the watermark.
public class BoundedOutOfOrderGenerator
implements WatermarkGenerator<Transaction> {
private long maxTimestamp = Long.MIN_VALUE;
private final long maxOutOfOrdernessMillis;
public BoundedOutOfOrderGenerator(long maxOutOfOrdernessMillis) {
this.maxOutOfOrdernessMillis = maxOutOfOrdernessMillis;
}
@Override
public void onEvent(Transaction txn, long eventTimestamp,
WatermarkOutput output) {
// Track the maximum event time seen
maxTimestamp = Math.max(maxTimestamp, eventTimestamp);
}
@Override
public void onPeriodicEmit(WatermarkOutput output) {
// Emit watermark = max_seen - out_of_orderness
output.emitWatermark(
new Watermark(maxTimestamp - maxOutOfOrdernessMillis));
}
}
Auto watermark interval: 200ms (default)
onEvent: tracks maxTimestamp
onPeriodicEmit (every 200ms): emits Watermark(maxTimestamp - 5000)
How does a punctuated watermark generator work?
A punctuated generator emits a watermark inside onEvent when it sees a specific marker record. This is useful when your source has explicit completeness markers like an end-of-batch record.
public class PunctuatedGenerator
implements WatermarkGenerator<Transaction> {
@Override
public void onEvent(Transaction txn, long eventTimestamp,
WatermarkOutput output) {
// Emit watermark when we see an end-of-batch marker
if (txn.isEndOfBatch()) {
output.emitWatermark(new Watermark(eventTimestamp));
}
}
@Override
public void onPeriodicEmit(WatermarkOutput output) {
// No periodic emission -- watermarks are purely punctuated
}
}
Stream: T1 T2 T3 [END_BATCH ts=10:05] T4 T5 [END_BATCH ts=10:10]
▲ ▲
Watermark(10:05) Watermark(10:10)
What happens when out-of-orderness is too high?
A large out-of-orderness delay means the watermark trails far behind the latest event time. Every event-time window waits that much longer before firing, increasing end-to-end latency and holding more open windows in memory.
// Too high: 60 seconds of out-of-orderness
WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(
Duration.ofSeconds(60));
// Watermark = max_seen - 60s
// Every 1-minute window waits an extra 60s before firing
// 2x the windows open in memory at any time
// End-to-end latency increases by 60 seconds
What happens when out-of-orderness is too low?
A zero or very small delay means the watermark hugs the latest event time. Any slightly-delayed event arrives after the watermark has passed its window and is treated as late, silently understating results.
// Too low: 0 seconds (monotonous timestamps)
WatermarkStrategy.<Transaction>forMonotonousTimestamps();
// Watermark = max_seen - 0
// An event delayed by even 1 millisecond is dropped as late
// Settlement totals may undercount transactions
Out-of-orderness too LOW (0s): Event at 10:00:30, arrives at 10:00:31 Watermark already at 10:00:31 (from another event) 10:00:30 < 10:00:31 → DROPPED! Total is wrong. Out-of-orderness correct (5s): Event at 10:00:30, arrives at 10:00:31 Watermark at 10:00:26 (max 10:00:31 - 5s) 10:00:30 > 10:00:26 → ACCEPTED. Total is correct.
How do you choose the right out-of-orderness?
Measure the actual lateness distribution of your source and set the out-of-orderness to cover roughly the 99th percentile of observed delay. Handle the rare tail with allowed lateness plus a side output.
// Step 1: Measure your data's lateness distribution
// ingestion_time - event_time for each record
// P50 = 200ms, P95 = 2s, P99 = 5s, P99.9 = 30s
// Step 2: Set out-of-orderness to cover P99
WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(
Duration.ofSeconds(5)); // covers 99% of late events
// Step 3: Handle the 1% tail with allowed lateness + side output
stream.keyBy(txn -> txn.getAccountId())
.window(TumblingEventTimeWindows.of(Time.minutes(1)))
.allowedLateness(Time.seconds(30)) // covers P99.9
.sideOutputLateData(lateTag); // capture the rest
How do you handle idle sources?
If a Kafka partition stops producing, its watermark stops advancing and holds back the global watermark, freezing all windows. Use withIdleness to mark idle sources so they do not block progress.
WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(
Duration.ofSeconds(5))
.withTimestampAssigner((txn, ts) -> txn.getEventTimeMillis())
.withIdleness(Duration.ofMinutes(1)); // mark idle after 1 min
// If a partition produces nothing for 1 minute,
// it is excluded from watermark calculation
// so other partitions can still advance the global watermark
What are common mistakes?
The most common mistake is guessing the out-of-orderness instead of measuring it. Another mistake is forgetting withIdleness, causing a single idle partition to freeze all windows indefinitely.
// WRONG -- arbitrary guess, too tight
WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(
Duration.ofMillis(100)); // Guess: most events arrive in 100ms
// Reality: mobile network delays cause 5s gaps → dropped events
// CORRECT -- measured from real data
WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(
Duration.ofSeconds(5)); // Measured: P99 lateness = 5s
What should a senior engineer know?
The professional approach is not to guess but to measure the real lateness distribution, set the out-of-orderness to the 99th percentile, and handle the tail with allowed lateness and a side output. Use punctuated watermarks when the source has explicit batch-completion markers for precise time advancement without delay-based guessing.
Visual Overview
Custom Watermark Strategy Components ┌──────────────────────────────────────────┐ │ WatermarkStrategy<Transaction> │ ├──────────────────────────────────────────┤ │ │ │ createTimestampAssigner() │ │ ┌──────────────────────────────────┐ │ │ │ TimestampAssigner │ │ │ │ (txn, ts) -> txn.getEventTime() │ │ │ └──────────────────────────────────┘ │ │ │ │ createWatermarkGenerator() │ │ ┌──────────────────────────────────┐ │ │ │ WatermarkGenerator │ │ │ │ onEvent: track maxTimestamp │ │ │ │ onPeriodicEmit: emit WM │ │ │ └──────────────────────────────────┘ │ └──────────────────────────────────────────┘ Out-of-orderness spectrum: ◀── EAGER (low delay) RELAXED (high delay) ──▶ Fast results Complete results Drops late events Catches late events Low memory High memory Use for: alerts Use for: settlements
Key Takeaway
A custom watermark strategy in Flink is how you take control of event-time progress when the built-in strategies are not enough. You implement the WatermarkStrategy interface, which is a factory for two collaborators: a timestamp assigner that pulls the event-time timestamp from each record, and a watermark generator that decides when to advance the watermark.
A periodic generator emits the watermark on a fixed interval, which is how the built-in bounded-out-of-orderness strategy works. A punctuated generator emits a watermark directly inside onEvent when it sees a special marker record, which is ideal when your source has explicit completeness markers like an end-of-batch record, because you can advance time exactly when the data guarantees a batch is finished.
The out-of-orderness setting is the dial everyone must understand. Too high means every window waits longer, increasing latency and memory. Too low means slightly-delayed events are dropped, silently understating results. The professional approach is to measure the real lateness distribution, set the out-of-orderness to the 99th percentile, and handle the rare tail with allowed lateness and a side output.
At the National Bank of India the nightly settlement feed arrived in explicit batches, each ending with an end-of-batch marker, so Rajesh wrote a punctuated custom strategy that advanced the watermark the instant the marker arrived, removing the latency of a fixed time delay while guaranteeing completeness. He also protected against idle partitions with withIdleness so a stalled source did not freeze all windows.
Remember one sentence: The watermark strategy pairs a timestamp extractor with a generator that advances the watermark, and the out-of-orderness dial trades latency against completeness, so measure your data's real delay rather than guessing.
---