Stream Processing Semantics and Correctness

In the typical case, a batch job answers a question about a period that has ended and a streaming job answers one about a period that has not. Neither is bound to that — a batch job can compute a snapshot of today, and a stream can replay last year’s events — but the typical case shows where the difficulty sits: it is the open period, not the software, that makes stream processing hard to reason about. Yesterday’s revenue is a number. Today’s revenue, computed continuously, is a number that is correct as of a boundary someone chose, over the data that had arrived by then, and it may be different in an hour. Every design decision below exists to make that sentence precise enough to act on.

The rules below are illustrated with Apache Beam, Apache Flink, and Apache Kafka. The concepts apply beyond those projects; their documented rules make the distinctions concrete.

Two clocks, and the one your question is about

Every record has two times attached to it: the event timestamp it carries, saying when the thing happened, and the processing time each operator observes when it handles the record — which differs from operator to operator, and from the time of arrival at a broker. The distinction between event time and processing time is the foundation, and the only question that matters in design is which one the business question refers to.

For a figure meant to count when sales happened, use event time. “Sales in the 10:00 hour” means sales that happened between 10:00 and 11:00, not sales the pipeline handled then. Aggregating on processing time would make the same hour’s figure depend on how fast the pipeline was running, and a re-run after a fix would move sales into different hours — which is the clearest sign that the wrong clock was used.

Processing time has its uses, and they are operational rather than analytical: how far behind the pipeline is, how long records waited, whether throughput dropped at 03:00. Keep those measurements, and keep them separate from the business figures.

Windows: the grouping, not the schedule

An unbounded stream has no end, so any aggregate over a period needs a bounded group to aggregate over. Not every aggregate does: a running total or a per-key counter updates a fixed number of state variables forever, and Beam offers a single global window for exactly that shape — with the caveat that an unbounded input in the global window needs a non-default trigger before it emits anything. What windows are for is producing results per period. That group is a window — in Beam’s description, Beam “uses windowing to divide a continuously updating unbounded PCollection into logical windows of finite size.” The important word is logical: a window is a set of records selected by their event times, not a moment when something runs. The same window can be computed early, recomputed later, and revised again, and it is still the same window. Details are in stream windowing.

Shape	What it groups	Fits
Fixed (tumbling)	A “consistent duration, non overlapping time interval” — with 30-second windows, 0:00:00 to 0:00:30, then 0:00:30 to 0:01:00. Each record belongs to exactly one	Reporting periods: hourly revenue, daily counts, anything that has to add up to a total without double counting
Sliding	Intervals that “can overlap”: 60 seconds of data, a new window every 30 seconds. Beam notes that “most elements in a data set will belong to more than one window”	Running averages and trend detection, where you want a smooth recent view rather than boundaries
Session	Elements “within a certain gap duration of another element,” applied per key; a record arriving after the gap “initiates the start of a new window”, and a late record that lands within the gap of two existing sessions merges them into one	Activity that comes in bursts — a user’s visit, a device’s period of use — where the boundary belongs to the data rather than the clock

Two consequences are worth stating to consumers before they see the numbers. Sliding windows overlap, so summing across them counts records several times; they describe, they do not tot up. And session windows have no fixed length, so a dashboard built on them cannot promise “every session by 09:00” — a session that is still open has no end yet, by definition — and a late record can merge two sessions that were already reported into one, so a session result is replaceable rather than final.

The watermark: a guess with consequences

A window over event time cannot be closed by looking at the clock, because a record from 10:05 may arrive at 10:20. The mechanism that decides when to stop waiting is the event-time watermark: Beam describes it as “the system’s notion of when all data in a certain window can be expected to have arrived in the pipeline,” and states the rule that follows — “once the watermark progresses past the end of a window, any further element that arrives with a timestamp in that window is considered late data.”

The guide’s own example shows how little the watermark promises. With five-minute windows and a watermark assuming about 30 seconds of lag, the first window closes at 5:30; a record arriving at 5:34 with a timestamp of 3:38 is late. Nothing was broken. The watermark was an estimate, and this record fell outside it.

That is why “how late is too late” is a business decision rather than a configuration detail. A payments figure may need to wait; an operational dashboard may prefer a fast answer that is occasionally revised. The same platform supports both, and the difference is entirely in what someone decided to wait for.

When a result is emitted, and what happens to the previous one

Closing a window and publishing a figure are separate acts, and the second has two settings that are usually decided by accident. Beam names them precisely. A trigger determines “when to emit the aggregated results of each window (referred to as a pane)”; the accumulation mode “determines whether the system accumulates the window panes as the trigger fires, or discards them.” Both are covered in trigger and accumulation mode.

