Stateful Stream Processing Internals: Exactly-Once Semantics with Chandy-Lamport Snapshots

In high-scale enterprise architectures, batch processing (nightly ETL pipelines) is no longer sufficient. Businesses require continuous, real-time computation over unbounded streams of data: fraud detection during credit card swipes, dynamic ride-share surge pricing, automated trading risk calculations, and real-time AI feature engineering.

Stateless stream processing—such as filtering logs or mapping JSON keys—is relatively simple. However, real-world business logic is almost always stateful: calculating a user's running transaction count over a rolling 1-hour window, tracking user session funnels, or aggregating multi-tenant metrics.

When stateful stream processing runs across dozens of distributed worker nodes, a terrifying problem emerges: What happens when a node crashes in the middle of processing 10 million events? How do you recover distributed state without losing events (At-Least-Once) and without double-counting transactions (At-Most-Once)?

The solution is Exactly-Once Processing Semantics (EOS), powered by the Chandy-Lamport distributed snapshot algorithm and Asynchronous Checkpoint Barrier Alignment. Today, engines like Apache Flink, Kafka Streams, and RisingWave process billions of events per second with mathematically proven consistency guarantees.

In this technical deep dive, we dissect the internal architecture of stateful stream engines, master Event Time vs. Processing Time, explore Watermarks for out-of-order data, examine RocksDB State Backends, break down Chandy-Lamport Checkpointing, and analyze Two-Phase Commit (2PC) Sinks.

The Three Notions of Time: Why Wall-Clock Time Fails

In distributed stream processing, the physical clock on a server (Wall-Clock Time) is fundamentally untrustworthy. Network jitter, cellular packet reordering, and mobile device offline buffering mean events frequently arrive out of order.

Watermarks: Handling Out-of-Order & Late-Arriving Events

When aggregating events by Event Time into time windows (e.g. 12:00 - 12:10), how does a streaming operator know when all events for that window have arrived so it can safely emit the final result and close the window?

Streaming engines solve this using Watermarks ($\mathcal{W}(t)$). A Watermark is a specialized control element injected into the data stream that asserts: "We assume no future events with event timestamp $\le t$ will arrive."

Out-of-Order Stream with Bounded Out-of-Orderness Watermarks (Lag = 5s):

Event Stream Timeline:
[e: 12:00:03] ──> [e: 12:00:08] ──> [e: 12:00:05 (Late!)] ──> [ Watermark W(12:00:03) ] ──> [e: 12:00:14] ──> [ Watermark W(12:00:09) ]
                                                                             │
                                                                             ▼
                                                     Closes 12:00:00 - 12:00:05 Window & Fires Aggregate!

If an event arrives after the Watermark has passed (anomalous late data), the engine can either:

1. Drop the event silently.
2. Emit a window update / correction.
3. Redirect late events to a Dead Letter Queue (Side Output) for manual reconciliation.

State Management: Heap vs. Embedded RocksDB State Backends

Stateful operators (aggregations, joins, pattern matchers) must persist gigabytes or terabytes of active state. Streaming engines employ two primary state backend architectures:

Chandy-Lamport Distributed Snapshots & Barrier Alignment

The crown jewel of modern stream processing is the Asynchronous Barrier Snapshotting (ABS) algorithm (an optimized variant of the 1985 Chandy-Lamport distributed snapshot algorithm).

Rather than pausing the entire distributed cluster (Stop-The-World) to take a backup, Checkpoint Barriers ($S_n$) are injected into the input streams at the sources and flow downstream alongside regular data records:

Asynchronous Checkpoint Barrier Alignment in a DAG Operator:

Channel 1: [ Event D ] ───> [ Barrier S_1 ] ───> [ Event B ] ───> [ Event A ]
                                    │
Channel 2: [ Barrier S_1 ] ───> [ Event C ]
                 │
                 ▼
┌────────────────────────────────────────────────────────┐
│ Stateful Operator (e.g., Rolling Sum)                  │
│                                                        │
│ 1. Receives Barrier S_1 from Channel 2                 │
│ 2. Pauses processing Channel 2 (Buffers incoming data) │
│ 3. Continues processing Channel 1 until Barrier S_1    │
│ 4. BOTH Barriers Arrived! -> Take Instant State Freeze │
│ 5. Asynchronously writes State snapshot to S3 / HDFS    │
│ 6. Unbuffers Channel 2 and emits Barrier S_1 downstream│
└────────────────────────────────────────────────────────┘

Why Barrier Alignment Guarantees Exactly-Once Consistency:

• All records before the barrier are included in Snapshot $n$.
• All records after the barrier are strictly excluded from Snapshot $n$ and will be part of Snapshot $n+1$.
• State persistence to durable storage (AWS S3, Google Cloud Storage, Ceph) happens asynchronously in the background without blocking active record throughput!

End-to-End Exactly-Once: The Two-Phase Commit (2PC) Sink

A streaming engine can guarantee internal exactly-once state recovery, but if it writes output records to an external system (e.g. PostgreSQL, Kafka, or ClickHouse), a worker crash could cause external systems to receive duplicate records during restart.

