Distributed Stream Processing: Watermarks, Windowing, and Chandy-Lamport Checkpoints

1. The Core Engineering Challenge: Unbounded Streams & The Reality of Out-of-Order Time

Project Identity, Version & License Baseline

Traditional batch processing systems (such as Hadoop, Spark, or daily SQL ETLs) operate over bounded, static datasets:

• Data is collected over a fixed interval (e.g. all transactions from midnight to midnight).
• A batch job runs over the static file on disk.
• If errors occur, the job is simply re-run.

Modern cloud architectures, however, generate unbounded, continuous event streams:

• Financial transactions, fraud detection signals, IoT sensor telemetry, and mobile application telemetry stream into systems 24/7/365.
• Business requirements demand real-time alerting within seconds, making multi-hour batch runs obsolete.

Processing unbounded streams at high scale introduces fundamental distributed systems challenges:

The Out-of-Order Stream Phenomenon:
---------------------------------------------------------------------------------
Physical Wall Clock Time:   10:00:01     10:00:02     10:00:03     10:00:04
                                |            |            |            |
Events entering Engine:     [Event A]    [Event B]    [Event C]    [Event D]
Event Timestamp (Embedded):  (09:59:58)   (10:00:01)   (09:59:50!)  (10:00:03)
                                                           ^
                                                SKEWED LATE EVENT!
                                                (Generated 13 seconds ago,
                                                 delayed by cellular tunnel!)

1. Non-Deterministic Network Delay: Mobile devices enter elevators, switch towers, buffer packets, and drop connections. An event generated at 09:59:50 may physically reach the streaming engine at 10:00:03, completely out of chronological sequence.
2. Infinite Data with Finite Memory: A stream processor cannot hold the entire stream in memory before computing an aggregation. State must be segmented into discrete time windows (e.g. 5-minute tumbling windows) and periodically purged.
3. Failure Recovery Under Mutation: When a streaming worker node crashes mid-stream, how does the system recover its in-memory sliding aggregations without either dropping records (at-most-once) or double-counting financial balances (at-least-once)?

2. Mental Model: The Three Notions of Time

To reason about stream processing correctly, we must untangle physical server clocks from event causality. Modern stream engines (Apache Flink, Google Cloud Dataflow) distinguish three distinct notions of time:

SYSTEM ARCHITECTURE BLUEPRINTDISTRIBUTED SYSTEMS

⤢Expand

ARCHITECTURAL FLOW · STEP 1 OF 3

1. Unbounded Kafka Stream & Watermark Generator

Establishes the ingress and structural boundary across Unbounded Kafka Stream & Watermark Generator: Events carry physical event timestamps: e1(t=12s) -> e2(t=8s) -> e3(t=15s); Out-of-Order Skew: e2 arrived 4s late; due to mobile carrier network lag.

Architectural Invariant: Invariant: "No more events with t <= 10s".

ARCHITECTURAL FLOW · STEP 2 OF 3

2. Keyed Window Operator (Flink) & Window Trigger Evaluation

Coordinates core data-plane execution and state transitions through Keyed Window Operator (Flink) & Window Trigger Evaluation: KeyBy: sensor_id; Window: Tumbling [0 .. 10s); Embedded State: RocksDB on local NVMe.

Architectural Invariant: 3. Session: Dynamic gap inactivity Closes after 15m gap in user events Merges overlapping session windows.

ARCHITECTURAL FLOW · STEP 3 OF 3

3. Asynchronous Barrier Snapshotting & Durable State Snapshot (S3 / HDFS)

Enforces durability, state convergence, and output verification across Asynchronous Barrier Snapshotting & Durable State Snapshot (S3 / HDFS): Coordinator injects Barrier c_1 Barrier flows inline with events Zero stop-the-world pipeline pause! Stream continues processing while saving; On Barrier arrival: Operator flushes; RocksDB incremental SSTable snapshot Sends ACK(c_1) to Coordinator Checkpoint committed once all ACK.

