9.3 Event Time, Watermarks, and End-to-End Semantics: How Stream Processing Handles Late Arrivals and Replay

Batch reports are generated every morning, but frontline teams need real-time alerts when equipment temperatures remain abnormally high for ten consecutive minutes. After a network outage, sensor events from yesterday arrive today. When a job restarts, it reprocesses a batch of messages. If we process events based on arrival time, historical state drifts. If we only ensure "each event is processed exactly once," downstream alerting systems may still receive duplicate requests.

The challenge in stream processing isn't simply reading messages in a loop; it's maintaining clear temporal and state semantics over unbounded, out-of-order, and replayable data.

Learning Objectives

• Distinguish between event time, ingestion time, and processing time;
• Use windows and watermarks to manage out-of-order events and state;
• Explain the boundaries between at-most-once, at-least-once, and end-to-end exactly-once semantics;
• Design stream pipelines that are replayable, idempotent, and observable.

1. Unbounded data never "completes"

A batch table has an enumerable snapshot; a stream continuously generates new data. The system must decide:

• When to trigger computation;
• When a window can emit a preliminary result;
• Whether to update, retract, or discard late-arriving data;
• How long to retain state;
• Where to resume from after a failure.

Stream processing often operates in micro-batch or record-by-record mode, but both approaches must address these semantic questions.

2. Three Clocks

• Event time: The time when an event occurs in the source system;
• Ingest time: The time when the platform first receives the event;
• Processing time: The time when an operator actually processes the event.

A ten-minute window based on processing time is easy to implement, but results will change when replaying historical data. Business metrics typically use event time, while retaining ingest time to measure latency. Source clock drift, time zones, and device restarts must all be monitored for quality assurance.

3. Window Definitions Are Business Rules

• tumbling window: a fixed window with no overlap;
• sliding window: a window that overlaps with a defined slide step;
• session window: a window that closes dynamically based on user activity intervals.

To define "the past ten minutes," you must specify the boundaries, time zone, trigger frequency, and how results are updated. A single event can enter multiple sliding windows, and both state retention and computational cost increase with greater overlap.

Example in Structured Streaming style:

from pyspark.sql import functions as F

counts = (
    events
    .withWatermark("event_time", "30 minutes")
    .groupBy(
        F.window("event_time", "10 minutes"),
        F.col("equipment_id"),
    )
    .agg(F.avg("temperature").alias("avg_temperature"))
)

The output mode and whether the sink supports updates must be confirmed in conjunction with the query type.

4. Watermark is not "current time minus 30 minutes"

A watermark signals to the engine a strategy for event time progression and allowable lateness, enabling the system to clean up old window states. It is typically tied to the actual observed event time progression and should not be assumed to equate to wall-clock time.

Setting it too short results in more late events being unable to update results; setting it too long increases state, checkpoint, and recovery overhead. The watermark should be chosen based on real-world latency distributions, business-defined window tolerances, and resource constraints, and its progress should be monitored alongside rejected or missed late events.

A watermark is not a guarantee of completeness. Sources may stall, certain partitions may lag, and multiple input streams require a global progress strategy to be defined.

5. Deduplication Requires a Stable Event Identity

Message system re-delivery, producer retries, and replay all result in duplicates. Prefer using a stable event_id generated at the source, and define identity scope and retention period.

Deduplicating by row value alone will accidentally remove events that are identical in value but genuinely occur twice; storing all IDs across an infinite history would make the state unbounded. Instead, use a combination of event ID, event timestamp, and watermark to manage deduplication state, and record how to handle duplicate events that fall outside the retention window.

Business updates must also be distinguished: duplicates, revisions, cancellations, and late arrivals. They are not the same type of operation.

6. Processing Count Is Not End-to-End Result Semantics

Common terms:

• at-most-once: May lose messages, but avoids retries that cause duplicates;
• at-least-once: Ensures no message loss, but retry on failure may result in duplicates;
• exactly-once: Within a clearly defined system boundary, each input produces a result or state change exactly once.

