Architecture Study: Apache Flink — Distributed State, Checkpoints & Exactly-Once Streaming

1. Problem & Architecture Scope

Project Identity, Version & License Baseline

Modern real-time applications—ranging from financial fraud detection and algorithmic trading to dynamic ride-pricing and IoT telemetry—require stream processing engines capable of:

1. Low Latency & High Throughput: Processing millions of events per second with single-digit millisecond latency.
2. Arbitrary Stateful Computations: Maintaining large state (e.g., 30-day user sliding windows, machine learning feature accumulators) that vastly exceeds the physical RAM of a single server.
3. Deterministic Fault Tolerance: Guaranteeing that in the event of hardware failures, worker crashes, or network partitions, computation results are mathematically identical to a failure-free execution (Exactly-Once Processing Semantics).

Early streaming engines failed on these requirements:

• Micro-Batch Engines (e.g., Spark Streaming v1): Incurred 100 ms–1,000 ms latency overhead by treating streaming as a continuous series of tiny batch jobs.
• At-Least-Once Processing (e.g., Apache Storm): Relied on record-level acknowledgments. Upon failure, upstream sources replayed entire event sequences, causing duplicated side effects, incorrect aggregations, and double-counted financial transactions.

Apache Flink was engineered from the ground up as a true pipelined, event-driven stream processor where batch is merely a special case of streaming (bounded streams).

This architecture study examines Flink’s core internals:

• Distributed Runtime Topology: JobManager, TaskManagers, and Task Slot allocation.
• Asynchronous Barrier Snapshotting (ABS): Flink’s adaptation of the Chandy-Lamport algorithm for continuous distributed checkpointing.
• State Backends: Heap-based state versus out-of-core incremental RocksDB storage.
• End-to-End Exactly-Once: Two-Phase Commit protocol coordination across Kafka sources and sinks.

2. Mental Model & Architectural Abstraction

Flink models computation as a Stateful Dataflow Graph:

[ Kafka Source ] ──(record)──> [ KeyBy Operator ] ──(keyed stream)──> [ Stateful Window ] ──> [ Exactly-Once Sink ]
       │                                │                                      │                          │
       │                                ▼                                      ▼                          ▼
       │                        [ Keyed State ]                        [ Keyed State ]                    │
       │                        (StateBackend)                         (StateBackend)                     │
       │                                │                                      │                          │
       ▼                                ▼                                      ▼                          ▼
=============================================================================================================
                              CHANDY-LAMPORT ASYNCHRONOUS BARRIER SNAPSHOTTING
=============================================================================================================
[ Barrier 42 ] ──────────────> [ Barrier 42 ] ───────────────────────> [ Barrier 42 ] ──────────> [ Commit 42 ]
(Snapshot Kafka Offsets)        (Snapshot Operator State)               (Snapshot Window State)     (Commit Sink Txn)

• Dataflow Operators as State Machines: Every transformation (filter, map, join, aggregate) can maintain state. The state is not stored in an external database; it lives locally on the same machine running the operator.
• Streams Partitioned by Key: Using keyBy(), Flink partitions the data stream by key into parallel substreams, binding state keys deterministically to specific worker task slots.
• Barriers as Logical Clock Ticks: Checkpoint barriers are special meta-events flowing through the operator pipeline. They do not interrupt record processing; they act as distributed logical delimiters dividing the continuous stream into discrete epochs ($e_n,e_{n+1}$).

3. Core Terminology & Guarantees

4. System Architecture & Component Anatomy

SYSTEM ARCHITECTURE BLUEPRINTDISTRIBUTED SYSTEMS

⤢Expand

ARCHITECTURAL FLOW · STEP 1 OF 3

1. JobManager Control Plane & Dispatcher and JobMaster

Establishes the ingress and structural boundary across JobManager Control Plane & Dispatcher and JobMaster: Transforms JobGraph into ExecutionGraph; K8s / YARN Worker TaskSlot Allocation; Triggers periodic Checkpoint Barriers Collects State Handles from Operators Commits metadata upon all-task ACKs.

Architectural Invariant: ZooKeeper / Kubernetes Leader Election Zero single point of failure in control tier.