Architectural Invariant: Sink commits database transaction! Guarantees zero duplicate side effects.

+-------------------------------------------------------------------------------+
|                       THE THREE STREAMING TIME ABSTRACTIONS                   |
+-------------------------------------------------------------------------------+
| Notion of Time       | Where Assigned?         | Deterministic?| Resilient?   |
+----------------------+-------------------------+---------------+--------------+
| 1. Event-Time        | At origin (Client device| YES           | 100% immune  |
|                      | sensor clock / payload) | (Reproducible)| to net delays|
|                      |                         |               |              |
| 2. Ingestion-Time    | At message broker       | Moderately    | Sensitive to |
|                      | (Kafka broker entry WAL)|               | broker queues|
|                      |                         |               |              |
| 3. Processing-Time   | At operator CPU machine | NO (Arbitrary | Corrupted by |
|                      | (local server wall-time)| machine clock)| retries/lag  |
+-------------------------------------------------------------------------------+

Why Processing-Time Corrupts Analytics

Consider computing an alert: “Trigger notification if a user makes $>3$ failed login attempts within 5 minutes”:

• Scenario: A user makes 1 failed attempt at 10:00, 10:01, and 10:02 while driving through a mountain pass with no signal. The phone buffers the events locally.
• At 10:30, the car exits the mountain. All 3 events burst into the server simultaneously within a 200-millisecond span.
• Processing-Time Failure: If the engine groups by server arrival time, it evaluates all 3 events inside the 10:30 - 10:35 window and triggers a false security alert!
• Event-Time Correctness: If the engine groups by embedded event timestamps, it correctly segments the events into the historical 10:00 - 10:05 window, preserving true historical causality.

Foundational Bridge: Real-Time Ad-Click Aggregation, Attribution Join Windows & Click-Fraud Deduplication

One of the highest-volume production applications of stateful event-time stream processing is real-time advertising click aggregation and billing attribution (1M+ ad impressions/sec and 50k+ clicks/sec):

1. Idempotent Click Deduplication via Keyed State + TTL: Mobile clients and ad SDKs retry clicks across flaky cellular networks. Before billing an advertiser, the stream processor keys the stream by click_id and checks an EmbeddedRocksDBStateBackend ValueState<Boolean> configured with a 24-hour State TTL (backed by a Bloom filter prefix to avoid unnecessary SSD reads), dropping duplicate retries in $O(1)$ time.
2. Interval Joins for Impression-to-Click Attribution: An ad click at timestamp $t_{\text{click}}$ must be joined against its originating ad impression emitted up to 30 minutes earlier ($t_{\text{imp}}\in [t_{\text{click}}-30\text{ min},t_{\text{click}}]$). Flink’s Interval Join buffers Impression and Click streams in RocksDB keyed by (ad_id, user_id) and uses watermark progression to evict expired impressions once $\text{Watermark}(t)>t_{\text{imp}}+30\text{ min}$.
3. Two-Stage Hot-Key Pre-Aggregation + Exactly-Once Billing Sinks: Viral ad campaigns (campaign_id) create severe Zipfian partition skew. Appending a random salt campaign_id#(0..15) for 10-second pre-aggregation windows followed by a second unsalted keyBy(campaign_id) rollup eliminates TaskManager CPU hotspots while TwoPhaseCommitSink (Kafka / ClickHouse / ledger DB) guarantees zero double-billing across TaskManager failovers.

3. Watermarks: The Clock of the Unbounded Stream

If an engine computes aggregations based on Event-Time, a critical question emerges: When is a time window complete?

If we are aggregating events for the window $[10:00-10:05)$, when can the engine safely close the window, emit the result, and discard the intermediate state from memory?

• If it closes the window at 10:05:01, an event generated at 10:04:58 that arrives at 10:05:05 will be lost.
• If it waits indefinitely for late events, the window can never emit results, causing infinite latency.

The Watermark Invariant

