How to Handle Late-Arriving Data and Out-of-Order Events in Streaming Pipelines with Kafka

Use event-time processing with watermarks to bound allowed lateness, combined with upsert sinks that issue corrections when late events arrive after window closure.

In production Kafka environments, events rarely arrive in perfect chronological order due to network latency, sensor disconnections, or producer clock skews. If your streaming job simply processes records as they arrive, you risk dropping late data or keeping windows open indefinitely, leading to incorrect aggregations and unbounded state growth. The Data Engineering Zoomcamp by DataTalksClub demonstrates a robust production pattern using Apache Flink SQL to handle late-arriving data and out-of-order events while maintaining deterministic results.

The Challenge of Disorder in Stream Processing

In a Kafka-driven streaming pipeline, the broker provides no guarantee that event timestamps align with ingestion order. A sensor might batch-upload buffered readings after reconnecting, or a mobile device might emit delayed location updates from a low-connectivity area. Without explicit handling, these late events either get assigned to the wrong time window or discarded entirely, corrupting aggregated metrics like hourly revenue totals or trip counts.

Three Techniques for Handling Late Data

The Zoomcamp workshop implements a three-pillar strategy that combines Flink’s event-time semantics with upsert capabilities to guarantee eventual correctness.

Event-Time Columns and Watermarks

Instead of using processing time (wall-clock), the job extracts the actual event timestamp from the payload and declares a watermark that trails the highest observed timestamp by a configurable delay. This watermark acts as a progress metric—when it passes a window’s end time, Flink knows it can safely emit the aggregate.

In 07-streaming/workshop/src/job/aggregation_job.py (lines 35-37), the DDL defines:

event_timestamp AS TO_TIMESTAMP_LTZ(tpep_pickup_datetime, 3),
WATERMARK FOR event_timestamp AS event_timestamp - INTERVAL '5' SECOND

The TO_TIMESTAMP_LTZ function converts the raw epoch milliseconds to a proper TIMESTAMP(3), while the watermark declaration instructs Flink to wait 5 seconds after observing the latest timestamp before considering that point in time complete.

Allowed Lateness Configuration

The interval specified in the watermark definition—here, INTERVAL '5' SECOND—determines how much disorder the pipeline tolerates. Events arriving within this 5-second buffer are still counted toward their respective windows because the window remains open until the watermark advances past it.

Upsert Sinks with Primary Keys

When an event arrives after the watermark has already passed the window boundary, the window has technically closed and its result emitted. To prevent data loss, the downstream sink must support corrections. The workshop uses a JDBC sink declared with a composite primary key:

CREATE TABLE processed_events_aggregated (
    window_start TIMESTAMP(3),
    PULocationID INT,
    num_trips    BIGINT,
    total_revenue DOUBLE,
    PRIMARY KEY (window_start, PULocationID) NOT ENFORCED
) WITH (
    'connector' = 'jdbc',
    'url'       = 'jdbc:postgresql://postgres:5432/postgres',
    'table-name'= 'processed_events_aggregated',
    'username'  = 'postgres',
    'password'  = 'postgres',
    'driver'    = 'org.postgresql.Driver'
);

This definition appears in 07-streaming/workshop/src/job/aggregation_job.py (lines 13-14). The NOT ENFORCED clause indicates Flink manages the key constraint, not PostgreSQL. When a late event triggers an updated aggregate, Flink emits an UPDATE statement rather than an INSERT, correcting the previously published result. Without this primary key, late events would be silently dropped.

Complete Flink SQL Implementation

The full job combines these elements into a cohesive pipeline. The Kafka source configuration ensures historical processing on first run while checkpointing maintains exactly-once semantics:

CREATE TABLE events (
    PULocationID      INT,
    DOLocationID      INT,
    trip_distance     DOUBLE,
    total_amount      DOUBLE,
    tpep_pickup_datetime BIGINT,
    event_timestamp   AS TO_TIMESTAMP_LTZ(tpep_pickup_datetime, 3),
    WATERMARK FOR event_timestamp AS event_timestamp - INTERVAL '5' SECOND
) WITH (
    'connector'               = 'kafka',
    'properties.bootstrap.servers' = 'redpanda:29092',
    'topic'                   = 'rides',
    'scan.startup.mode'       = 'earliest-offset',
    'properties.auto.offset.reset' = 'earliest',
    'format'                  = 'json'
);

The 'scan.startup.mode' = 'earliest-offset' setting (lines 41-42 in aggregation_job.py) ensures the job ingests all existing topic data on initial deployment, while subsequent restarts rely on Flink checkpoints to resume without duplication.