End-to-end guarantees depend on the entire processing chain: replayable source, progress tracking, deterministic transformations, checkpoints, and sink idempotency or transactional commits. A system engine claiming exactly-once semantics does not guarantee that any external HTTP API will execute exactly once, external components must independently support idempotency or transactional safety.

A practical pattern for achieving this is to use batch or event idempotency keys:

begin transaction
  if output_id not committed:
      write result rows
      record output_id
commit

The result and the commit marker must be written atomically within the same transaction boundary. Writing the result first and then separately recording the marker leaves an open window for failure and data inconsistency.

7. Checkpoint Record Execution Progress

A checkpoint typically includes source offsets, state, and query metadata, enabling a job to resume from the point of failure. It is not a long-term backup of business data and does not guarantee recovery across arbitrary changes in code or schema.

Before deploying changes, verify the following:

• New code is compatible with the schema of existing state;
• The checkpoint path is unique and persists across restarts;
• Source data retention period covers the longest possible failure window;
• The sink can safely accept replayed data;
• How to reconstruct state from raw logs if in-place upgrades are not feasible.

Maintaining a replayable original event log is a critical capability for fixing logical errors and re-populating results.

8. Backpressure and System Health

When the input rate consistently exceeds the processing rate, lag accumulates. Monitoring should include:

• The latest offset from the source and the offset processed so far;
• The number of input rows per batch, processing duration, and scheduling interval;
• Watermark/event-time lag;
• State row count, state size, and checkpoint duration;
• Counts of late arrivals, duplicates, parsing failures, and dead-letter queue entries;
• Sink latency, errors, and idempotent conflict occurrences.

Before scaling out, determine whether the bottleneck lies in source rate limiting, shuffle operations, hot keys, state bloat, or sink throughput. Simply increasing consumer parallelism indefinitely may merely shift the pressure downstream.

9. How Late-Arriving Results Are Published

You have several options:

• Append: Write the result only once after the window is sufficiently determined; introduces higher latency.
• Update/UPSERT: When late-arriving data arrives, update the existing result key.
• Revert/Change Log: Emit both the old value (revert) and the new value.
• Initial and Final Values: Provide a low-latency preliminary result, followed by a final closure and version annotation.

The downstream system must understand the chosen semantics. Writing a stream that may update into a table that only supports append will result in multiple conflicting "final values."

10. Testing and Exercise

Construct a deterministic sequence of events covering the following scenarios:

Out-of-order but within watermark
Late by more than the allowed delay range
Same event_id re-delivered
Same content, different event_id
Job crashes before and after writing
Source partition temporarily stalls
Schema extension and incompatible schema changes
Full replay from an old offset

Validation is not limited to a single output, it must also include post-restart state, repeated side effects, and final convergence outcomes.

Common Misconceptions

• Stream processing is just faster batch processing: Unbounded inputs require progress tracking, state management, and semantics for correction.
• Data before a watermark is guaranteed to be complete: It’s an engineering strategy for handling late data and state, not a factual guarantee.
• An engine’s exactly-once semantics equals end-to-end exactly-once semantics: The final boundary is determined by the sink and external side effects.
• Checkpoints can replace original logs: Logical errors and incompatible upgrades still require replaying the source.

Exercise

1. Define event time, window, watermark, and update semantics for a ten-minute temperature alarm event.
2. Generate out-of-order and duplicate events to verify whether the window results eventually converge.
3. Design a database upsert key that is safe against job retries.
4. Sketch a troubleshooting path for a job with growing state, showing lag, watermark, and state size over time.

Summary

Reliable stream processing treats time, state, and failure as normal inputs. A watermark determines the trade-off between late data and resource usage, checkpoints enable recovery, and end-to-end delivery ensures guarantees span the entire boundary, from source through engine to sink.

The next chapter explores data ethics: just because a system can collect, infer, and make automated decisions doesn't mean it should.