Event Time Windows Watermarks and Out-of-Order Events

Published by Programming Line in Data Engineering in Depth. Last updated 7th September 2026.

Purpose and Outcome

Streaming data rarely arrives in the same order in which the business activity happened. A mobile app can buffer clicks while offline, an IoT gateway can reconnect after a network gap, and Kafka partitions can preserve order per key while still interleaving many keys at different speeds. Event-time windows solve the data engineering problem behind those situations: compute results by the timestamp inside the event, not by the instant the processor happened to read it.

In this lesson, the outcome is practical: you should be able to choose a window type, understand what a watermark promises, reason about out-of-order events, and debug why a streaming aggregation emitted too early, emitted too late, or never emitted at all. In a Kafka-centered pipeline, these ideas sit between ingestion and serving. Kafka stores the ordered log, but the stream processor, such as Kafka Streams, Flink, Spark Structured Streaming, or Beam, decides when a window is complete enough to publish.

How Event Time Works Internally

A streaming record has at least two useful clocks. Event time is when the source says the event happened, often a field such as created_at or sensor_ts. Processing time is when the processing task handles the record. Some systems also track ingestion time, the moment the broker or pipeline accepted the event. Event time is normally the correct clock for metrics such as orders per hour, page views per session, or average temperature per five minutes because those questions describe the real world, not the scheduler.

A window groups event-time instants into bounded buckets. A tumbling window has fixed, non-overlapping buckets, for example 12:00:00 through 12:00:59 and then 12:01:00 through 12:01:59. A hopping or sliding window overlaps, so one event can update several buckets. A session window closes after a period of inactivity and is common for user behavior. Internally, a windowed aggregation usually stores state under a compound key: the grouping key plus the window start and end. A count of purchases by region for a one-minute tumbling window is not just keyed by region; it is keyed by (region, window_start, window_end).

The hard part is deciding when to emit the result. If the processor emits as soon as wall clock time passes 12:01, a delayed 12:00:45 event may be excluded. If it waits forever, state grows without bound and downstream tables never stabilize. A watermark is the processor’s moving estimate that event times at or before a threshold are complete enough to close. A common bounded-out-of-orderness strategy is watermark = maximum_event_time_seen - delay. If the maximum observed event time is 12:05:30 and the delay is two minutes, the watermark is 12:03:30. Windows ending at or before 12:03:30 can fire.

A watermark is not a wall clock and it is not a guarantee that no older event can ever appear. It is a correctness and latency choice made by the pipeline. Larger delays capture more late data but keep state longer and increase result latency. Smaller delays publish quickly but push more records into late-data handling. Systems differ in details: some emit append-only final results, some emit update records for late changes, and some allow both an on-time firing and later correction firings.

API and Configuration Anatomy

Most stream processors expose the same conceptual knobs even when their syntax differs. First, a timestamp extractor tells the system which field is event time and what to do with missing or malformed timestamps. Second, a window assigner maps the timestamp to one or more windows. Third, a watermark strategy describes expected disorder, often as a fixed delay, a source-provided watermark, or a custom generator. Fourth, an allowed-lateness or grace setting controls whether records that arrive after the first close may still update the result. Finally, a trigger or output mode decides whether results are emitted once, updated, or retracted.

For example, a Kafka Streams topology might group records by customer, apply a one-minute time window, and set a grace period for late arrivals. Flink and Beam usually make the watermark strategy more explicit. Spark Structured Streaming uses event-time columns and watermark declarations before aggregating. The vocabulary changes, but the shape is stable: timestamp extraction, window assignment, watermark advancement, state retention, and output semantics.

Worked Example 1: Processing Time Versus Event Time

The following small program shows why processing-time windows are often wrong for business metrics. The second event happened at second 65, but it arrived before an event that happened at second 20. Processing time puts all three records in the first arrival minute. Event time puts two records in minute 0 and one record in minute 60.

from collections import defaultdict

events = [
    {"id": "a", "event_time": 3, "arrival_time": 1, "amount": 5},
    {"id": "b", "event_time": 65, "arrival_time": 2, "amount": 7},
    {"id": "c", "event_time": 20, "arrival_time": 3, "amount": 11},
]