True End-to-End Exactly-Once Processing requires pairing checkpointing with a Two-Phase Commit (2PC) Sink:

Two-Phase Commit (2PC) Streaming Sink Protocol:

Phase 1: Pre-Commit (During Checkpoint Barrier Arrival)
  • Sink opens a transaction: TX_42.
  • Writes all processed output records inside TX_42.
  • When Barrier S_1 arrives, sink flushes and pre-commits TX_42.
  • Writes TX_42 ID into the operator's checkpoint state metadata.

Phase 2: Final Commit (After Coordinator Confirms Global Checkpoint Success)
  • Job Manager notifies all operators: "Checkpoint 1 is durable on S3!"
  • Sink officially executes: COMMIT TX_42!
  • Output records are now visible to external downstream consumers!

If a crash occurs before Phase 2, the system restarts from Checkpoint $n-1$ and rolls back/aborts any uncommitted transactions, guaranteeing that external databases never observe ghost or duplicated records.

Architectural Comparison: Apache Flink vs. Kafka Streams vs. RisingWave

Hands-On: Building an Event-Time Window Aggregator in Python

Below is a working implementation demonstrating Event Time extraction, Bounded Out-of-Order Watermarking, and Tumbling Window aggregation:

from dataclasses import dataclass
from typing import List, Dict
import time

@dataclass
class Event:
    user_id: str
    amount: float
    event_time: int  # Epoch timestamp in seconds

class TumblingEventTimeWindow:
    def __init__(self, window_size_sec: int = 10, max_lateness_sec: int = 3):
        self.window_size = window_size_sec
        self.max_lateness = max_lateness_sec
        self.current_watermark = 0
        # WindowStart -> { user_id: total_amount }
        self.windows: Dict[int, Dict[str, float]] = {}

    def process_event(self, event: Event):
        # 1. Update Watermark (Monotonically increasing heuristic: max_time - max_lateness)
        if event.event_time - self.max_lateness > self.current_watermark:
            self.current_watermark = event.event_time - self.max_lateness
            self._trigger_watermark()

        # 2. Check if event is too late (behind current watermark)
        window_start = (event.event_time // self.window_size) * self.window_size
        window_end = window_start + self.window_size

        if window_end <= self.current_watermark:
            print(f"[DROPPED LATE EVENT] Event for {event.user_id} at {event.event_time}s arrived after Watermark {self.current_watermark}s!")
            return

        # 3. Aggregate into window state
        if window_start not in self.windows:
            self.windows[window_start] = {}

        state = self.windows[window_start]
        state[event.user_id] = state.get(event.user_id, 0.0) + event.amount
        print(f"[ACCUMULATED] User: {event.user_id}, Amount: ${event.amount:.2f} in Window [{window_start}s - {window_end}s]")

    def _trigger_watermark(self):
        # Close all windows whose window_end <= current_watermark
        closed_windows = [w for w in self.windows.keys() if w + self.window_size <= self.current_watermark]
        for w_start in sorted(closed_windows):
            w_end = w_start + self.window_size
            results = self.windows.pop(w_start)
            print(f"\n==== EMITTING FINAL WINDOW [{w_start}s - {w_end}s] (Watermark: {self.current_watermark}s) ====")
            for user, total in results.items():
                print(f"  -> User: {user} | Total Spent: ${total:.2f}")
            print("=================================================================\n")

# --- Simulation of Out-of-Order Stream ---
if __name__ == "__main__":
    aggregator = TumblingEventTimeWindow(window_size_sec=10, max_lateness_sec=3)
    
    stream = [
        Event("alice", 50.0, 100),  # Window [100 - 110]
        Event("bob", 30.0, 102),    # Window [100 - 110]
        Event("alice", 20.0, 101),  # Out-of-order! (101 arrives after 102) -> Window [100 - 110]
        Event("alice", 15.0, 112),  # Window [110 - 120] -> Advances Watermark to 109s
        Event("bob", 40.0, 115),    # Window [110 - 120] -> Advances Watermark to 112s (Closes [100-110]!)
        Event("charlie", 99.0, 103),# LATE ARRIVAL! Window [100-110] already closed -> DROPPED!
    ]

    for event in stream:
        aggregator.process_event(event)

Summary & Engineering Checklist

• Standardize on Event Time: Never aggregate historical or mobile telemetry using Processing Time. Always extract event timestamps from data records and generate bounded out-of-order watermarks.
• Isolate State with RocksDB: For production workloads exceeding tens of gigabytes of operator state, migrate from in-memory JVM heaps to embedded RocksDB state backends to eliminate GC pauses.
• Configure Checkpoint Intervals Thoughtfully: Align checkpoint intervals (e.g. every 10–30 seconds) to balance recovery time (RTO) against S3 I/O write amplification.
• Enforce 2PC Sinks for Financial/Critical Data: Guarantee true end-to-end consistency by pairing Chandy-Lamport snapshots with transactional Two-Phase Commit sinks.