ARCHITECTURAL FLOW · STEP 2 OF 3

2. TaskManager 1 (Slot 1 and 2) & Source: Kafka Topic (P0)

Coordinates core data-plane execution and state transitions through TaskManager 1 (Slot 1 and 2) & Source: Kafka Topic (P0): KeyedState: Sliding Windows / Joins; Flushes SSTs asynchronously to S3; Downstream operator grants input credits to upstream producer Zero credits = upstream stops reading; propagates back to Kafka consumer Eliminates out-of-memory crashes during downstream data skews.

Architectural Invariant: Downstream operator grants input credits to upstream producer Zero credits = upstream stops reading; propagates back to Kafka consumer Eliminates out-of-memory crashes during downstream data skews.

ARCHITECTURAL FLOW · STEP 3 OF 3

3. Asynchronous Snapshots & Chandy-Lamport Barriers

Enforces durability, state convergence, and output verification across Asynchronous Snapshots & Chandy-Lamport Barriers: Injected into data stream Separates epoch n from epoch n+1; Aligned: Buffers fast input channels; Unaligned: Snapshot in-flight buffers.

Architectural Invariant: 3. Pending uncommitted sink transactions are aborted Zero Duplicate Writes, Zero Data Loss.

Flink Runtime Components:

1. JobManager (Master Node):
   • Dispatcher: Provides a REST API accepting user JobGraph submissions.
   • JobMaster: Coordinates execution of a single JobGraph, scheduling tasks and monitoring health.
   • Checkpoint Coordinator: Injects periodic checkpoint barriers, tracks worker state handles, and writes committed snapshot metadata to Amazon S3 or HDFS.
   • ResourceManager: Negotiates worker containers and task slots from Kubernetes, YARN, or standalone pools.
2. TaskManagers (Worker Nodes):
   • Task Slots: Execute assigned operator tasks (e.g., FlatMapTask, WindowOperatorTask).
   • Network Stack: Netty-based multiplexed TCP transport with partitioned memory buffer pools (Input Channels and Result Partitions).
   • State Engine: Manages local memory and disk storage for stateful transformations.
3. Checkpoint Storage Tier:
   • Highly durable distributed object storage (Amazon S3, Google Cloud Storage, or Ceph) holding immutable operator state SST files and metadata.

5. Detailed Request Flows & Protocols

5.1 The Chandy-Lamport Checkpoint Flow

CheckpointCoordinator         Source Task               Keyed Operator               Sink Task (2PC)
          │                        │                           │                           │
          ├── 1. Trigger CP(Epoch 42) ─>│                       │                           │
          │                        ├── 2. Snapshot Source Offsets                       │
          │                        │      Emit Barrier(42) ───>│                           │
          │                        │                           ├── 3. Barrier Alignment    │
          │                        │                           │      Snapshot Local State │
          │                        │                           │      Emit Barrier(42) ───>│
          │                        │                           │                           ├── 4. Pre-Commit Txn
          │<── State Handle ACK ───┤                           │                           │      Snapshot Sink State
          │<── State Handle ACK ───────────────────────────────┤                           │
          │<── State Handle ACK ───────────────────────────────────────────────────────────┤
          │                                                                                │
          ├── 5. All ACKs Received: Commit Checkpoint 42 to S3                             │
          │                                                                                │
          ├──────────────────────── 6. notifyCheckpointComplete(42) ──────────────────────>│
          │                                                                                ├── 7. Formal Commit Txn

1. Triggering: The Checkpoint Coordinator on the JobManager initiates checkpoint epoch $42$, sending trigger RPCs to all stream sources.
2. Barrier Injection: Sources record their current message offsets (e.g., Kafka partition 0 offset 109,241), persist their state handle, and inject Barrier(42) into all output channels.
3. Barrier Alignment:
   • When an operator receives Barrier(42) from one input channel, it temporarily buffers records from that channel while continuing to read from other channels until Barrier(42) arrives from all inputs (Aligned Checkpoint).
   • Once aligned, the operator takes an instantaneous snapshot of its keyed state (e.g., flushes memtables to SST files in RocksDB) and immediately forwards Barrier(42) downstream.
