---
title: "When Can a Late AI Input Still Change the Streaming Result?"
description: "Leviathan Systems — Technical content Published 2026-10-21; updated 2026-10-21."
canonical: "https://www.leviathansystems.co/articles/ai-streaming-watermark-event-time-window-state"
last-updated: "2026-10-21"
---

# When Can a Late AI Input Still Change the Streaming Result?

Leviathan Systems — Technical content
Published 2026-10-21; updated 2026-10-21.

A late record can still change a streaming aggregate when it belongs to the window and remains eligible under the query's lateness and state-retention rules. In Spark 3.5.6 micro-batch window aggregation, a five-minute watermark delay is measured against event-time progress, rather than elapsed wall-clock waiting. It protects sufficiently recent event-time inputs from lateness dropping; it does not promise a result five minutes after arrival. Whether the buyer sees a changing count or a finalized row also depends on the output mode.

## Identify the time and the aggregate being promised

An AI feature feed might promise a complete count of equipment observations for each ten-minute interval. That promise leaves a practical question: can an observation delivered later revise the interval's count, and what does the consumer receive before revision ends? An input-rate figure alone cannot answer either part.

This explanation uses Apache Spark 3.5.6 Structured Streaming's micro-batch model, one input stream, one event-time column and an ordinary tumbling-window count. It evaluates a proposed data deliverable. Joins, multiple-stream watermark policies and other streaming engines need their own semantics.

Event time is the timestamp carried in the selected input column. Processing time concerns when the engine handles the record. A record describing 10:09 can reach processing after a record describing 10:12 without changing which event-time interval it belongs to. Establish what the chosen column actually represents before interpreting lateness: an equipment observation timestamp and a later ingestion timestamp describe different events.

A window groups records by that timestamp. Spark's window boundaries include the start and exclude the end, so [10:00, 10:10) includes 10:09 and excludes exactly 10:10. The engine retains intermediate aggregate state so another eligible record can alter an existing count; retaining a count does not imply retaining every original input record.