The pairing matters because it decides what a consumer sees. A window that fires three times in accumulating mode publishes a figure that grows toward completeness — each emission replaces the last, and the final one is the answer. The same window in discarding mode publishes three partial figures the consumer has to combine itself, and a consumer that treats any one of them as the total is wrong through no fault of its own.

“Combine” is doing real work in that sentence, and it is only addition for sums and counts. Two panes reporting averages of 15 and 90 do not combine to the average of the window — adding gives 105 where the answer might be 40 — and the same is true of distinct counts, percentiles, and ratios. To use discarding mode with those, emit the pieces that are combinable: sum and count instead of an average, a mergeable sketch or a set instead of a distinct count, numerator and denominator instead of a rate. Accumulating mode avoids the arithmetic and imports a different requirement: each emission is meant to supersede the previous one, so the destination has to make “latest” identifiable. An upsert keyed by the aggregation key and the window together works — the 10:00 window for store A and the 10:00 window for store B are different results, and a key made of the window alone lets one overwrite the other — provided it also compares an ordered revision within that scope rather than blindly overwriting, since a late-arriving earlier pane can otherwise replace a newer one. Appending works too, if each row carries the aggregation key, the window, and a pane identifier and readers select the latest per key and window. Session windows add a condition of their own: a late element can merge two sessions that were already emitted, and upserting the merged session leaves the two earlier rows in place, so the destination needs a retraction, or a replacement relationship between the old identifiers and the new one — or the arrangement has to be confined to windows that do not merge. What does not work is appending bare values with nothing to distinguish them, which leaves every partial figure in place and no way to tell which is current. And “latest known” is not the same as “final” — a window can still be revised while lateness is allowed.

Neither mode is more correct; only one matches what the destination does with the number.

Defaults are conservative and worth knowing. Beam’s default behavior is to output “the aggregated result when it estimates all data has arrived,” then discard subsequent data for that window — one emission, nothing late. Everything more responsive than that is something you asked for, including the obligation to explain revisions.

Late data: three answers, and the default is the strictest

Once a window has closed, a record belonging to it can be dropped, used to revise the result, or routed somewhere for separate handling. The setting that keeps the window’s state alive long enough for the second option is allowed lateness, and its default is the part that surprises teams: Beam states that “the default windowing configuration has an allowed lateness value of 0,” with the default behavior being to “discard late data.” An engine quietly dropping late records is not misbehaving; it is doing what an unconfigured pipeline does.

Allowed lateness is not free, either. Keeping a window open means keeping its state, so a long lateness horizon multiplies the state a job carries — which is the subject of the next section, and the reason this is a trade rather than a preference. Beyond the horizon the streaming job no longer holds the window, so a correction has to come from somewhere else. The usual route is a batch recomputation of the affected period, which presumes that the raw input, the reference data as it was, and a validated, versioned output are all kept; a separate correction stream, or a bounded recomputation of the affected keys, can serve where the destination can be updated in place. Whether the change is announced as a restatement or applied and logged follows the agreed policy on size and kind rather than a blanket rule: a provisional pane refreshing within its contract is expected, a change to a figure consumers were told was final is not. The batch equivalent of the whole arrangement — recompute a trailing range on every run — is covered in late data and lookback.

State is the thing you are really operating

A stateless transformation — parse this record, drop that field — is easy to operate because nothing is remembered. Everything interesting in streaming remembers: a count per key, a window’s partial aggregate, the last value seen for a device, a deduplication set, a join’s buffered side. That accumulated memory is what stateful stream processing manages, and it is the part that turns a streaming job from a program into a system with an operational life.

State has to survive failure, which is what checkpoints are for. In an aligned Flink checkpoint, sources record offsets and insert numbered checkpoint barriers. Each barrier marks the boundary through the job graph. An operator with two inputs aligns their barriers and snapshots state through that boundary, without including later records. A copy-on-write state backend lets processing continue while the older state is written asynchronously. That is the aligned form. Flink can also take unaligned checkpoints, in which a barrier overtakes the records buffered ahead of it and those in-flight records are stored in the checkpoint instead — a checkpoint that completes under backpressure, at the price of being larger to write and restore.

Three operational facts follow, and they are the ones to carry into a design review.

