How does Flink SQL handle late data in windowed aggregations

• Late data – An event whose event-time timestamp falls inside a window but physically arrives after the watermark has already passed that window's end.
• Default behavior – Flink SQL drops late events because the window has already been computed and closed.
• Watermark as primary control – The watermark delay (out-of-orderness) absorbs normal disorder; sizing it correctly is the first line of defense.
• Allowed lateness – Keeps the window state alive for an extra grace period so late events update the already-emitted result, producing retractions.
• Side output – At the DataStream level, routes events that are late even beyond allowed lateness into a separate stream for reconciliation.

What is late data in Flink?

Late data is an event that belongs to a window by its event time but physically arrives after the watermark has already passed that window's end. The window has been computed, and the result has gone downstream.

-- Window [10:00, 10:01) fires when watermark reaches 10:01
-- Event at 10:00:30 arrives when watermark is already 10:01:05
-- This event is LATE for window [10:00, 10:01)

Timeline:
  Watermark: ─── 10:00:50 ─── 10:01:00 ─── 10:01:05 ───▶
                                  ▲
                     Window [10:00,10:01) FIRES

  Late event arrives: eventTime=10:00:30, arrivalTime=10:01:05
  Watermark (10:01:05) > window_end (10:01:00) → LATE!
  Default: DROPPED

How does the watermark control lateness?

The watermark delay is the first and most important tool. By declaring WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND, you absorb up to ten seconds of out-of-orderness before firing. Events later than that are dropped.

CREATE TABLE nbi_transactions (
    transaction_id STRING,
    account_id     STRING,
    amount         DECIMAL(15,2),
    event_time     TIMESTAMP(3),
    WATERMARK FOR event_time AS event_time - INTERVAL '10' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'nbi-transactions',
    'properties.bootstrap.servers' = 'broker1:9092',
    'format' = 'json'
);

-- Watermark absorbs 10 seconds of delay
-- Events up to 10 seconds late are processed normally
-- Events more than 10 seconds late are dropped

What is the default behavior in Flink SQL?

In pure Flink SQL, the default behavior is to silently drop late events. The window has already fired, and reopening it is not free.

-- Windowed aggregation: late events dropped silently
SELECT
    window_start, window_end, account_id,
    SUM(amount) AS total
FROM TABLE(
    TUMBLE(TABLE nbi_transactions,
           DESCRIPTOR(event_time),
           INTERVAL '1' MINUTE))
GROUP BY window_start, window_end, account_id;

-- Event at 10:00:30, watermark at 10:01:10
-- Window [10:00, 10:01) already fired → event dropped
-- No error, no log, no side output in pure SQL

How does allowed lateness work?

Allowed lateness keeps the window state alive for a grace period after it fires. Late events within the grace period trigger a recomputation and emit an updated result, producing a retraction and a corrected row downstream.

import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;

// DataStream API: allowed lateness
OutputTag<Transaction> lateTag = new OutputTag<>("late-txns") {};

SingleOutputStreamOperator<String> result = txnStream
    .assignTimestampsAndWatermarks(
        WatermarkStrategy.<Transaction>forBoundedOutOfOrderness(
                Duration.ofSeconds(10))
            .withTimestampAssigner(
                (txn, ts) -> txn.getEventTimeMillis()))
    .keyBy(txn -> txn.getAccountId())
    .window(TumblingEventTimeWindows.of(Time.minutes(1)))
    .allowedLateness(Time.minutes(5))     // keep window open 5 extra min
    .sideOutputLateData(lateTag)           // capture very late events
    .apply(windowFunction);

// Late events within 5 minutes → window re-fires, corrected result
// Late events beyond 5 minutes → side output
DataStream<Transaction> lateEvents = result.getSideOutput(lateTag);

Allowed lateness flow:
  Window [10:00, 10:01) fires at watermark 10:01:10
  Result: total = 500,000

  Late event at 10:00:45 arrives at 10:03:00 (within 5-min grace)
  Window re-fires: total = 550,000 (retraction + correction)

  Very late event at 10:00:20 arrives at 10:08:00 (beyond 5-min grace)
  → Routed to side output for reconciliation

What is a side output?

A side output captures events that are late even beyond the allowed lateness, routing them to a separate stream for manual review or reconciliation rather than losing them.

// Capture side output for reconciliation
DataStream<Transaction> lateEvents = result.getSideOutput(lateTag);

// Write late events to a reconciliation table
lateEvents.addSink(new JdbcSink<>(
    "INSERT INTO nbi_late_transactions VALUES (?, ?, ?, ?)",
    new TransactionStatementBuilder(),
    JdbcExecutionOptions.builder().build(),
    new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
        .withUrl("jdbc:postgresql://nbi-db:5432/reconciliation")
        .build()
));

What is the trade-off between the three tools?

Watermark sizing, allowed lateness, and side outputs are three distinct tools for three distinct needs.

| Tool | What It Does | Cost | Use When |
|---|---|---|
| Watermark delay | Absorbs normal disorder before firing | Adds latency to every window | Always (first line of defense) |
| Allowed lateness | Keeps window open for a grace period | Extra state + retraction handling | When late updates are acceptable |
| Side output | Captures very late events separately | Separate processing path | When losing data is unacceptable |

// Three-layer defense:
// 1. Watermark: absorbs 10s of normal disorder
// 2. Allowed lateness: 5 minutes of grace (retractions)
// 3. Side output: captures the extreme tail (reconciliation)

What are common mistakes?

The most common mistake is relying solely on allowed lateness without sizing the watermark correctly. Another is not handling retractions from allowed lateness in the downstream sink.

// WRONG -- allowed lateness but sink does not handle retractions
result.addSink(new AppendOnlySink());  // Cannot handle updates!
// Late event causes retraction → sink crashes or produces wrong results

// CORRECT -- use an upsert-capable sink
result.addSink(new UpsertJdbcSink());  // Handles retractions

What should a senior engineer know?

The real fix for most lateness is a correctly sized watermark, not aggressive allowed lateness, because keeping windows open costs state. Allowed lateness produces updates, so the sink must handle retractions or upserts. Side outputs guarantee you never silently lose data, which is essential in finance where a dropped transaction is a correctness and audit problem. There is always a completeness-versus-latency trade-off.