How Apache Flink Works Internally: Stream Processing Architecture, Checkpointing, and Exactly-Once Semantics From Job Submission to Stateful Processing

Most engineers who reach for Apache Flink already understand the surface contract: you define a dataflow, attach sources and sinks, and Flink executes it continuously over an unbounded stream with fault tolerance. What they often do not know is the specific sequence of mechanisms that makes that contract hold: how a job graph becomes a set of deployed threads, how Flink takes a consistent snapshot of distributed mutable state without pausing execution, and exactly what “exactly-once” means when your sink is an external database that has no knowledge of Flink at all.

These details matter operationally. Engineers who treat Flink as a black box hit checkpoint timeout alerts, RocksDB memory pressure, and watermark stalls that they cannot diagnose because they lack the vocabulary for what is happening at the engine layer. This article gives you that vocabulary.

JobManager and TaskManager: Slot-Based Resource Scheduling

A running Flink cluster has two process types. The JobManager is the control plane: one process per job (in Application Mode) or one shared process across jobs (in Session Mode). It contains three logical components:

• Dispatcher: accepts job submissions, starts a JobMaster per job.
• ResourceManager: negotiates with the underlying cluster manager (YARN, Kubernetes, or standalone) to acquire TaskManager slots.
• JobMaster: owns the lifecycle of a single job, schedules its tasks, and drives checkpointing.

TaskManagers are the worker processes. Each TaskManager exposes a fixed number of task slots. A slot is a thread with a reserved fraction of the TaskManager’s managed memory. Slots are not isolated processes; multiple slots within the same TaskManager share its JVM heap and off-heap memory regions, though each slot runs in its own thread and owns its own state partition. The number of slots per TaskManager is configured with taskmanager.numberOfTaskSlots and defaults to 1. Parallelism is the primary scaling lever: setting an operator’s parallelism to N causes Flink to request N slots for that operator, distributed across available TaskManagers.

Flink can colocate operators from the same job within the same slot (slot sharing groups), which reduces the minimum slot count required. By default, all operators share a slot unless you explicitly call slotSharingGroup("name") to isolate a memory-heavy operator.

From API to ExecutionGraph: Operator Chaining

When you submit a job, the client serializes a StreamGraph built from your API calls. The JobMaster transforms it through two intermediate representations:

1. JobGraph: collapses StreamGraph nodes that can run in the same thread into chained operators. Two operators chain when they share the same parallelism, are connected by a forward partition strategy, and neither has disabled chaining. Chaining eliminates the serialization and network overhead of an intermediate record buffer — chained operators pass records as Java object references through a direct method call.

2. ExecutionGraph: the physical deployment plan. Each logical operator vertex is expanded by its parallelism into individual ExecutionVertex instances. Each ExecutionVertex becomes an Execution, which the JobMaster schedules to a concrete task slot.

Understanding chaining is critical for performance. A long chain of stateless operators (map, filter, flatMap) that runs in a single thread with zero network crossing is far cheaper than the same operators separated by shuffle edges. If you break a chain unnecessarily with .disableChaining() or a keyBy between two transformations, you pay serialization and cross-task buffer transfer costs for every record.

// Conceptual representation of what the Flink API compiles into
// (TypeScript pseudocode illustrating the graph structure)

interface OperatorNode {
  id: string;
  parallelism: number;
  chainable: boolean;
  partitionStrategy: "forward" | "hash" | "broadcast" | "rebalance";
}

interface ExecutionVertex {
  operatorId: string;
  subtaskIndex: number;
  assignedSlot: string | null;
}

function buildExecutionGraph(
  operators: OperatorNode[],
  jobParallelism: number
): ExecutionVertex[] {
  const vertices: ExecutionVertex[] = [];
  for (const op of operators) {
    const p = op.parallelism ?? jobParallelism;
    for (let i = 0; i < p; i++) {
      vertices.push({ operatorId: op.id, subtaskIndex: i, assignedSlot: null });
    }
  }
  return vertices;
}

Distributed Snapshots: The Chandy-Lamport Checkpoint Algorithm

Flink’s fault tolerance is based on asynchronous distributed snapshots. The algorithm is a variant of Chandy-Lamport: the JobMaster periodically injects special records called checkpoint barriers into every source subtask. These barriers flow through the dataflow graph alongside regular records, carrying a checkpoint ID.

When an operator subtask receives a barrier on one input channel, it begins recording state changes to a staging area. When it has received barriers on all input channels (aligned checkpoint mode), it atomically snapshots its state to durable storage and forwards the barrier downstream. The checkpoint completes globally when every sink operator acknowledges barrier receipt to the JobMaster.

Aligned checkpointing provides exactly-once semantics. Operators buffer records arriving after the first barrier on any channel until the remaining barriers arrive. Under backpressure this buffering can grow large, increasing checkpoint latency.