References: [SW356](https://spark.apache.org/docs/3.5.6/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.withWatermark.html), [WIN356](https://spark.apache.org/docs/3.5.6/api/python/reference/pyspark.sql/api/pyspark.sql.functions.window.html), [ST356](https://raw.githubusercontent.com/apache/spark/v3.5.6/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala)

## Read five minutes as event-time protection

Spark 3.5.6 documents a one-sided guarantee for watermarking an aggregation: data less than the declared delay behind the latest processed event time is guaranteed to be aggregated. With a five-minute delay and latest event time 10:12, a 10:09 record is three minutes behind and lies within that protection. This is a lateness guarantee for the stated aggregation, not a guarantee against unrelated application filtering or failure.

The reverse inference is invalid. Data more than five minutes behind may or may not be aggregated. The watermark API also explains that coordination across partitions can leave the actual watermark farther behind. The delay number alone therefore establishes neither an exact rejection boundary for every arriving record nor an exact output time.

State cleanup has additional applicability conditions. The guide requires Append or Update output, aggregation by the event-time column or its window, and watermark information attached to that same column before aggregation. Those conditions describe the query whose promise is being evaluated; simply mentioning a five-minute watermark does not establish them.

References: [SG356](https://spark.apache.org/docs/3.5.6/structured-streaming-programming-guide.html), [SW356](https://spark.apache.org/docs/3.5.6/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.withWatermark.html)

## WM-63: two records remain revisable at watermark 10:07

WM-63 is an original fictional timeline, with no query executed. All timestamps use one same-day UTC basis. Count observations for one identified equipment group in ten-minute tumbling windows. Use a five-minute watermark delay and Append output. The aggregation satisfies the cleanup conditions above. Assume valid timestamps, no duplicates, no other filters and successful processing of every eligible record specified here.

In trigger A, observations R1 at 10:02 and R2 at 10:05 contribute to [10:00, 10:10). Its retained count is two. A third processed observation, P at 10:12, belongs to the following window. P establishes the maximum observed event time of 10:12 without adding to the earlier window's count.

For this timeline, stipulate that the resulting watermark is available to trigger B as 10:07: 10:12 minus five minutes. Spark's guide illustrates watermark use across trigger boundaries, rather than immediate expiration whenever a new maximum appears. The example supplies these exact trigger watermarks; it does not infer an observed runtime watermark solely from the delay label.

The earlier window ends at 10:10, which is still ahead of 10:07. Its state remains retained. Append has not emitted that window's count as a final result.

References: [SG356](https://spark.apache.org/docs/3.5.6/structured-streaming-programming-guide.html), [WIN356](https://spark.apache.org/docs/3.5.6/api/python/reference/pyspark.sql/api/pyspark.sql.functions.window.html), [WT356](https://raw.githubusercontent.com/apache/spark/v3.5.6/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/WatermarkTracker.scala), [ST356](https://raw.githubusercontent.com/apache/spark/v3.5.6/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala)

## A 10:09 observation changes the count to three

Trigger B now receives R3 with event time 10:09. It belongs to [10:00, 10:10), remains ahead of the supplied 10:07 watermark and is only three minutes behind the maximum 10:12. Processing it changes the retained count from two to three. R3 does not advance the maximum event time.

The aggregate has changed even though Append has still emitted no row for this window. An internal count of three and a consumer-visible final row are different facts.

Now imagine a period with no further event-time progress. Advancing the wall clock by five minutes leaves the maximum observed event time at 10:12. It supplies no newer event timestamp from which to derive a watermark past 10:10. A subsequent trigger can apply an already advanced watermark, but waiting alone cannot create the missing event-time progress. WM-63 assigns no wall-clock completion deadline.

References: [SW356](https://spark.apache.org/docs/3.5.6/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.withWatermark.html), [WT356](https://raw.githubusercontent.com/apache/spark/v3.5.6/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/WatermarkTracker.scala), [ST356](https://raw.githubusercontent.com/apache/spark/v3.5.6/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala)

## Later event-time progress lets this window finalize

Trigger C processes a new observation Q at 10:17, outside the earlier window. Its count remains three. The new maximum yields a watermark of 10:12 after subtracting the five-minute delay. Stipulate that trigger D uses that advanced watermark and successfully completes finalization and state removal for [10:00, 10:10), whose end 10:10 is now behind 10:12. Also stipulate that the sink receives its Append row with count three.

Only after that completed cleanup does trigger E receive R4 at 10:06. R4 belongs to the earlier interval, but is eleven minutes behind the maximum 10:17 and outside the five-minute protection. In this case the finalized window's state has already been removed under the advanced watermark, so R4 cannot revise its emitted count. The result remains three.

The exclusion follows from the supplied completed cleanup and finalized-window boundary, rather than a universal rule that every record older than five minutes is dropped. Nor does finalization prove that all observations of the physical interval have arrived: R4 itself demonstrates a later arrival. Final means closed under this query's lateness policy. Reconstructing a broader history would be a separate data deliverable.

References: [WT356](https://raw.githubusercontent.com/apache/spark/v3.5.6/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/WatermarkTracker.scala), [ST356](https://raw.githubusercontent.com/apache/spark/v3.5.6/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala)

## The same count has different delivery meanings by output mode

Compare the main Append case with a hypothetical Update version of the same count and input sequence. With successful output stipulated, Update can emit count two after trigger A and count three after trigger B. The second value revises the same equipment-window aggregate. Adding those two counts to obtain five would double-count the evolving result. The receiving system needs to associate each revision with its window and group.

Append instead withholds this aggregate until finalization, then emits the one row with count three in the stipulated trigger D. A buyer who needs one finalized row and a buyer who needs an earlier revisable value are asking for different outputs. Update's state cleanup does not itself supply a separate final-result message.

Complete mode writes the entire current result table and preserves this aggregation's old state rather than using the watermark to remove it. Its name describes output coverage of that result table; it does not certify that no further source data can arrive. The removed-state premise of WM-63's Append case cannot simply be transferred to a Complete-mode offer.

The writer's output contract also leaves the downstream use to establish. Emission of a row does not prove that a particular model consumed that version or recomputed an earlier prediction.

References: [OUT356](https://spark.apache.org/docs/3.5.6/api/python/reference/pyspark.ss/api/pyspark.sql.streaming.DataStreamWriter.outputMode.html), [ST356](https://raw.githubusercontent.com/apache/spark/v3.5.6/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala)

## Compare two offers with the same advertised input rate

Suppose two providers advertise the same input rate and five minutes of allowed lateness. Offer Live supplies Update results for a dashboard or current feature view. Offer Closed supplies Append rows for a dataset that expects one finalized aggregate per equipment window. Even with identical incoming records, their deliverables differ.

For WM-63, Live can show two and then three before Closed supplies its finalized three. Live's useful evidence identifies the equipment-window key, the revisions made available and how the receiving view treats replacement values. Closed's useful evidence identifies the window definition, applicable watermark and completed finalization, together with the resulting row. A declaration of input throughput supplies none of that evidence.

Have the provider explain the late 10:09 and post-cleanup 10:06 arrivals against its actual offered version and query. It should distinguish data protected by the delay, data outside that guarantee, and windows already finalized. If the requirement also promises a wall-clock deadline, the offer needs a separate account of processing and delivery timing, including periods without newer event timestamps. Five minutes of event-time allowance cannot substantiate that deadline.

Choose the deliverable around how the result will be used. A live feature can be useful while still revisable. A fixed extract can require declared finalization. If the buyer later needs both the revised history and the value available to a past prediction, preserve that distinction explicitly; the related historical-feature guide explains the separate available-at-time question. Neither output mode proves which version an earlier prediction actually used.

References: [OUT356](https://spark.apache.org/docs/3.5.6/api/python/reference/pyspark.ss/api/pyspark.sql.streaming.DataStreamWriter.outputMode.html), [SW356](https://spark.apache.org/docs/3.5.6/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.withWatermark.html)

## Locate this boundary after source capture and log visibility

Spark's event-time watermark is a timestamp used in state and late-data decisions. Kafka's high watermark and last stable offset are partition-log boundaries tied to replication and transaction visibility. The Kafka clients 4.1.2 endOffsets contract defines them in offset terms, rather than minutes of event time. A Kafka record being readable therefore does not establish whether an aggregation can still include it. The related Kafka guide follows that earlier visibility boundary.

Upstream capture is another decision. OPC UA Part 4's Variable queue rules can preserve only the newest Notification when queue size is one. Once earlier Notifications have been displaced before delivery, keeping Spark window state longer cannot recreate their missing observations. The related OPC UA guide explains sampling, filtering and queue retention.

For an offered AI input, identify which gap is at issue: an observation absent upstream, a record not yet visible to the reader, or a delivered record outside an aggregate's retained horizon. Each requires evidence from its own stage. The word complete becomes useful only when it names the input population and the boundary being promised.

References: [K412](https://kafka.apache.org/41/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html), [OPC4](https://reference.opcfoundation.org/specs/OPC-10000-4/5.13.1.5)

## Scope physical equipment connections separately

If the chosen platform also requires physical connections between equipment, Leviathan's Structured Cabling & Fiber Installation service is the relevant route for agreed fiber and copper installation, pathways, endpoint labels and connection records. Use the related project inquiry with a nonconfidential equipment and site outline, required connections and material-supply responsibilities to discuss scope and availability.

That inquiry covers physical installation. Spark integration, streaming-query behavior and AI data-processing deliverables remain with the separately appointed software and platform providers.

References: [L1](https://www.leviathansystems.co/services/structured-cabling)

## Primary sources

- [SG356 — Apache Spark 3.5.6 — Structured Streaming Programming Guide](https://spark.apache.org/docs/3.5.6/structured-streaming-programming-guide.html). Checked 2026-10-21. Micro-batch event-time aggregation: trigger-boundary illustrations, conditions for watermark cleanup and the one-sided lateness guarantee. Applies to the stated single-stream window count.
- [SW356 — PySpark 3.5.6 — DataFrame.withWatermark](https://spark.apache.org/docs/3.5.6/api/python/reference/pyspark.sql/api/pyspark.sql.DataFrame.withWatermark.html). Checked 2026-10-21. Defines the event-time maximum and delay relationship, finalization purpose and possible additional watermark lag from coordination across partitions.
- [WIN356 — PySpark 3.5.6 — functions.window](https://spark.apache.org/docs/3.5.6/api/python/reference/pyspark.sql/api/pyspark.sql.functions.window.html). Checked 2026-10-21. Inclusive window starts, exclusive ends and tumbling-window behavior support the interval membership used in WM-63.
- [OUT356 — PySpark 3.5.6 — DataStreamWriter.outputMode](https://spark.apache.org/docs/3.5.6/api/python/reference/pyspark.ss/api/pyspark.sql.streaming.DataStreamWriter.outputMode.html). Checked 2026-10-21. Defines Append, Update and Complete output populations. Output selection alone does not establish a downstream model's consumption.
- [ST356 — Apache Spark v3.5.6 — aggregation state and output implementation](https://raw.githubusercontent.com/apache/spark/v3.5.6/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/statefulOperators.scala). Checked 2026-10-21. The ordinary aggregation StateStoreSaveExec branches support Append emission on state eviction, Update emission of modified rows and Complete preservation of aggregate state.
- [WT356 — Apache Spark v3.5.6 — WatermarkTracker](https://raw.githubusercontent.com/apache/spark/v3.5.6/sql/core/src/main/scala/org/apache/spark/sql/execution/streaming/WatermarkTracker.scala). Checked 2026-10-21. Supports watermark progress from observed event-time statistics and delay, without substituting elapsed wall-clock time. Multiple-stream policy is outside this example.
- [K412 — Apache Kafka — KafkaConsumer, clients 4.1.2 API](https://kafka.apache.org/41/javadoc/org/apache/kafka/clients/consumer/KafkaConsumer.html). Checked 2026-10-21. Reading Transactional Messages and endOffsets distinguish partition-log visibility boundaries from an event-time watermark.
- [OPC4 — OPC Foundation — OPC10000-4 v1.05.07, 5.13.1.5: Queue parameters](https://reference.opcfoundation.org/specs/OPC-10000-4/5.13.1.5). Checked 2026-10-21. Variable DataValue Notification retention, especially the newest-value behavior of queue size one. This brief comparison does not extend those rules to Events.
- [L1 — Leviathan Systems — Structured Cabling & Fiber Installation](https://www.leviathansystems.co/services/structured-cabling). Checked 2026-10-21. Physical installation to agreed pathways, endpoint schedules and labels with identified material responsibilities. Relevant to a separate equipment-connection project.

## Related Leviathan resources

- [Kafka offsets: log visibility and returned AI input records](/articles/kafka-partition-offsets-transaction-visibility)
- [Industrial AI feeds: source sampling, queues and retained changes](/articles/industrial-ai-opcua-sampling-queue-publishing)
- [Historical AI features: event time and available-at time](/articles/historical-ai-features-event-time-known-time-join)
- [Structured Cabling & Fiber Installation](/services/structured-cabling)
- [Discuss a separate physical equipment-cabling project](/contact?service=structured-cabling&from=%2Farticles%2Fai-streaming-watermark-event-time-window-state)

## Does the platform project also need physical cabling?

Share a nonconfidential equipment and site outline, required physical connections, material responsibilities and handoff needs to discuss installation scope and availability.

[Discuss the scope](/contact?service=structured-cabling&from=/articles/ai-streaming-watermark-event-time-window-state)