4. Pre-Commit at Sink: When Barrier(42) reaches the TwoPhaseCommitSink, it initiates a pre-commit on the external transaction (e.g., Kafka transaction flink-txn-42).
5. Global Commit: Once the Checkpoint Coordinator collects state confirmation ACKs from every task, it writes the master checkpoint metadata file to durable S3.
6. Formal Commit: The Coordinator broadcasts notifyCheckpointComplete(42) to sinks. The sink formally commits the Kafka transaction, making output records visible to consumers.

6. Mathematical Modeling & Hardware Physics

6.1 State Memory Sizing & Sizing Equations

Let:

• $N_{\text{keys}}$ be the number of active distinct keys in the stream.
• $S_{\text{entry}}$ be the average size of state per key in bytes.
• $W_{\text{duration}}$ be the sliding window length in seconds.
• $R_{\text{ingest}}$ be the event ingestion rate (events/second).

For a keyed window aggregating user events over duration $W_{\text{duration}}$:

If $N_{\text{keys}}=10,000,000$, $S_{\text{entry}}=256\text{ bytes}$, and no raw events are retained: $$\text{Keyed State}=10^7\times 256\text{ bytes}=2.56\text{ GB (per partition group)}$$

Across 100 parallel worker slots with 20 stateful operators: $$\text{Total Cluster State}\approx 5.12\text{ TB}$$

Because 5 TB exceeds practical cluster RAM budgets, Flink offloads keyed state to NVMe SSDs via EmbeddedRocksDBStateBackend.

6.2 RocksDB Incremental Checkpoint Network Physics

In naive full checkpointing, an operator with 100 GB of state must upload 100 GB over the network to S3 every 10 seconds, demanding:

Under Incremental Checkpoints:

• RocksDB flushes immutable SST files to disk.
• Only newly generated SST files (created by MemTable flushes and level compactions during epoch $n$) are uploaded.
• Older SST files are referenced in the checkpoint metadata by their immutable S3 URI.
• For an update mutation rate of $5\%$, network transfer drops to:

7. Production TypeScript Implementation

This self-contained TypeScript simulation implements:

1. ChandyLamportOperator: Simulates barrier alignment, keyed state mutations, and asynchronous state snapshots.
2. TwoPhaseCommitSink: Simulates external transactional sinks (Kafka 2PC).
3. StreamCheckpointCoordinator: Coordinates global distributed checkpoint epochs and recovery rollbacks.

Execute with: npx tsx flink-stream-capstone.ts

/**
 * ARCHITECTURE STUDY: Apache Flink Distributed Stream Architecture
 * Simulation of Chandy-Lamport Barrier Alignment, Keyed State,
 * Two-Phase Commit Sink, and Fault Recovery Rollbacks.
 */

// ==========================================
// 1. Stream Event & Barrier Definitions
// ==========================================

export type StreamItem = 
  | { type: 'RECORD'; key: string; value: number; timestamp: number }
  | { type: 'BARRIER'; epoch: number };

export interface StateHandle {
  operatorId: string;
  epoch: number;
  stateDump: Record<string, number>;
}

// ==========================================
// 2. Chandy-Lamport Operator with Keyed State
// ==========================================

export class ChandyLamportOperator {
  public id: string;
  private keyedState: Map<string, number> = new Map();
  private inputChannels: number;
  private blockedChannels: Set<number> = new Set();
  private activeBarrierEpoch: number | null = null;
  private alignmentBuffer: StreamItem[] = [];

  constructor(id: string, inputChannels: number = 2) {
    this.id = id;
    this.inputChannels = inputChannels;
  }