Unaligned checkpointing (Flink 1.11+) eliminates that stall. When an operator receives the first barrier, it immediately forwards the barrier downstream without waiting for the remaining channels. The in-flight records that arrived after the first barrier and before the others are included in the checkpoint alongside the operator state. Recovery replays those buffered records to reconstruct the exact pre-failure state. The tradeoff: checkpoint size grows proportionally to the volume of in-flight data, which can be significant under sustained backpressure. Configure it with execution.checkpointing.unaligned.enabled: true and pair it with execution.checkpointing.unaligned.forced: true when checkpoint intervals are consistently exceeded.

State is written asynchronously using copy-on-write: the operator continues processing new records while the snapshot of the previous state version is uploaded to the configured checkpoint storage (typically S3 or HDFS). The checkpoints.dir config points to the durable target.

// Conceptual barrier flow through a two-input operator
// Illustrates aligned checkpoint buffering behavior

interface CheckpointBarrier {
  checkpointId: number;
  timestamp: number;
}

type InputRecord = { data: unknown } | CheckpointBarrier;

class AlignedOperator {
  private buffered: Map<number, unknown[]> = new Map(); // channelId -> buffered records
  private barriersSeen: Set<number> = new Set(); // channelIds that delivered the barrier
  private pendingBarrier: CheckpointBarrier | null = null;
  private readonly numInputChannels: number;

  constructor(channels: number) {
    this.numInputChannels = channels;
  }

  receive(channelId: number, record: InputRecord): void {
    if ("checkpointId" in record) {
      this.barriersSeen.add(channelId);
      this.pendingBarrier = record;
      if (this.barriersSeen.size === this.numInputChannels) {
        this.snapshotState(record.checkpointId);
        // release buffered records and forward barrier downstream
        this.barriersSeen.clear();
        this.buffered.clear();
      }
    } else if (this.pendingBarrier !== null && !this.barriersSeen.has(channelId)) {
      // barrier seen on other channels but not this one yet — buffer
      const buf = this.buffered.get(channelId) ?? [];
      buf.push(record.data);
      this.buffered.set(channelId, buf);
    } else {
      this.process(record.data);
    }
  }

  private snapshotState(checkpointId: number): void {
    // asynchronously write managed state to checkpoint storage
    console.log(`Snapshotting state for checkpoint ${checkpointId}`);
  }

  private process(data: unknown): void {
    // operator business logic
  }
}

State Backends: HashMapStateBackend vs EmbeddedRocksDBStateBackend

Flink’s stateful operators — keyed aggregations, windows, async I/O — store state in one of two backends:

HashMapStateBackend (formerly MemoryStateBackend / FsStateBackend in older Flink versions) stores all state as deserialized Java objects in the JVM heap. Reads and writes are direct object access with no serialization overhead. Checkpointing copies these objects to the configured checkpoint directory. It is fast for small-to-medium state sizes but JVM GC pressure scales with state size, and a single large key group can cause GC pauses that stall processing.

EmbeddedRocksDBStateBackend stores state as serialized byte arrays in a local RocksDB instance on the TaskManager’s disk. Reads and writes require serialization and deserialization on every access. The benefit is that the working set that fits in the RocksDB block cache stays fast, while cold state spills to disk without touching the JVM heap. This makes it the correct choice for large state, long-running windows, and jobs where state per key can grow unboundedly.

Incremental checkpointing is only available with EmbeddedRocksDBStateBackend. RocksDB’s LSM structure produces immutable SSTable files; Flink tracks which SSTables are new since the last checkpoint and uploads only those. For jobs with large state, this reduces checkpoint upload time from minutes to seconds after the first full checkpoint.

RocksDB memory budgeting is the most common source of TaskManager OOM errors in production. RocksDB does not use JVM heap; it allocates off-heap memory for its block cache and write buffer manager. If you do not set state.backend.rocksdb.memory.managed: true, each RocksDB instance allocates independently and multiple slots on the same TaskManager will collectively overrun the container’s memory limit. With managed memory enabled, Flink caps the aggregate allocation to the taskmanager.memory.managed.fraction portion of the configured TaskManager memory. A typical starting configuration:

state.backend: rocksdb
state.backend.rocksdb.memory.managed: true
state.backend.incremental: true
taskmanager.memory.managed.fraction: 0.4
execution.checkpointing.interval: 60000
execution.checkpointing.mode: EXACTLY_ONCE

Watermarks and Event-Time Processing

Flink processes records in two time domains. Processing time uses the wall clock of the machine executing the operator, which is simple but not reproducible. Event time uses a timestamp embedded in each record, which enables correct handling of out-of-order events but requires knowing when to declare a window closed.