The aggregation query uses tumbling windows keyed by the event-time descriptor:

INSERT INTO processed_events_aggregated
SELECT
    window_start,
    PULocationID,
    COUNT(*)          AS num_trips,
    SUM(total_amount) AS total_revenue
FROM TABLE(
    TUMBLE(TABLE events, DESCRIPTOR(event_timestamp), INTERVAL '1' HOUR)
)
GROUP BY window_start, PULocationID;

This query appears in 07-streaming/workshop/src/job/aggregation_job.py (lines 66-74).

How the Pipeline Handles Late Events in Practice

The execution flow distinguishes between two arrival scenarios:

1. Events within the lateness buffer: As documented in 07-streaming/workshop/README.md (lines 56-84), if a record arrives within 5 seconds of the window closing, it is included in the initial aggregate because the watermark has not yet advanced. The window remains open, and the state updates silently.

2. Events after the watermark: If a record arrives after the 5-second grace period (diagram in lines 88-114 of README.md), the window has already been emitted and its state purged from memory. However, Flink forwards the event to the sink, which recognizes the primary key conflict and issues an upsert, overwriting the stale aggregate in PostgreSQL.

This mechanism guarantees that final results converge to the correct values regardless of arrival order, while the watermark ensures state for old windows can be garbage-collected to prevent memory exhaustion.

Testing with Simulated Late Events

To validate the configuration, the workshop includes 07-streaming/workshop/src/producers/producer_realtime.py, which intentionally injects disorder:

import random, time, json
from kafka import KafkaProducer
from datetime import datetime, timedelta

producer = KafkaProducer(
    bootstrap_servers="redpanda:29092",
    value_serializer=lambda v: json.dumps(v).encode("utf-8")
)

while True:
    ts = datetime.utcnow()
    # 20% chance to make the event late by 3-10 seconds

    if random.random() < 0.2:
        ts -= timedelta(seconds=random.randint(3, 10))
        label = "LATE"
    else:
        label = "ON TIME"

    record = {
        "PULocationID": random.randint(1, 265),
        "DOLocationID": random.randint(1, 265),
        "trip_distance": round(random.uniform(0.5, 25.0), 2),
        "total_amount": round(random.uniform(5.0, 80.0), 2),
        "tpep_pickup_datetime": int(ts.timestamp() * 1000)   # epoch ms

    }
    producer.send("rides", record)
    print(f"{label} -> PU={record['PULocationID']} ts={ts.isoformat()}")
    time.sleep(0.2)

Running this producer against the Flink job demonstrates that late events are either incorporated into open windows or used to update closed ones via the upsert sink.

Summary

• Event-time processing is essential when temporal ordering matters; use TO_TIMESTAMP_LTZ to extract timestamps from payloads rather than relying on ingestion time.
• Watermarks bound the memory usage of windowed operations by defining a grace period (e.g., INTERVAL '5' SECOND) for late data.
• Upsert sinks with PRIMARY KEY ... NOT ENFORCED enable correction of previously emitted results when events arrive after the watermark.
• Configure Kafka consumer offsets with earliest-offset for backfilling and rely on Flink checkpoints for failure recovery to ensure exactly-once semantics.

Frequently Asked Questions

What happens to events that arrive after the watermark has passed?

Events arriving after the watermark advances past their window end are not silently dropped. Instead, Flink forwards them to the sink, which performs an upsert operation using the defined primary key. As shown in 07-streaming/workshop/README.md (lines 88-114), this updates the previously stored aggregate in PostgreSQL, ensuring eventual correctness.

How do I choose the right watermark interval?

The interval represents a trade-off between latency and completeness. A shorter interval (e.g., 5 seconds) yields lower latency but risks missing more late data, while a longer interval (e.g., 30 seconds) improves accuracy at the cost of delayed results. Analyze your source systems' delay distribution—network telemetry from 07-streaming/workshop/src/producers/producer_realtime.py suggests most delays fall under 10 seconds for the tested scenario.

Why use NOT ENFORCED for the primary key constraint?

Flink manages the upsert logic and duplicate detection; the NOT ENFORCED clause tells Flink that the downstream database does not need to enforce uniqueness constraints. This prevents unnecessary overhead in the database while allowing Flink to generate UPDATE statements for late-event corrections, as implemented in aggregation_job.py (lines 13-14).

Can this pattern handle events that are hours or days late?

The pattern supports arbitrarily late events, but only if the watermark interval is configured to accommodate them. However, keeping windows open for hours consumes significant memory. For extreme lateness (days), consider reprocessing historical data in batch mode or using a side-output stream to capture very late events for separate handling rather than keeping streaming windows open indefinitely.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}