  processItem(channelId: number, item: StreamItem): { emit: StreamItem[]; snapshot?: StateHandle } {
    if (item.type === 'RECORD') {
      // If this channel already delivered its barrier for the active epoch, buffer the record
      // until all remaining channels align so post-barrier records do not pollute the snapshot!
      if (this.blockedChannels.has(channelId)) {
        this.alignmentBuffer.push(item);
        return { emit: [] };
      }

      // Update local keyed state (running sum aggregation)
      const current = this.keyedState.get(item.key) || 0;
      this.keyedState.set(item.key, current + item.value);
      return { emit: [item] };
    }

    // Item is a Checkpoint Barrier
    const epoch = item.epoch;
    this.activeBarrierEpoch = epoch;
    this.blockedChannels.add(channelId);

    // Barrier Alignment: wait until barrier arrives on ALL input channels
    if (this.blockedChannels.size === this.inputChannels) {
      this.blockedChannels.clear();
      this.activeBarrierEpoch = null;

      // Perform local snapshot BEFORE replaying buffered post-barrier records
      const snapshot: StateHandle = {
        operatorId: this.id,
        epoch,
        stateDump: Object.fromEntries(this.keyedState),
      };

      const emitted: StreamItem[] = [{ type: 'BARRIER', epoch }];

      // Replay buffered records that arrived on blocked channels during alignment
      const buffered = [...this.alignmentBuffer];
      this.alignmentBuffer = [];
      for (const bufferedItem of buffered) {
        const replayed = this.processItem(0, bufferedItem);
        emitted.push(...replayed.emit);
      }

      // Forward barrier and any replayed post-barrier records downstream
      return { emit: emitted, snapshot };
    }

    // Waiting for alignment: do not forward barrier yet
    return { emit: [] };
  }

  restore(snapshot: StateHandle): void {
    this.keyedState = new Map(Object.entries(snapshot.stateDump));
  }

  getState(key: string): number | undefined {
    return this.keyedState.get(key);
  }
}

// ==========================================
// 3. Two-Phase Commit Sink (Kafka 2PC)
// ==========================================

export class TwoPhaseCommitSink {
  public id: string;
  private pendingTransactions: Map<number, Array<{ key: string; value: number }>> = new Map();
  public committedData: Array<{ key: string; value: number }> = [];

  constructor(id: string) {
    this.id = id;
  }

  processItem(item: StreamItem, currentEpoch: number): { snapshotPending?: boolean } {
    if (item.type === 'RECORD') {
      if (!this.pendingTransactions.has(currentEpoch)) {
        this.pendingTransactions.set(currentEpoch, []);
      }
      this.pendingTransactions.get(currentEpoch)!.push({ key: item.key, value: item.value });
      return {};
    }

    // Barrier arrival triggers Phase 1: Pre-Commit
    return { snapshotPending: true };
  }

  commit(epoch: number): void {
    // Phase 2: Formal Commit upon JobManager ACK
    const tx = this.pendingTransactions.get(epoch);
    if (tx) {
      this.committedData.push(...tx);
      this.pendingTransactions.delete(epoch);
    }
  }

  abort(epoch: number): void {
    this.pendingTransactions.delete(epoch);
  }
}

// ==========================================
// 4. Checkpoint Coordinator & Harness
// ==========================================

