Handling Late-Arriving Events with Allowed Lateness in ksqlDB

Out-of-order data is inevitable in stream processing. Network latency, producer retries, and upstream backpressure cause events to arrive after their processing window has closed. ksqlDB controls how long windowed operations accept these late records with the GRACE PERIOD clause. Configuring it correctly is the difference between accurate aggregates and silently dropped data—and between a bounded state store and one that grows for a full day.

Why Late Events Matter in Windowed Aggregations

When you define a windowed aggregation—tumbling, hopping, or session—ksqlDB assigns each record to a window based on its event timestamp, not the time it was consumed. The aggregate is updated as records arrive, and the window’s result becomes final once the window closes. If a record whose timestamp belongs to an already-closed window arrives after the grace period, it is dropped. The result is an undercount that downstream analytics never recover.

Consider counting page views per minute. A producer retry storm causes a batch of events to arrive 30 seconds after their window ended. Without enough allowed lateness, those views never make it into the count. For teams monitoring SLAs or revenue metrics, that loss is material and invisible. The GRACE PERIOD setting keeps the window’s state open for a specified duration after the window’s end, so late records are still folded into the correct aggregate.

This behavior is tightly coupled to the underlying state store, as covered in Windowed Aggregations and Joins in ksqlDB. A longer grace period keeps each window’s state alive longer, which directly increases state store size and the duration over which downstream consumers may receive updates. Once the grace period elapses, the window is closed, its state becomes eligible for cleanup, and no further updates are accepted.

Configuring Allowed Lateness with GRACE PERIOD

GRACE PERIOD is specified inside the WINDOW clause, alongside SIZE (and ADVANCE BY for hopping windows). It accepts a duration and a time unit—MILLISECONDS, SECONDS, MINUTES, HOURS, or DAYS.

CREATE TABLE page_views_per_minute AS
SELECT
    page_id,
    COUNT(*) AS view_count,
    WINDOWSTART AS window_start,
    WINDOWEND AS window_end
FROM page_views_stream
WINDOW TUMBLING (SIZE 1 MINUTE, GRACE PERIOD 30 SECONDS)
GROUP BY page_id
EMIT CHANGES;

Here the tumbling window is one minute long with a 30-second grace period. For 30 seconds after a window ends, ksqlDB still accepts records whose timestamps fall inside that window and updates the count. After that, the window closes and its state is finalized.

The same clause applies to hopping and session windows. Hopping windows also require ADVANCE BY; session windows take an inactivity-gap duration as their first parameter:

-- Hopping window: 1-hour windows advancing every 10 minutes
CREATE TABLE hourly_traffic AS
SELECT
    sensor_id,
    AVG(reading) AS avg_reading
FROM sensor_readings
WINDOW HOPPING (SIZE 1 HOUR, ADVANCE BY 10 MINUTES, GRACE PERIOD 5 MINUTES)
GROUP BY sensor_id
EMIT CHANGES;

-- Session window: 5-minute inactivity gap
CREATE TABLE user_sessions AS
SELECT
    user_id,
    COUNT(*) AS event_count
FROM click_stream
WINDOW SESSION (5 MINUTES, GRACE PERIOD 2 MINUTES)
GROUP BY user_id
EMIT CHANGES;

If you omit GRACE PERIOD, ksqlDB applies a default of 24 hours. This default exists to avoid accidental data loss, but it means every window stays open for a full day, which can drive large and unnecessary state on high-throughput streams. The 24-hour default was deprecated upstream in Kafka Streams under KIP-633, so relying on it can also produce warnings. Always set an explicit grace period that matches your maximum expected lateness and your tolerance for state size.

GRACE PERIOD is independent of RETENTION. GRACE PERIOD governs how long late records are accepted into a window; RETENTION governs how long the materialized windowed result remains queryable by pull queries (and must be at least window size + grace period).

Operational Impact on State Stores

Window state is held in RocksDB, the embedded state store that backs ksqlDB’s persistent queries. Every open window consumes memory and disk. A longer grace period keeps more windows open at once, so it directly raises the resident key count and on-disk size.

The effective retention of a window’s state is window size + grace period. A one-minute tumbling window with a 30-second grace period keeps each window’s state for about 90 seconds past the window start. Hopping windows are heavier: because windows overlap, a single record belongs to SIZE / ADVANCE BY windows at once, multiplying the state held per key. Factor both SIZE and ADVANCE BY into any capacity estimate.

To watch state store growth, scrape the RocksDB property metrics that Kafka Streams exposes through JMX. The metric you want is estimate-num-keys (the RocksDB estimate of live keys) on the stream-state-metrics group, scoped to window stores via the rocksdb-window-state-id tag. These property metrics were added in KIP-607 and record at INFO level:

# Sample the estimated key count of every RocksDB window store via JMX
kafka-run-class kafka.tools.JmxTool \
  --object-name 'kafka.streams:type=stream-state-metrics,thread-id=*,task-id=*,rocksdb-window-state-id=*' \
  --attributes estimate-num-keys,total-sst-files-size \
  --jmx-url service:jmx:rmi:///jndi/rmi://localhost:9999/jmxrmi

estimate-num-keys and total-sst-files-size rise as the grace period grows and more windows stay open concurrently. If the working set outgrows RocksDB’s block cache, reads spill to disk and tail latencies climb; left unchecked, a runaway grace period on a high-cardinality key can exhaust disk entirely. Treat these two attributes as the leading indicators when sizing a grace period.

Allowed Lateness in Windowed (Stream-Stream) Joins

Allowed lateness is not limited to aggregations. Stream-stream joins are windowed: the WITHIN clause defines the join window, and GRACE PERIOD defines how long out-of-order records are accepted on either side before the join result is finalized. (Stream-table joins, by contrast, are non-windowed—each stream record triggers a lookup against the table’s current state, so they accept neither WITHIN nor GRACE PERIOD. To tolerate late updates on a reference dataset, model it as a stream and join stream-to-stream.)