• State size is a main term in recovery cost. A restart has to restore state before the job processes anything, and with the usual local state backends a job holding tens of gigabytes per instance downloads and loads that state first. How long that takes depends on the backend and the recovery path — Flink’s disaggregated state management, for example, can resume without downloading the whole state, paying in remote reads afterwards — and on the bytes actually transferred, the I/O and parallelism available, and the gap between the first record processed and normal throughput. Measure an actual restore rather than deriving a duration from size alone, and compare it with the recovery target you have promised.
• Checkpoint duration is a signal, not a constant. With aligned checkpoints, a slow or skewed input holds up the whole checkpoint, so rising times often point at backpressure somewhere upstream of where the alarm will eventually go off. That is not the only cause: the snapshot’s synchronous and asynchronous phases and the storage they write to can be slow on their own. Read the parts — start delay, alignment time, and the sync and async snapshot durations — rather than the total, so the fix goes to the right place.
• State grows from decisions, not from traffic alone. A longer lateness horizon or wider join window can raise the amount of state a job retains for the same input rate. State grows without bound when new keys or entries keep accumulating without removal, or when stored values themselves keep growing. Keeping a fixed set of keys with fixed-size values does not require expiry to stay bounded. Investigate the number of keys, entries per key, value sizes, retention rules, and watermark progress when memory keeps increasing.

Exactly-once: a claim about state, not about the world

Flink’s exactly-once guarantee concerns the effect of each event on its managed state. It permits repeated processing after recovery and does not by itself guarantee one occurrence of an external effect. A record can be read twice after a restart; what the guarantee covers is that the engine’s own counters and aggregates end up as if it had been read once.

Extending the guarantee through the pipeline requires replayable sources and transactional or idempotent sinks. Both halves are requirements on systems the streaming job does not own.

Link in the chain	What it must provide	What fails without it
Source	Replay from a recorded position, so a restart can re-read what was in flight	Records that were read but not yet checkpointed are simply gone; no downstream mechanism recovers them
Engine	Consistent snapshots of state with the source positions they correspond to	State and position disagree after a restart, producing double counts or gaps
Sink	Either idempotent writes — keyed, and guarded by a version so a replayed older result does not overwrite a newer one — or a transaction committed with the checkpoint — a two-phase commit sink	A replay after failure duplicates rows, emails, or payments, however correct the engine’s state is
Downstream reader	Willingness to read only committed data. Kafka’s isolation.level is the concrete case: read_committed will “only return transactional messages which have been committed,” while the default read_uncommitted returns “all messages, even transactional messages which have been aborted”	A transactional pipeline whose consumer reads uncommitted output has bought the cost of transactions and none of the benefit

That last row is the one most often missed, and it has a cost worth stating: in read_committed mode a consumer reads only up to the last stable offset — the offset before the first open transaction — so it cannot read to the end while a transaction is in flight. Correctness of this kind is paid for in latency, which is exactly the trade the business should be asked about rather than told about.

And some effects cannot be made exactly-once at all. An email that has been sent is sent; a payment API without an idempotency key will charge twice. For those, the honest design is to make the effect idempotent at the destination or to accept and reconcile — which is why a periodic reconciliation against the source remains part of a streaming platform, not a sign that something is wrong with it.

What to settle before the pipeline exists

A changing figure is difficult to use when producers and consumers have not agreed whether it may change. Four questions settle that, and they are answerable by the people who want the data rather than by the team building it.

1. Which clock, and which grouping? Event time or processing time; fixed, sliding, or session windows. This fixes what the number means.
2. How long do we wait? The lateness horizon, stated as a duration with a reason — observed delays, a reporting deadline, a regulator. It is measured against the watermark, not the wall clock: a window expires when the watermark passes its end plus the allowance, so a stalled watermark keeps state alive longer in real time and a jump expires it sooner. It is a state cost, and it feeds a freshness commitment without being one — the time to a first result, the period during which the figure may still be revised, and the point of finality are three separate promises, and the freshness a consumer observes also depends on watermark progress, the trigger, and when the sink makes results visible.
3. May the figure change after publication, and how will anyone know? Accumulating or discarding, revisions announced or silent. A number that may move needs to say so where it is read.
4. When is it final? The moment after which the figure will not change is a business decision, not a property of the technology. Without it, every report is provisional forever, which is its own kind of unusable.

Write the answers next to the data, state them as targets someone owns — the same discipline an objective needs — and the arguments move from “the dashboard is wrong” to “the dashboard is doing what we agreed.” That is the whole return on taking these semantics seriously: not a more accurate number, but a number whose accuracy can be discussed.

References

• Apache Beam Programming Guide (windowing, triggers, accumulation modes, watermarks and late data)
• Basics of the Beam model, Aggregation
• Apache Flink Documentation, Windows: Late elements considerations
• Apache Flink Documentation, Learn Flink: Fault Tolerance via State Snapshots
• Apache Flink Documentation, Checkpointing under backpressure
• Apache Flink Documentation, Disaggregated State Management
• Apache Kafka Documentation, Consumer Configs (isolation.level)