A Watermark $W(t)$ is a special, lightweight control record that flows inline with the data stream:

When an operator processing a window $[0,T)$ receives a watermark $W(t)$ where $t\ge T$:

1. The operator declares that all events for that window have arrived.
2. It executes the window function (e.g., SUM, AVG, COUNT).
3. It emits the final aggregated result downstream.
4. It purges the window’s intermediate state from local RAM/disk to prevent memory leaks.

Watermark Progression through an Operator:
---------------------------------------------------------------------------------
Stream Input:
  (t=7s)  -->  (t=9s)  -->  [Watermark W(10s)]  -->  (t=12s)  -->  (t=14s)

Window [0 .. 10s):
- Receives (t=7s): accumulates value into local state.
- Receives (t=9s): accumulates value into local state.
- Receives [Watermark W(10s)]:
  -> TRIGGER FIRE!
  -> Emits Window Aggregate result for [0 .. 10s).
  -> Cleans up RocksDB state.

Bounded Out-of-Orderness Heuristics

In real-world networks, we cannot know the true physical upper bound of network delay. Stream processors employ a Bounded Out-of-Orderness Watermark Generator:

Where ${\Delta }_{\text{lag}}$ is a tuned latency buffer (e.g. $5\text{ seconds}$):

• If the highest timestamp observed so far is $10:00:15$, the engine emits a watermark of $W(10:00:10)$.
• This gives late-arriving packets a 5-second grace window to navigate network congestion before their corresponding windows close.

Handling Unavoidable Late Data

What happens if an event belonging to a window $[T_{\text{start}},T_{\text{end}})$ arrives after the watermark has already closed that window ($T_{\text{end}}\le W(t)$, or $e_t\le W(t)$ for point-in-time operators)?

1. Drop Silently (Default): The late event is discarded to protect window closure guarantees.
2. Allowed Lateness: The window remains in memory for an additional grace period (e.g., 1 hour). When a late event arrives, the engine updates the aggregate and emits an updated retraction stream.
3. Side Output (Dead-Letter Queue): The late event is diverted into a dedicated secondary Kafka topic for manual reconciliation or auditing.

4. Stateful Stream Windows: Tumbling, Sliding, and Session

Stream aggregations are organized into three primary mathematical window typologies:

+-------------------------------------------------------------------------------+
|                       STREAMING WINDOW TYPOLOGY MATRIX                        |
+-------------------------------------------------------------------------------+
| Window Type     | Overlap? | Boundaries              | Primary Use Case       |
+-----------------+----------+-------------------------+------------------------+
| 1. Tumbling     | NO       | Fixed, non-overlapping  | Hourly revenue totals, |
|                 |          | [0..10), [10..20)       | periodic dashboard sync|
|                 |          |                         |                        |
| 2. Sliding      | YES      | Overlapping fixed size  | 10-minute moving avg   |
|                 |          | Size=10m, Slide=1m      | calculated every min   |
|                 |          |                         |                        |
| 3. Session      | DYNAMIC  | Inactivity gap timeout  | Web user clickstream   |
|                 |          | Closes on 15m idle gap  | browsing sessions      |
+-------------------------------------------------------------------------------+

Sliding Window Memory Amplification

A sliding window of size $W_{\text{size}}$ with slide step $W_{\text{slide}}$ creates overlapping computation:

If calculating a 1-hour moving average that updates every 10 seconds:

Naive implementations duplicate every incoming record into 360 distinct window buckets in memory. Production engines (such as Flink) use Paned / Sliced Windowing:

• The stream is segmented into non-overlapping paned intervals of size $W_{\text{slide}}$ ($10\text{ seconds}$).
• When the 1-hour window fires, the engine aggregates the pre-computed panes, reducing memory consumption by $360\times$!

5. The Chandy-Lamport Distributed Snapshot Algorithm

Stateful stream processing engines maintain gigabytes of state in local memory (e.g. key-value running totals, ML model weights, user session histories). If a machine blows a power supply, how does the distributed engine recover its global state without halting processing or producing inconsistent calculations?