The mechanism for declaring event-time progress is the watermark. A watermark with value t asserts that no future event with timestamp less than t will arrive. Flink propagates watermarks through the dataflow graph alongside data records. When a watermark reaches a window operator, any window whose end time is less than or equal to the watermark is triggered for evaluation and emission.

Each parallel source subtask maintains its own watermark. For a keyed window operator with multiple upstream subtasks, the effective watermark is the minimum across all input channels. This is why a single slow or stalled source subtask can prevent the entire pipeline from advancing: it holds the minimum watermark down, causing windows to accumulate indefinitely.

You assign watermarks at the source:

// TypeScript pseudocode illustrating the watermark assignment pattern
// In a real Flink DataStream API (Java/Scala/Python), you would use
// WatermarkStrategy.forBoundedOutOfOrderness or a custom WatermarkGenerator

interface WatermarkStrategy<T> {
  // Maximum expected out-of-orderness in milliseconds
  maxOutOfOrderness: number;
  // Extract event timestamp from the record
  extractTimestamp: (record: T) => number;
}

function boundedOutOfOrdernessWatermark<T>(
  strategy: WatermarkStrategy<T>
): (record: T, previousWatermark: number) => number {
  return (record: T, previousWatermark: number): number => {
    const eventTime = strategy.extractTimestamp(record);
    const candidateWatermark = eventTime - strategy.maxOutOfOrderness;
    return Math.max(candidateWatermark, previousWatermark);
  };
}

Late events arriving after the watermark has passed their window can be routed to a side output rather than dropped silently. Setting an allowedLateness on the window operator extends the window’s lifetime by that duration, accepting late events within the grace period. After that, events route to the configured late data side output.

Backpressure and Credit-Based Flow Control

Flink’s network stack uses a credit-based flow control mechanism between TaskManagers. Each receiver maintains a set of exclusive and floating buffers. When a sender wants to transmit data to a downstream task, it first checks available credits from the receiver. If no credits are available, the sender blocks. The blocked sender causes its upstream operators to fill their output buffers. This backpressure propagates upstream through the task graph until it reaches the source, where the consumer of an external system (Kafka, Kinesis) simply stops polling.

This design means Flink has natural backpressure without explicit rate limiting code, and without unbounded buffer growth. The Flink web UI surfaces backpressure metrics per operator subtask. An operator showing high backpressure is a consumer that cannot keep up. An operator showing high output buffer utilization is either a fast producer feeding into a slow consumer, or a bottleneck itself. Both indicators are available at /jobs/<id>/vertices/<vertex-id>/backpressure in the REST API.

Network buffer sizing directly affects throughput. Each network buffer is 32KB by default (taskmanager.memory.segment-size). Increasing it reduces system call frequency but increases per-buffer memory cost. The total network memory is bounded by taskmanager.memory.network.fraction.

Exactly-Once Semantics and Two-Phase Commit Sinks

Flink’s checkpoint mechanism gives you exactly-once state consistency within the Flink dataflow. Extending that guarantee to external sinks requires that the sink participate in the checkpoint protocol via a two-phase commit (2PC) pattern.

The TwoPhaseCommitSinkFunction (and the newer Committer-based sink API) works as follows:

1. Pre-commit (beginTransaction): for each checkpoint interval, the sink opens a new transaction with the external system and writes records into it without committing.
2. On barrier receipt (preCommit): the sink flushes its write buffer and makes the transaction durable but not yet visible. For Kafka, this means calling producer.flush() and entering the commitTransaction call sequence. For JDBC sinks, it means completing the write but holding the transaction open.
3. On checkpoint completion (commit): the JobMaster notifies all operators that the checkpoint succeeded. The sink then commits the transaction, making the records visible.
4. On recovery: if the process crashes after preCommit but before commit, the transaction is either committed (if checkpoint completed before crash) or aborted (if checkpoint was incomplete). This is recoverable only if the external system supports transaction abort.

The practical constraint: exactly-once to Kafka sinks requires Kafka’s transactional producer (idempotent + transactions). Exactly-once to a JDBC sink requires the database to support multi-statement transactions with rollback. Exactly-once to S3 requires the multipart upload abort capability. Systems without transactional semantics can only achieve at-least-once.

// Pseudocode illustrating the 2PC sink lifecycle events
// Maps to Flink's TwoPhaseCommitSinkFunction contract

interface Transaction {
  id: string;
  writes: unknown[];
}

class ExactlyOnceSink {
  private currentTx: Transaction | null = null;

  // Called at the start of each checkpoint interval
  beginTransaction(): Transaction {
    this.currentTx = { id: crypto.randomUUID(), writes: [] };
    return this.currentTx;
  }

  // Called with each incoming record
  write(record: unknown): void {
    if (!this.currentTx) throw new Error("No active transaction");
    this.currentTx.writes.push(record);
  }