def window_start(ts, size=60):
    return (ts // size) * size

processing_time_totals = defaultdict(int)
event_time_totals = defaultdict(int)
for e in events:
    processing_time_totals[window_start(e["arrival_time"])] += e["amount"]
    event_time_totals[window_start(e["event_time"])] += e["amount"]

print("processing time", dict(sorted(processing_time_totals.items())))
print("event time", dict(sorted(event_time_totals.items())))

The deterministic output is processing time {0: 23} and event time {0: 16, 60: 7}. The event-time result matches the question: what amount occurred in each event minute?

Worked Example 2: Closing Windows with a Watermark

This example adds a watermark with 30 seconds of tolerated out-of-orderness. The processor keeps a maximum event time seen so far. When event d advances the maximum to 130, the watermark becomes 100, so the window ending at 60 can safely emit.

events = [
    {"id": "a", "event_time": 3, "arrival_time": 1, "amount": 5},
    {"id": "b", "event_time": 65, "arrival_time": 2, "amount": 7},
    {"id": "c", "event_time": 20, "arrival_time": 3, "amount": 11},
    {"id": "d", "event_time": 130, "arrival_time": 4, "amount": 2},
]

max_seen_event_time = float("-inf")
allowed_lateness = 30
open_windows = {}

def assign_window(ts, size=60):
    start = (ts // size) * size
    return (start, start + size)

for event in events:
    max_seen_event_time = max(max_seen_event_time, event["event_time"])
    watermark = max_seen_event_time - allowed_lateness
    w = assign_window(event["event_time"])
    open_windows[w] = open_windows.get(w, 0) + event["amount"]
    ready = [key for key in open_windows if key[1] <= watermark]
    for key in sorted(ready):
        print(f"emit {key}: {open_windows.pop(key)} at watermark {watermark}")

The expected output is emit (0, 60): 16 at watermark 100. Notice that event b did not force the first window to close because the watermark after b was only 35. Event c could still arrive and be included.

Worked Example 3: Late Events and Side Output

After a window has fired, a pipeline needs a policy for records that belong to it. Some systems update the aggregate if the record is within a grace period. Others route it to a dead-letter topic, audit table, or replay queue. This example accepts an out-of-order event for a still-open window, but sends a record for a finalized window to side output.

events = [
    {"id": "a", "event_time": 3, "arrival_time": 1, "amount": 5},
    {"id": "b", "event_time": 65, "arrival_time": 2, "amount": 7},
    {"id": "c", "event_time": 20, "arrival_time": 3, "amount": 11},
    {"id": "d", "event_time": 130, "arrival_time": 4, "amount": 2},
    {"id": "late-still-open", "event_time": 65, "arrival_time": 5, "amount": 4},
    {"id": "too-late", "event_time": 10, "arrival_time": 6, "amount": 100},
]

max_seen_event_time = float("-inf")
out_of_orderness = 30
allowed_lateness_after_close = 15
windows = {}
closed_until = float("-inf")
side_output = []

for event in events:
    max_seen_event_time = max(max_seen_event_time, event["event_time"])
    watermark = max_seen_event_time - out_of_orderness
    final_deadline = event["event_time"] + allowed_lateness_after_close
    if final_deadline <= closed_until:
        side_output.append(event["id"])
        continue
    start = (event["event_time"] // 60) * 60
    key = (start, start + 60)
    windows[key] = windows.get(key, 0) + event["amount"]
    for window in sorted(list(windows)):
        if window[1] <= watermark:
            print(f"upsert {window}: {windows[window]}")
            closed_until = max(closed_until, window[1] + allowed_lateness_after_close)

print("side output", side_output)

The expected output includes upsert (0, 60): 16 and side output ['too-late']. The event named late-still-open updates the 60 to 120 window because that window has not reached its final deadline. The event named too-late is not silently mixed into a closed report. In a real pipeline, the side output should be observable and replayable.

Design Choices and Trade-offs

Choose event time when the business question depends on when events occurred. Choose processing time only for operational questions such as records processed per minute or alerting on worker throughput. Use tumbling windows for non-overlapping summaries, hopping windows for rolling metrics, and session windows for behavior separated by inactivity. The choice affects state size: a hopping window can multiply writes because one event belongs to several windows.

Watermark delay is the central trade-off. A five-second delay gives low latency but may miss mobile and cross-region data. A one-hour delay improves completeness but holds state, delays final output, and can increase checkpoint cost. Partitioning also matters. If watermarks are computed per partition, an idle Kafka partition can hold the global watermark back unless the framework supports idleness detection. If watermarks are computed globally without regard for partition skew, one fast partition may make slow partitions look late.

Output mode affects downstream design. Append-only final windows are simple for files and warehouses. Upsert outputs need a primary key containing the grouping key and window boundaries. Retractions or correction records require consumers that understand updates. For analytics tables, the most reliable pattern is usually an upsert sink keyed by window plus dimension, with a separate column showing whether the row is provisional or final.

Failure Modes and Troubleshooting

Symptom: windows never emit. The likely cause is a watermark that is not advancing. Check source partitions for idleness, timestamp extraction errors, and metrics such as current watermark, maximum event time, and records consumed per partition. Correct by enabling idle-partition handling, fixing timestamp parsing, or ensuring each source emits watermarks.

Symptom: many records are marked late. The likely cause is a watermark delay that is too aggressive or event timestamps that are assigned from ingestion time at one source and device time at another. Compare event time to arrival time in a sampled topic, grouped by producer and partition. Correct by normalizing timestamp fields, increasing bounded out-of-orderness, or routing known slow producers through a separate stream with its own watermark policy.

Symptom: downstream totals change after users saw a report. The cause is late updates emitted as upserts or retractions. Inspect sink keys and output mode, then check whether the dashboard reads provisional rows as final. Correct by exposing window finality, delaying publication until the watermark passes the final deadline, or designing the sink to show corrections explicitly.

Symptom: state storage grows unexpectedly. The cause is long windows, large grace periods, high-cardinality grouping keys, or watermarks stuck behind idle partitions. Diagnose state-store size, checkpoint duration, and key cardinality. Correct by reducing grace where acceptable, using pre-aggregation, handling idleness, or moving rare dimensions out of the streaming aggregation.

Reliability and Performance Implications

Event-time processing is reliable only when timestamps are trustworthy. Producers should use a clear timestamp field, time zone convention, and serialization type. Prefer epoch milliseconds or an unambiguous timestamp with offset. Validate that timestamps are within plausible bounds before they enter core aggregates. A record from year 2099 can push a watermark far ahead and make normal records appear late; a record from 1970 can keep state around if accepted into old windows.

Performance depends on state retention and firing frequency. Every open window consumes memory or local state-store space, and every correction increases sink writes. Checkpointed engines also pay for larger state during recovery. Reliability improves when late records are not dropped blindly: route them to a side topic with original payload, parsed timestamp, observed watermark, and reason. That side topic supports audits and controlled replay.

Hands-on Lab

Prerequisites: Python 3 and a terminal are enough for the local simulation. No Kafka cluster is required because the lab isolates window and watermark mechanics.

Steps: Create a temporary file named watermark_lab.py. Paste the code from Worked Example 2. Run python3 watermark_lab.py and confirm that only the (0, 60) window emits after the event at time 130. Then change allowed_lateness from 30 to 80 and rerun it. The first window should not emit at event time 130 because the watermark becomes 50, which is before the window end of 60. Finally, append a new event with event_time 200. The first window should emit when the watermark advances beyond 60.

Verification: Record the watermark calculation for each input row and compare it with emitted windows. A window may emit only when window_end <= watermark. If your output differs, print max_seen_event_time, watermark, and open_windows after each event.

Cleanup: Delete the temporary file after the lab. If you adapted the logic in a real stream processor, roll back experimental grace or watermark changes before deploying; those settings affect state retention and visible result timing.

Assessment Exercises

A stream has a maximum observed event time of 10:15:00 and a bounded out-of-orderness delay of three minutes. Which one-minute tumbling windows may fire, and which must remain open?
You receive complaints that yesterday’s hourly revenue changed after publication. Explain whether the cause is likely processing time, event time with late upserts, or duplicate input, and list the metrics you would inspect first.
Design a key for an upsert sink that stores counts by customer segment in five-minute windows. Include the fields required to avoid overwriting the wrong window.
A Kafka topic has one idle partition and three active partitions. Explain how that can affect the global watermark and name one correction that does not drop data.
Given a mobile producer that can be offline for 20 minutes, argue for or against a five-minute watermark delay. Include the downstream latency and completeness trade-off.

Summary

Event-time windows let a streaming system answer questions according to when events happened. Watermarks provide the moving threshold that turns unbounded input into bounded, publishable windows. Out-of-order events are normal, so the pipeline must choose a lateness policy, state retention period, and output mode deliberately. The practical engineering work is measuring disorder, setting watermarks from evidence, and making late-data behavior explicit instead of accidental.