In 1985, K. Mani Chandy and Leslie Lamport formulated the Chandy-Lamport Distributed Snapshot Algorithm: a protocol to record a consistent global state of an asynchronous distributed system without freezing execution.

Apache Flink adapts this principle as Asynchronous Barrier Snapshotting (ABS):

Asynchronous Barrier Snapshotting (ABS) Pipeline:
---------------------------------------------------------------------------------
1. Checkpoint Coordinator periodically injects a Checkpoint Barrier (c_1)
   into the Source Operators.

2. Barriers flow INLINE with normal event records through the operator DAG:
   [Record A] -> [Record B] -> [Barrier c_1] -> [Record C] -> [Record D]

3. Barrier Alignment at Operator with Multiple Inputs:
   Input Channel 1:  [Record B] --> [Barrier c_1]
   Input Channel 2:  [Record Z] --> [Record Y] --> [Barrier c_1]

   - When Barrier c_1 arrives on Channel 1, the operator PAUSES Channel 1.
   - It continues processing Channel 2 until Barrier c_1 arrives on Channel 2!
   - (This guarantees no records from post-checkpoint leak into pre-checkpoint state).

4. State Snapshot:
   - Once all input barriers arrive, the operator takes an asynchronous snapshot
     of its local state (writing to S3 / HDFS in the background).
   - It immediately emits Barrier c_1 downstream.
   - It unpauses all input channels and resumes standard stream processing!

The Invariant of Consistent Cuts

A global snapshot represents a Consistent Cut:

• If an event $e$ is included in the snapshot, the effect of all events that causally preceded $e$ ($e_{\text{prev}}\to e$) are also included in the snapshot.
• No event that occurred after the barrier is included in the snapshot.

Because the state snapshot is flushed to distributed object storage (S3) asynchronously via background threads, stream processing throughput experiences zero stop-the-world pauses.

6. End-to-End Exactly-Once Processing: Two-Phase Commit Sinks

A common fallacy in distributed systems is believing that “internal exactly-once processing” guarantees that end users will never observe duplicate data.

Consider what happens when a stream processor writes to an external PostgreSQL database:

1. Operator processes 100 transaction records and executes INSERT INTO payments ....
2. The database updates the table.
3. Before the checkpoint coordinator confirms that Checkpoint #42 is complete, the stream processor server crashes!
4. The system restarts and restores its internal state from the previous successful checkpoint (Checkpoint #41).
5. The engine replays the last 100 records from Kafka… and inserts them into PostgreSQL a second time!

The TwoPhaseCommitSinkFunction Protocol

To achieve End-to-End Exactly-Once, the stream processor must coordinate with external sink storage using Two-Phase Commit (2PC):

Two-Phase Commit Streaming Sink Lifecycle:
---------------------------------------------------------------------------------
Stream Operator                             External Sink (PostgreSQL / Kafka)
       |                                                    |
[Incoming Barrier c_1]                                      |
       |                                                    |
       |--- 1. Pre-Commit Phase -------------------------->|
       |    Opens Transaction T_1;                          | Writes data into T_1.
       |    Flushes all records prior to c_1.               | Does NOT commit yet!
       |                                                    |
       |--- 2. Snapshot Local State to S3 ----------------->|
       |    Acknowledges Checkpoint c_1 to Coordinator      |
       |                                                    |
[Coordinator broadcasts: "Checkpoint c_1 SUCCESS!"]         |
       |                                                    |
       |--- 3. Formal Commit Phase ------------------------>|
       |    Executes COMMIT on Transaction T_1.             |
       |                                                    | Transaction becomes
       |                                                    | visible to readers!

1. Pre-Commit Phase: When Barrier $c_1$ arrives at the sink, the sink closes the current transaction and opens a new one for subsequent events. The pre-committed transaction is durable, but invisible to external readers.
2. Global Consensus: The checkpoint coordinator waits until all operators across the entire distributed graph successfully persist their state to S3 and ACK Barrier $c_1$.
3. Commit Phase: The coordinator issues a broadcast notification. The sink formally executes COMMIT on the pre-committed transaction. External readers observe the data at the exact moment the global snapshot is guaranteed durable.

7. Local State Backends: In-Memory vs. Embedded RocksDB

Where does a stream processor store state before a checkpoint is triggered?

+-------------------------------------------------------------------------------+
|                       STATE BACKEND COMPARISON (APACHE FLINK)                 |
+-------------------------------------------------------------------------------+
| Backend              | Storage Location   | Capacity Bounds   | Performance   |
+----------------------+--------------------+-------------------+---------------+
| 1. HashMap State     | JVM Heap Memory    | Limited by JVM    | Microsecond   |
|    Backend           | (Java Objects)     | Garbage Collection| latency (fast)|
|                      |                    | (< 32-64 GB)      |               |
|                      |                    |                   |               |
| 2. Embedded RocksDB  | Local NVMe SSD     | Terabytes per node| Sub-millisecond|
|    State Backend     | (Out-of-Core LSM)  | (Limited only by  | (Requires C++ |
|                      |                    | physical disk)    | serialization)|
+----------------------+--------------------+-------------------+---------------+

Production Recommendation: Embedded RocksDB with Incremental Snapshots

For mission-critical production pipelines:

• Use Embedded RocksDB (covered in CORE-M08-003). RocksDB spills data to local NVMe SSDs in Log-Structured Merge Trees, allowing individual worker tasks to maintain tens of gigabytes of state without JVM GC pauses.
• Use Incremental Checkpointing: RocksDB writes new updates to immutable SSTable files. When taking a checkpoint, Flink copies only the newly generated SSTables to S3, reducing checkpoint upload volume by over $90\%$.

8. Stream Joins: Interval Joins & Temporal Tables

Joining two unbounded streams (e.g. matching an AdImpressions stream with an AdClicks stream) is fundamentally different from a relational SQL join:

• In relational databases, both tables are finite and queryable.
• In streaming, neither stream has an end; waiting for a matching click might require holding an impression in memory forever.

The Interval Join

An Interval Join constrains memory by requiring that matching events arrive within a bounded relative time window:

SELECT 
    impressions.ad_id, 
    clicks.click_timestamp, 
    impressions.impression_timestamp
FROM impressions
JOIN clicks ON impressions.ad_id = clicks.ad_id
WHERE clicks.click_timestamp BETWEEN impressions.impression_timestamp 
                                 AND impressions.impression_timestamp + INTERVAL '10' MINUTE;

State Cleanup: Once the upper bound of the interval passes watermark $W(t)$, the engine purges the impression from its RocksDB state, preventing unbounded memory leaks.

9. Operational Failure Modes & Production Troubleshooting

+-------------------------------------------------------------------------------+
|                      COMMON STREAM PROCESSING FAILURE MODES                   |
+-------------------------------------------------------------------------------+
| Symptom                  | Root Cause               | Mitigation Strategy     |
+--------------------------+--------------------------+-------------------------+
| 1. Checkpoint Alignment  | High backpressure causes | Enable Unaligned        |
|    Timeouts (OOM Crash)  | barriers to stall behind | Checkpoints (barriers   |
|                          | full input buffers       | overtake in-flight data)|
|                          |                          |                         |
| 2. Frozen Watermarks     | An idle Kafka partition  | Mark idle partitions via|
|    (No Windows Firing)   | with 0 events halts      | withIdleness(10s);      |
|                          | watermark advancement    | bypass dead partitions  |
|                          |                          |                         |
| 3. Severe Backpressure   | Downstream database sink | Scale sink parallelism; |
|    Cascades              | slow, backing up credit  | optimize batch insert   |
|                          | flow buffers to sources  | sizes; add write caches |
|                          |                          |                         |
| 4. RocksDB Compaction    | Heavy write state churn  | Isolate RocksDB WAL/SST |
|    Disk I/O Stalls       | saturates disk IOPS      | onto dedicated NVMe     |
+-------------------------------------------------------------------------------+

Deep Dive: Unaligned Checkpoints Under Backpressure

Under heavy backpressure, an operator’s input buffers become completely saturated with data. When a Checkpoint Barrier enters the input queue:

• It must wait behind tens of thousands of buffered records before reaching the operator.
• The checkpoint timer expires (typically 10 minutes), causing the checkpoint to fail.
• Repeated checkpoint failures prevent the engine from committing offsets, eventually leading to pipeline crashes.

Remediation: Unaligned Checkpoints: In unaligned mode, the barrier immediately overtakes in-flight records:

1. As soon as the barrier arrives at an input buffer, the operator immediately takes a snapshot of its state and copies the in-flight channel buffers directly into the checkpoint payload.
2. The barrier is forwarded downstream with zero waiting.
3. Checkpoints succeed in milliseconds even under $100\%$ backpressure.

10. Concrete Implementation: Stream Window & Watermark Engine

The following complete, runnable TypeScript implementation demonstrates the core mechanics of a Stateful Stream Processing Engine: injecting out-of-order event streams, computing watermarks, firing tumbling windows, and taking consistent state snapshots.

// stream-engine.ts
export interface StreamEvent {
  id?: string;
  key: string;
  value: number;
  eventTime: number; // Millisecond epoch
}

export interface WindowResult {
  key: string;
  windowStart: number;
  windowEnd: number;
  totalSum: number;
  eventCount: number;
}

export class StatefulStreamEngine {
  private readonly windowSizeMs: number;
  private readonly maxOutOfOrdernessMs: number;

  // Internal Operator State: windowStart -> (key -> { sum, count })
  private windowState = new Map<number, Map<string, { sum: number; count: number }>>();
  private maxObservedTimestamp = -1;
  private currentWatermark = -1;

  constructor(windowSizeMs: number, maxOutOfOrdernessMs: number) {
    this.windowSizeMs = windowSizeMs;
    this.maxOutOfOrdernessMs = maxOutOfOrdernessMs;
  }

  /**
   * Ingests a new event record and updates watermarks and window states
   */
  public processEvent(event: StreamEvent): WindowResult[] {
    // 1. Advance observed event timestamp and compute watermark
    if (event.eventTime > this.maxObservedTimestamp) {
      this.maxObservedTimestamp = event.eventTime;
      this.currentWatermark = this.maxObservedTimestamp - this.maxOutOfOrdernessMs;
    }

    // 2. Check if event is irrevocably late
    const windowStart = Math.floor(event.eventTime / this.windowSizeMs) * this.windowSizeMs;
    const windowEnd = windowStart + this.windowSizeMs;

    if (windowEnd <= this.currentWatermark) {
      // Event arrived after watermark passed window end: dropped as late data
      return [];
    }

    // 3. Accumulate into stateful window
    let keyMap = this.windowState.get(windowStart);
    if (!keyMap) {
      keyMap = new Map();
      this.windowState.set(windowStart, keyMap);
    }

    const current = keyMap.get(event.key) || { sum: 0, count: 0 };
    keyMap.set(event.key, {
      sum: current.sum + event.value,
      count: current.count + 1,
    });

    // 4. Trigger and fire any windows that are now closed by the current watermark
    return this.evaluateWatermarkTriggers();
  }

  /**
   * Advances watermark explicitly (e.g., during idle periods or heartbeats)
   */
  public advanceWatermark(watermarkTimestamp: number): WindowResult[] {
    this.currentWatermark = Math.max(this.currentWatermark, watermarkTimestamp);
    return this.evaluateWatermarkTriggers();
  }

  private evaluateWatermarkTriggers(): WindowResult[] {
    const emittedResults: WindowResult[] = [];
    const closedWindows: number[] = [];

    for (const [windowStart, keyMap] of this.windowState.entries()) {
      const windowEnd = windowStart + this.windowSizeMs;

      // Invariant: If current watermark has passed windowEnd, window is sealed!
      if (this.currentWatermark >= windowEnd) {
        for (const [key, aggregate] of keyMap.entries()) {
          emittedResults.push({
            key,
            windowStart,
            windowEnd,
            totalSum: aggregate.sum,
            eventCount: aggregate.count,
          });
        }
        closedWindows.push(windowStart);
      }
    }

    // Purge sealed windows from memory
    for (const start of closedWindows) {
      this.windowState.delete(start);
    }

    return emittedResults;
  }

  /**
   * Chandy-Lamport / ABS Snapshot: Exports consistent snapshot of active state
   */
  public snapshotState(): string {
    const serializedState: Record<string, Record<string, { sum: number; count: number }>> = {};
    for (const [winStart, keyMap] of this.windowState.entries()) {
      serializedState[winStart] = Object.fromEntries(keyMap.entries());
    }

    return JSON.stringify({
      watermark: this.currentWatermark,
      maxObservedTime: this.maxObservedTimestamp,
      state: serializedState,
    });
  }

  /**
   * Restores operator state from a durable checkpoint snapshot
   */
  public restoreFromSnapshot(snapshotJson: string): void {
    const data = JSON.parse(snapshotJson);
    this.currentWatermark = data.watermark;
    this.maxObservedTimestamp = data.maxObservedTime;
    this.windowState.clear();

    for (const [winStartStr, keyMapObj] of Object.entries(data.state)) {
      const winStart = Number(winStartStr);
      const keyMap = new Map(Object.entries(keyMapObj as Record<string, { sum: number; count: number }>));
      this.windowState.set(winStart, keyMap);
    }
  }

  public getActiveWindowCount(): number {
    return this.windowState.size;
  }

  public getCurrentWatermark(): number {
    return this.currentWatermark;
  }
}

function runDemo(): void {
  console.log("=== Stream Processing: Event Time, Watermarks & Tumbling Windows ===");
  const engine = new StatefulStreamEngine(5000, 2000); // 5s windows, 2s out-of-order tolerance
  console.log(`Initialized StatefulStreamEngine: Window=5000ms, AllowedLateness=2000ms`);
  engine.processEvent({ id: "e1", eventTime: 1000, key: "sensor-A", value: 24.5 });
  engine.processEvent({ id: "e2", eventTime: 4500, key: "sensor-A", value: 25.0 });
  engine.processEvent({ id: "e3", eventTime: 2000, key: "sensor-A", value: 26.5 }); // Out-of-order
  console.log(`Ingested 3 events with out-of-order timestamps (1000, 4500, 2000)`);
  console.log(`Advancing watermark past window boundary to 7000ms...`);
  const windowsFired = engine.advanceWatermark(7000);
  console.log(`Windows triggered: ${windowsFired.length}`);
  for (const win of windowsFired) {
    console.log(`  Window [${win.windowStart}ms - ${win.windowEnd}ms]: key='${win.key}', count=${win.eventCount}, sum=${win.totalSum}`);
  }
}

runDemo();

Simulation Execution Output & State Transitions

=== Stream Processing: Event Time, Watermarks & Tumbling Windows ===
Initialized StatefulStreamEngine: Window=5000ms, AllowedLateness=2000ms
Ingested 3 events with out-of-order timestamps (1000, 4500, 2000)
Advancing watermark past window boundary to 7000ms...
Windows triggered: 1
  Window [0ms - 5000ms]: key='sensor-A', count=3, sum=76

11. Exercises with Observable Deliverables

Exercise 1: Simulate Out-of-Order Watermark Progression

Deliverable: Using StatefulStreamEngine:

1. Initialize an engine with windowSizeMs = 10000 (10 seconds) and maxOutOfOrdernessMs = 3000 (3 seconds delay).
2. Ingest the following event sequence for key "sensor_1":
   • Event 1: value = 10, eventTime = 1000
   • Event 2: value = 20, eventTime = 4000
   • Event 3: value = 30, eventTime = 2000 (out of order, but within 3s bound!)
   • Event 4: value = 40, eventTime = 13500 (advances max timestamp to 13500; watermark becomes $13500-3000=10500$)
3. Verify that Window [0 .. 10000) fires immediately upon Event 4, emitting totalSum = 60 and eventCount = 3.
4. Ingest Event 5: value = 50, eventTime = 5000 (arrives after watermark 10500). Verify that it is correctly rejected as late data.

Exercise 2: Verify Chandy-Lamport State Recovery

Deliverable:

1. Take a snapshot of the engine using snapshotState() after ingesting initial records.
2. Intentionally mutate or destroy the active engine instance.
3. Instantiate a new StatefulStreamEngine and execute restoreFromSnapshot().
4. Resume processing and assert that subsequent window calculations yield identical results with zero lost state.

12. Summary & The Streaming Architect’s Rules of Thumb

+-------------------------------------------------------------------------------+
|                    THE STREAM PROCESSING DECISION GUIDE                       |
+-------------------------------------------------------------------------------+
| Requirement               | Recommended Architecture                          |
+---------------------------+---------------------------------------------------+
| Time Semantics            | Always Event-Time; never Processing-Time for ETL  |
| Watermark Strategy        | Bounded out-of-orderness (e.g. 5s - 30s buffer)   |
| State Backend             | Embedded RocksDB with incremental SSD snapshots   |
| Large Scale Windows       | Sliced / Paned windowing to prevent memory bloat  |
| Non-Blocking Snapshots    | Chandy-Lamport Asynchronous Barrier Snapshotting  |
| End-to-End Exactly-Once   | Idempotent source + 2PC Transactional Sink        |
| Backpressure Defense      | Unaligned Checkpoints + Credit-based TCP buffers  |
| Idle Partitions           | Explicit withIdleness() to prevent watermark stall|
+-------------------------------------------------------------------------------+

1. Never Depend on Processing-Time for Business Correctness: Processing-Time couples analytical correctness to unpredictable network latency and machine load. Always embed event timestamps at the source and process in Event-Time.
2. Watermarks Are Latency-Completeness Knobs: A smaller watermark delay ($\Delta =1\text{s}$) produces faster alerts but increases late-data drops. A larger delay ($\Delta =60\text{s}$) guarantees completeness but introduces a 60-second lag to all downstream analytics.
3. Internal Exactly-Once is Not End-to-End: A streaming engine that guarantees internal exactly-once will still write duplicates to databases during crash recovery unless the sink implements Two-Phase Commit (2PC) or idempotent upserts (ON CONFLICT DO UPDATE).
4. Isolate State from Garbage Collection: Never store hundreds of gigabytes of streaming state in the JVM heap. Use embedded out-of-core LSM engines (RocksDB) to bypass GC stop-the-world pauses.

13. Design Alternatives & Stream Architecture Trade-Off Matrix

14. Authoritative References & Version Context

• Tyler Akidau et al. (2015), “The Dataflow Model: A Practical Approach to Balancing Correctness, Latency, and Cost in Massive-Scale, Unbounded, Out-of-Order Data Processing”, VLDB ‘15 — doi.org/10.14778/2824032.2824076
• Paris Carbone et al. (2015), “Apache Flink: Stream and Batch Processing in a Single Engine”, IEEE Data Engineering Bulletin — data-artisans.com
• Apache Flink 1.18/1.19 Official Documentation — State & Fault Tolerance — nightlies.apache.org/flink/flink-docs-master/docs/dev/datastream/fault-tolerance/checkpointing/
• Version Context: Stream engine architecture reflects Apache Flink 1.18+ / 2.0 with embedded RocksDB state backends and Kafka 3.7+ sources.