  // Called when checkpoint barrier passes through this operator
  preCommit(tx: Transaction): void {
    // flush writes, make durable but not visible
    console.log(`Pre-committing transaction ${tx.id} with ${tx.writes.length} records`);
  }

  // Called after JobMaster confirms full checkpoint completion
  commit(tx: Transaction): void {
    console.log(`Committing transaction ${tx.id} — records now visible`);
  }

  // Called if checkpoint fails or job recovers from a previous checkpoint
  abort(tx: Transaction): void {
    console.log(`Aborting transaction ${tx.id}`);
  }
}

Production Considerations

Checkpoint interval tuning: shorter intervals reduce recovery time (you replay less data from the source) but increase cluster overhead because state upload and barrier injection consume CPU and I/O. Start at 60 seconds and tighten based on your recovery time objective. If checkpoints consistently exceed the interval, enable unaligned checkpoints before reducing the interval further.

State size management: unbounded state is the most common production failure mode. Every keyed state that grows without a TTL will eventually fill the TaskManager’s managed memory and trigger spill-to-disk or OOM. Use StateTtlConfig to expire state that has not been accessed within a defined window. For session windows with long gaps, set both the session gap and the TTL with some headroom.

RocksDB memory budgeting: set state.backend.rocksdb.memory.managed: true and verify that your TaskManager’s total memory (taskmanager.memory.process.size) accounts for JVM heap + managed (RocksDB) + network buffers + metaspace. A TaskManager with 8GB configured and managed.fraction: 0.4 gives RocksDB 3.2GB of block cache and write buffer budget across all slots. If you have 4 slots each with a large state operator, that 3.2GB is shared, not per-slot.

Network buffer sizing: increase taskmanager.memory.network.fraction if you observe throughput saturation with small record sizes. Flink fills buffers before flushing them across the network; small record sizes cause many small flushes unless you increase the buffer flush timeout (task.network.max.sendDelay) to allow buffers to fill.

Job restart strategies: configure restart-strategy: exponential-delay with a reasonable cap rather than fixed-delay with short intervals. Rapid restarts after failures can cause a thundering herd against external systems. An exponential backoff with jitter prevents repeated failure loops when the external dependency is temporarily unavailable.

Watermark idleness: sources that go idle (no records arriving) stop advancing their watermark. A single idle Kafka partition in a source with multiple partitions holds the minimum watermark down for the entire pipeline. Use WatermarkStrategy.withIdleness(Duration.ofMinutes(1)) to let Flink exclude idle sources from the minimum calculation after a configurable idle period.

Tradeoffs Comparison

Dimension	Apache Flink	Spark Structured Streaming	Kafka Streams	Apache Beam	Amazon Kinesis Data Analytics
Processing model	True streaming (record-at-a-time)	Micro-batch (configurable trigger interval)	True streaming	Portability layer (runner-dependent)	Managed Flink
State management	Rich keyed state with multiple backends	Stateful only via stateful mapGroupsWithState; limited	RocksDB per partition, local-only	Runner-dependent	Flink semantics, managed RocksDB
Exactly-once	Yes, with 2PC sinks	Yes, via WAL + idempotent sinks	Yes, with Kafka transactions	Runner-dependent	Yes, inherits Flink
Latency profile	Sub-second to millisecond	Seconds (micro-batch interval)	Sub-second	Runner-dependent	Sub-second
Checkpointing	Chandy-Lamport barriers, async	Structured Streaming WAL + offset commits	Kafka offset commits only	Runner-dependent	Managed Flink checkpointing
Event-time support	First-class: watermarks, late data, side outputs	Available via event-time watermarks	Basic: stream-time joins only	First-class	Full Flink event-time
Operational complexity	High: cluster sizing, state tuning, RocksDB memory	Medium: Spark cluster management	Low: embedded library	Medium: depends on runner	Low: fully managed
Scaling model	Manually change parallelism + rescale job	Dynamic allocation on Spark clusters	Kafka partition count limits parallelism	Runner-dependent	Manual or autoscaling
SQL support	Flink SQL, Table API, full window and join support	Spark SQL with streaming extensions	KSQL (via Confluent) or manual	SQL via runners	Flink SQL
Best fit	Complex stateful streaming, event-time semantics, large state	Teams already on Spark, batch-streaming unification	Simple Kafka-to-Kafka enrichment without a separate cluster	Multi-runner portability requirements	Managed Flink without ops overhead

Closing Insight

The checkpoint barrier is the structural decision that makes Flink coherent as a system. The slot model, the operator chaining optimization, the two-phase commit sink protocol, the state backend memory model: all of these are consequences of needing to take a consistent distributed snapshot of a running dataflow without a global pause. Understanding that one algorithm clarifies why the rest of the system is designed the way it is, and why debugging a stuck checkpoint or a watermark that never advances requires looking at the same set of mechanisms from different angles.