export function runFlinkDemo() {
  console.log("=== ARCHITECTURE STUDY: Apache Flink Distributed Streaming ===");

  const operator = new ChandyLamportOperator("window-aggregator", 2); // 2 upstream channels
  const sink = new TwoPhaseCommitSink("kafka-sink-2pc");

  console.log("\n[1] Processing Stream Records on Channel 0 & Channel 1:");
  operator.processItem(0, { type: 'RECORD', key: 'user:101', value: 50, timestamp: 1000 });
  operator.processItem(1, { type: 'RECORD', key: 'user:102', value: 80, timestamp: 1001 });
  operator.processItem(0, { type: 'RECORD', key: 'user:101', value: 25, timestamp: 1002 });

  console.log(`  Current State for 'user:101': ${operator.getState('user:101')}`);
  console.log(`  Current State for 'user:102': ${operator.getState('user:102')}`);

  console.log("\n[2] Triggering Checkpoint Barrier (Epoch 1):");
  // Channel 0 receives barrier first
  console.log("  >> Barrier 1 arrives on Channel 0...");
  const res1 = operator.processItem(0, { type: 'BARRIER', epoch: 1 });
  console.log(`  Aligned yet? ${res1.snapshot !== undefined ? 'YES' : 'NO (Waiting for Channel 1)'}`);

  // Post-barrier Epoch 2 record arrives early on blocked Channel 0 -> buffered in alignmentBuffer!
  operator.processItem(0, { type: 'RECORD', key: 'user:101', value: 100, timestamp: 1003 });
  console.log(`  Buffered post-barrier record on Channel 0; pre-alignment state for 'user:101' remains: ${operator.getState('user:101')}`);

  // Channel 1 receives barrier -> Alignment complete! Snapshot saved at 75, then buffered record (+100) replayed!
  console.log("  >> Barrier 1 arrives on Channel 1...");
  const res2 = operator.processItem(1, { type: 'BARRIER', epoch: 1 });
  console.log(`  Aligned yet? ${res2.snapshot !== undefined ? 'YES' : 'NO'}`);
  console.log("  Operator Epoch 1 Snapshot Dump:", res2.snapshot?.stateDump);
  console.log(`  Post-alignment live state for 'user:101' after buffer replay: ${operator.getState('user:101')}`);

  // Forward to Sink -> Phase 1 Pre-Commit
  sink.processItem({ type: 'RECORD', key: 'user:101', value: 75 }, 1);
  sink.processItem({ type: 'BARRIER', epoch: 1 }, 1);
  console.log(`  Sink Committed Records before Master ACK: ${sink.committedData.length}`);

  // Coordinator receives all ACKs -> Phase 2 Formal Commit
  console.log("\n[3] JobManager persists Checkpoint 1 metadata to S3 and broadcasts Commit:");
  sink.commit(1);
  console.log(`  Sink Committed Records after Master ACK: ${sink.committedData.length}`);
  console.log("  Committed Record Data:", sink.committedData);

  // Failure and Rollback demonstration
  console.log("\n[4] Simulating Worker Crash during Epoch 2 (Fault Recovery):");
  operator.processItem(0, { type: 'RECORD', key: 'user:101', value: 999, timestamp: 2000 }); // uncommitted mutation
  console.log(`  Corrupted state before rollback: ${operator.getState('user:101')}`);

  console.log("  Rolling back operator to Checkpoint 1 state cut...");
  operator.restore(res2.snapshot!);
  sink.abort(2);
  console.log(`  Restored state for 'user:101': ${operator.getState('user:101')}`);
  console.log("\n=== Flink Simulation Complete. Invariants Verified. ===");
}

runFlinkDemo();

8. Real-World Case Studies

8.1 Uber (Real-Time Dynamic Pricing & Fraud)

• Runs one of the world’s largest Apache Flink deployments: over 10,000 Flink jobs running on Apache YARN and Kubernetes.
• Processes tens of petabytes of stream data per day with billions of events per second.
• Leverages RocksDB state backends with incremental checkpoints to S3, reducing checkpoint duration from 15 minutes to under 20 seconds.

8.2 Netflix (Keystone Stream Pipeline)

• Processes over 3 trillion events per day using Flink for telemetry aggregation, playhead positioning, and distributed personalization.
• Uses Savepoints (user-triggered, externally addressable checkpoint snapshots) to execute zero-downtime application code upgrades, schema migrations, and cluster resharding without losing window state.

8.3 Alibaba (Blink & E-Commerce Search)

• Created “Blink” (later merged into Apache Flink), optimizing Flink SQL, fine-grained failover, and credit-based network flow control to power the massive 11.11 Global Shopping Festival.

9. Edge Cases, Failure Modes & Antipatterns

9.1 Barrier Alignment Latency Spikes

• Failure Mode: When one input partition is heavily delayed (e.g., due to Kafka broker rebalance), an operator waiting for the lagging channel buffers megabytes of records from active channels, inflating heap memory and increasing end-to-end latency.
• Remediation: Enable Unaligned Checkpoints (execution.checkpointing.unaligned.enabled: true). Operators take snapshots of in-flight channel buffers immediately without waiting for alignment, trading slightly larger snapshot sizes for zero alignment delays.

9.2 RocksDB Serialization Bottlenecks

• Failure Mode: Operators performing complex transformations (e.g., reading and mutating large Java POJO lists in ValueState) incur massive CPU serialization and deserialization overhead on every record read/write.
• Remediation: Prefer MapState or ListState over monolithic ValueState holding large collections. Flink can read or append single entries in RocksDB without deserializing the entire collection.