Consider joining a stream of orders to a stream of shipments, where a shipment event may lag the matching order:

CREATE STREAM shipped_orders AS
SELECT
    o.order_id,
    o.customer_id,
    o.amount,
    s.shipment_id,
    s.carrier
FROM orders_stream o
INNER JOIN shipments_stream s
    WITHIN 1 HOUR GRACE PERIOD 15 MINUTES
    ON o.order_id = s.order_id
EMIT CHANGES;

WITHIN 1 HOUR matches records whose timestamps fall within one hour of each other; GRACE PERIOD 15 MINUTES keeps the join state open an additional 15 minutes for late arrivals on either side. The grace-period syntax here is WITHIN <size> <unit> GRACE PERIOD <size> <unit>—no parentheses or comma, unlike the aggregation WINDOW clause. For left/outer stream-stream joins, the grace period also delays emission of non-matched (null-padded) rows until the period elapses, which avoids spurious “no match” results that a later record would have satisfied.

Because both sides of a stream-stream join are buffered, state is held for the window size plus the grace period on each stream, roughly doubling the footprint compared with a single-stream aggregation. Plan capacity accordingly.

Tuning Grace Period for Production Workloads

Choosing a grace period is a measurement problem, not a guess. Start by characterizing actual lateness: the gap between an event’s timestamp and when ksqlDB processes it. Have producers stamp event time, and compute the difference against ingestion time so you can build a lateness distribution and read off its tail percentiles (p95, p99, p99.9).

ksqlDB has no built-in percentile aggregate, so compute these tails outside the engine—e.g., export the per-record lateness to a metrics backend (Prometheus histograms, a data warehouse, or an external job) and read percentiles there. Inside ksqlDB you can still surface coarse signals such as MAX(latency_ms) or a HISTOGRAM of bucketed lateness, but treat exact percentiles as an out-of-band calculation.

Once you know the distribution, set the grace period to cover your target percentile with a margin. If p99.9 lateness is 45 seconds, a 60-second grace period is a reasonable choice. Chasing the absolute maximum observed lateness is rarely worth it—the state-store cost of covering the last fraction of a percent usually dwarfs its value.

Account for downstream update semantics, too. Every late record that lands within the grace period emits an updated aggregate (EMIT CHANGES), so sinks must be idempotent: a database sink needs upserts, and a dashboard must overwrite, not append, previously reported values. If you instead want a single result per window after it closes, use EMIT FINAL rather than EMIT CHANGES—the result is emitted once when the grace period expires, at the cost of latency equal to the grace period.

Monitoring and Troubleshooting

Two things are worth watching continuously: how many records are being dropped for lateness, and how large the window state is growing.

• dropped-records-rate / dropped-records-total — task-level metrics on the stream-task-metrics group (kafka.streams:type=stream-task-metrics,thread-id=*,task-id=*), recorded at INFO level. These count records dropped within a task, including late records that missed the grace period. A persistently non-zero rate is the clearest sign your grace period is too short. (This metric replaced the older per-store late-record-drop and skipped-records-rate metrics removed by KIP-444.)
• estimate-num-keys / total-sst-files-size — the RocksDB window-store metrics shown earlier, for tracking state growth driven by the grace period.

To inspect a running query’s window configuration—window type, size, and grace period—use:

DESCRIBE page_views_per_minute EXTENDED;

DESCRIBE ... EXTENDED also reports per-query runtime statistics such as messages processed. To list the persistent queries themselves and their state, use SHOW QUERIES;. (ksqlDB exposes no /query-status REST endpoint; the /status endpoint reports the execution status of a submitted CREATE/DROP/TERMINATE command, not per-query state-store details.)

When dropped-records-rate is high, you have two levers: raise the grace period—which requires recreating the stream or table, since the window clause is fixed at creation—or attack the source of lateness upstream. Raising the grace period is a stopgap that trades accuracy for state size; reducing producer latency through batching, compression, and network tuning is the durable fix.

For deeper background on windowing and state-store behavior, see the Kafka Streams windowing documentation and the ksqlDB operations guide.

Grace Period and Exactly-Once Semantics

With exactly-once processing enabled, late records accepted within the grace period are folded into the aggregate and committed inside the same transactional cycle as everything else, so downstream consumers read consistent intermediate and final results. Exactly-once does not change what the grace period accepts—it changes the commit guarantees around those updates.

The completeness-vs-latency trade-off is inherent to the grace period regardless of the processing guarantee. With EMIT CHANGES, consumers see a stream of revisions as late records land; with EMIT FINAL, they wait until the grace period expires to see one authoritative result per window. Choose based on whether your consumers can absorb updates or need a single settled value.

Enable exactly-once at the session or query level before creating the query:

SET 'processing.guarantee' = 'exactly_once_v2';

CREATE TABLE accurate_page_views AS
SELECT
    page_id,
    COUNT(*) AS view_count
FROM page_views_stream
WINDOW TUMBLING (SIZE 1 MINUTE, GRACE PERIOD 30 SECONDS)
GROUP BY page_id
EMIT CHANGES;

exactly_once_v2 is the current transactional guarantee (it superseded the original exactly_once and exactly_once_beta values). The cost is higher end-to-end latency and the transactional overhead of coordinating commits.

The grace period is a small clause with outsized consequences for accuracy, state size, and downstream contract. Measure your real lateness distribution, set an explicit grace period instead of inheriting the 24-hour default, monitor dropped-records-rate and RocksDB key counts, and you can absorb the inevitable late event without either losing data or letting state run away.

Related Pages

• Stream Processing