9.3 Checkpoint Expiration & Timeout Thrashing

• Failure Mode: Under heavy backpressure, if an operator takes longer than checkpoint.timeout (e.g., 10 minutes) to align barriers and upload state, the JobManager cancels the checkpoint and immediately triggers another one, causing a continuous spiral of wasted CPU and disk I/O.
• Remediation: Configure minPauseBetweenCheckpoints (e.g., at least 2 minutes). This guarantees operators receive a guaranteed processing breather between consecutive snapshot runs.

10. Observability, Metrics & Production Telemetry

11. Guided Hands-On Lab / Engineering Exercises

Lab 1: Implement an Unaligned Checkpoint Channel Buffer Snapshot

• Objective: Extend ChandyLamportOperator to support unaligned checkpoints. When a barrier arrives on Channel 0, snapshot all in-flight buffers on Channel 1 without waiting for alignment.
• Deliverable: Verify that downstream processing latency is completely unaffected by channel skew.

Lab 2: Benchmark Aligned vs. Unaligned Checkpoints Under Heavy Skew

• Objective: Simulate two input channels where Channel 1 processes 10,000 records/sec and Channel 2 has an injected 2-second sleep delay.
• Deliverable: Measure memory buffer growth and time to checkpoint completion under both strategies.

Lab 3: Simulate TwoPhaseCommitSink Rollback on Simulated Network Partition

• Objective: Inject a simulated network timeout between Step 4 (Pre-Commit) and Step 6 (Formal Commit).
• Deliverable: Prove that uncommitted transactions are cleanly aborted and restarted from the previous checkpoint without duplicate side effects.

export interface UnalignedStateHandle extends StateHandle {
  inFlightChannelRecords: Record<number, StreamItem[]>;
}

export function snapshotUnaligned(
  operatorId: string,
  epoch: number,
  keyedState: Map<string, number>,
  channelQueues: Map<number, StreamItem[]>
): { emit: StreamItem[]; snapshot: UnalignedStateHandle } {
  const inFlightChannelRecords: Record<number, StreamItem[]> = {};
  for (const [chId, queue] of channelQueues.entries()) {
    // Capture all records sitting ahead of the barrier in lagging input channels
    const idx = queue.findIndex((item) => item.type === 'BARRIER' && item.epoch === epoch);
    inFlightChannelRecords[chId] = idx === -1 ? [...queue] : queue.slice(0, idx);
  }
  return {
    emit: [{ type: 'BARRIER', epoch }], // Forward barrier immediately without waiting!
    snapshot: {
      operatorId,
      epoch,
      stateDump: Object.fromEntries(keyedState),
      inFlightChannelRecords,
    },
  };
}

12. Comprehensive Summary & Decision Framework

State Backend Selection Decision Matrix

Production Deployment Checklist

1. Enable Incremental Checkpoints: When using EmbeddedRocksDBStateBackend, set state.backend.incremental: true to prevent full state uploads.
2. Configure ZooKeeper/Kubernetes HA: Run at least 2 JobManagers in active-standby configuration to guarantee zero control-plane single points of failure.
3. Set Minimum Pause Between Checkpoints: Always set minPauseBetweenCheckpoints to prevent continuous checkpoint retry storms.
4. Tune RocksDB Native Memory: Set state.backend.rocksdb.memory.managed: true to prevent out-of-memory container terminations caused by uncontrolled off-heap memory growth.

13. Authoritative References & Version Context

• K. Mani Chandy and Leslie Lamport (1985), “Distributed Snapshots: Determining Global States of Distributed Systems”, ACM Transactions on Computer Systems (TOCS) 3(1) — doi.org/10.1145/214451.214456
• Paris Carbone et al. (2015), “Apache Flink: Stream and Batch Processing in a Single Engine”, IEEE Data Engineering Bulletin 38(4) — data-artisans.com
• Apache Flink 1.18/1.19 Official Documentation — State Backends & Checkpointing — nightlies.apache.org/flink/flink-docs-master/docs/ops/state/state_backends/
• Version Context: Stream engine architecture reflects Apache Flink 1.18+ / 2.0 with RocksDB state backends and unaligned checkpointing.