Disaggregated State Management in Apache Flink 2.0

In one breath — the whole paper, compressed

Flink 1.x kept working state in an embedded RocksDB on the task manager's local disk, which in containerized cloud deployments forces a job to buy CPU cores just to buy disk space: at a threshold of 20 GB of state per core, 35% of the jobs in Alibaba's logistics business are disk-bound rather than CPU-bound. Flink 2.0 inverts the hierarchy so that a remote distributed file system (HDFS, OSS, S3) becomes the primary store and local memory and disk become a cache, via ForSt, an LSM-tree state store whose files are streamed continuously to the DFS through a Unified File System layer that emulates hard links using logical-to-physical mappings and reference counts. Because the working directory and the checkpoint directory now live in the same file system, checkpointing collapses into creating links and registering them with the Job Manager, and recovery or rescaling needs no data transfer at all. To pay for the resulting jump in state read latency (68 microseconds on local NVMe versus 23 milliseconds on OSS), the runtime splits each record into a non-state transformation, an asynchronous state access, and a post-state callback, and an Asynchronous Execution Controller permits only one in-flight computation per key so that per-key FIFO order, exactly-once checkpoints, and watermark completeness all survive out-of-order execution. On a 290 GB replica of the production logistics job, every checkpoint finishes within 3 seconds against 19.7% exceeding 30 seconds in Flink 1.20, scale-out is 49 times faster, and the job runs on 8 compute units instead of 16, halving its cost.

Before this paper — the world it landed in

Flink was built in the Map-Reduce era, when the same machines held both computation and state, and its state management still reflected that: keyed state is pre-partitioned into key-groups, each stream task owns its partitions under single-writer semantics, and an embedded backend - RocksDB on the local disk, or a Java-heap map - serves every read and write synchronously on the task's main thread. Checkpoints are an asynchronous two-phase commit driven by markers that the Job Manager injects into the sources: once markers align, the backend takes a synchronous local copy and then uploads it to an external DFS, while the Job Manager tracks the resulting file references for recovery. By the mid-2020s the deployment picture had changed completely - containerized jobs on Kubernetes, cloud compute units sold as fixed bundles of CPU, memory and disk, cheap object storage, and far fatter intra-datacenter networks. The workloads grew into that world too: Alibaba's Flink infrastructure absorbed an inbound flow of over 4.4 billion TPS during the 2024 Double 11 shopping event, and a single logistics job that keeps 60 days of orders and shipping updates carries hundreds of gigabytes to terabytes of state. The local disk under each container had quietly become the thing that decided how many machines you rented and how long a restart took.

The problem — what was actually breaking

Cloud compute units bundle CPU with a fixed local disk - one core with 20 GB in Alibaba Cloud's Realtime Computation Service, one core with 50 GB for an AWS Kinesis Processing Unit - so a job whose state outgrows the disk must rent extra cores it does not need, and at a threshold of 20 GB of state per core, 35% of the jobs in Alibaba's logistics business are disk-bound.
Checkpointing in Flink 1.x copies local state tables to temporary local storage in a synchronous phase and then uploads them to the DFS in an asynchronous phase, so its duration scales with state size: with an average incremental checkpoint of 1.89 GB, over 19.7% of checkpoints exceeded 30 seconds and more than 1.5% exceeded 50 seconds.
Recovery and rescaling are pause-and-restart operations in which every worker must download its state partition from the checkpoint and rebuild a local RocksDB instance, which takes over three minutes for a 290 GB job and can take hours for cluster-scale recovery or migration during a Double 11 peak.
Background state backend work interferes with the foreground query: compaction and checkpoint uploads produce periodic spikes in CPU, disk and network consumption that persist in Flink 1.20 even when the backend operations run asynchronously, forcing operators to reserve resources in advance for the spikes.
Simply moving state to remote storage makes reads two orders of magnitude slower - 68 microseconds on local NVMe and 199 microseconds on an ESSD PL1 disk, against 1.5 milliseconds on HDFS and 23 milliseconds on OSS - and Flink 1.x executes each record as a single atomic blocking operation on the main thread, so remote reads would sit directly on the critical path and leave I/O bandwidth idle.
Any redesign must not break the three guarantees users depend on - per-key FIFO processing order, exactly-once state commits between consecutive checkpoint markers, and low-watermark completeness - nor force existing applications through a disruptive migration.

Core ideas — the contributions, and why they work

DFS as primary state storage

Flink 2.0 inverts Flink 1.x's storage hierarchy: the remote distributed file system holds the authoritative working state, local memory and disk become an optional cache, and updates are streamed to the DFS continuously instead of being uploaded at checkpoint time. This takes the local disk out of the sizing equation entirely, so a terabyte-state job can run on as many cores as its computation actually needs. More subtly, it makes active state and checkpointed state neighbours in the same file system, which is what turns checkpointing and state migration from bulk data movement into metadata bookkeeping. The cost table is the direct consequence: the same logistics job needs 16 compute units on Flink 1.20 purely to stay under the disk limit, and 8 on Flink 2.0.

Asynchronous record execution

Because remote reads are roughly 20 to 300 times slower than local ones, Flink 2.0 refuses to let them block the main task thread. Each record's lifecycle is split into three stages - non-state transformation, state access, and post-state callback - with only the state access handed to a separate thread pool while the other two stay on the main thread as before. The main thread keeps pulling records while earlier ones wait on I/O, so CPU-bound transformations overlap with slow remote I/O and many state requests are in flight at once, which is what actually saturates remote bandwidth. Records that need no state access skip the machinery entirely and behave exactly as in Flink 1.x, which is why the whole model can be bypassed by configuration.

Key accounting for per-key order

Out-of-order execution is only safe because the Asynchronous Execution Controller permits exactly one in-flight computation per key within a task at any moment. A Key Accounting Unit records the key of every in-progress computation, including non-state transformations, state accesses and callbacks; a newly arriving record whose key is already busy is parked in a blocking buffer and released in FIFO order once its predecessor finishes. Combined with Flink's existing FIFO channels, which already deliver same-key records to a task in origin order, this reproduces exactly the per-key sequential semantics of Flink 1.x while allowing unbounded concurrency across independent keys. The single-writer atomicity per key that complex event processing and similar libraries build on therefore survives untouched.

Event-time epochs for watermarks

A watermark promises completeness: every record with a timestamp below it has already been processed. Async execution breaks that promise, since computations older than a watermark may still be pending when it arrives, so Flink 2.0 introduces epochs - the periods between two consecutive watermarks - as the unit of asynchronous progress tracking. An epoch is OPEN while it accepts records, CLOSED when the next watermark seals it, and FINISHED once all of its records have completed; an Epoch Manager keeps epochs in a queue and releases a watermark downstream only when its epoch is FINISHED and has reached the head. Because a later epoch can finish before an earlier one, the head-of-queue rule is what preserves both monotonicity and completeness, so event-time timers and windows fire exactly as they did under synchronous execution.

Async draining for exactly-once

Exactly-once in Flink means the state committed by a checkpoint reflects all records before the checkpoint marker and none after it, which synchronous execution gave for free. Flink 2.0 restores it with a draining step: when the aligned marker arrives, the operator blocks all subsequent input and asks the AEC to finish every pending computation, including records still sitting in the blocking buffer and any state accesses or callbacks they spawn. Only then is the local checkpoint taken and the marker emitted downstream, in the same atomic step that unblocks the input. This leaves the existing two-phase-commit protocol and its Job Manager coordination completely unchanged, which is why async execution can be turned on or off by reconfiguration without breaking checkpoint compatibility with Flink 1.x.

Unified File System with hard links

Distributed file systems disagree about almost everything a state store cares about: HDFS makes writes immediately visible while S3 is eventually consistent, and essentially none of them offer POSIX hard links, so sharing a file usually means copying its data. ForSt interposes a Unified File System layer that presents one logical file view over HDFS, OSS and S3, maintains the mapping and reference counts from logical files to physical locations, and normalizes object visibility. With that in place, a link or a move becomes a metadata entry rather than a data copy, which is precisely the primitive that makes checkpointing cheap and recovery transfer-free. It also lets the Job Manager keep its existing checkpoint lifecycle logic and merely delegate deletion, so the new layer is minimally intrusive to the rest of the runtime.

Compaction as a remote service

In Flink 1.x, LSM compaction runs on the same nodes as the operators and is amplified by periodic checkpointing, producing the CPU and I/O spikes that force over-provisioning. Because the working state now lives on shared storage, compactor workers elsewhere can read and rewrite those files without disturbing normal processing, so ForSt offers compaction-as-a-service with stateless compactors triggered by Flink tasks. Compute and compaction capacity then scale independently, compactors can be placed inside the storage cluster's network since compaction is I/O-bound, and spikes from many jobs are staggered across a shared pool instead of landing on one job's task managers at once. The paper is explicit that remote compaction is still an experimental feature in a Flink branch rather than a shipped default.

How it works — the mechanism, concretely

Async API: chaining state calls with THEN

Operators opt in by giving the runtime async hints. In the logistics streaming join, the non-state transformation - casting the record - is unchanged from Flink 1.x; the shipping state table is updated with asyncUpdate, and the order state table is read with asyncGetEntries followed by a THEN block that receives the fetched entries, runs applyJoin and emits the result. Independent state accesses, such as the two sides of the join, are issued concurrently, whereas THEN chaining is how a dependency is expressed and how sequential order between two state operations on the same record is enforced. Callbacks run on the main thread, and post-state callbacks take precedence over the non-state transformations of newly arriving records so that pending records complete before new work starts.

AEC scheduling and the blocking buffer

The main task thread pulls records from the FIFO input channel and submits state requests to the AEC. If record O1 with key 26 is in flight, the Key Accounting Unit holds key 26; when O2, also key 26, arrives, the accounting unit detects the conflict and parks O2 in a blocking buffer, while O3 with key 18 passes straight through and acquires a state thread. When O1's request completes, its callback plus the fetched state is enqueued on a callback channel that the main thread drains ahead of the input channel, and only after O1 finishes does the AEC resume the pending key-26 requests in FIFO order. The number of in-flight records and the buffer size are configurable, defaulting to 6000 records, which costs several MB per operator; when the buffer saturates the AEC blocks new records, which surfaces as backpressure.

Draining at an aligned checkpoint marker

The network layer aligns checkpoint markers across inputs exactly as before. On consuming the aligned marker for a checkpoint, the task blocks every record that follows it in the input order, which establishes that nothing after the marker is processed early. It then waits on the AEC until no computations remain pending for records preceding the marker: in-flight records, records held in the blocking buffer, and every derivative state access and callback they generate. When the AEC reports quiescence, the local checkpoint is taken, the marker is emitted downstream, and the regular input is unblocked in the same atomic step. Draining lengthens checkpoints, but the bounded in-flight count keeps that to seconds, and the unaligned Chandy-Lamport checkpoints inherited from Flink 1.x still work without draining, at the cost of checkpointing in-flight messages as part of the state.

Epoch queue mechanics

The Epoch Manager holds a queue of epochs; new epochs and their watermarks are enqueued at the tail and emitted from the head. Arriving records are assigned to the current OPEN epoch; the arrival of a combined watermark seals that epoch to CLOSED in the same atomic operation and opens a fresh one, and the sealed epoch becomes FINISHED once every record it holds has completed. Dequeuing requires FINISHED status, and a watermark may only be emitted once it reaches the head, so in the paper's example the finished epoch [30, 40] and its watermark are held back because the preceding epoch [20, 30] still has pending records. The combined input watermark itself is still derived by Flink's existing monotonic max-min reduction over the per-input watermarks, so only completeness, not monotonicity, needed new machinery.

Checkpointing as link creation and refcount decrement

Because ForSt streams its files to the working directory on the DFS, most files a checkpoint needs are already durably replicated there when the marker arrives. The backend creates hard-linked logical copies of the files to be checkpointed, with the UFS updating logical-to-physical mappings and incrementing reference counts, and those logical copies are registered with the Job Manager just as physical files were in Flink 1.x. When a checkpoint becomes obsolete, the Job Manager does not delete files directly: it issues a deletion against the hard-linked reference, the UFS decrements the reference count, and the physical file is removed only once the count reaches zero. This is the single refinement to the Flink 1.x protocol, and it is what preserves the Job Manager's checkpoint lifecycle management while removing the data movement underneath it.

Recovery and rescaling without data transfer

Reconfiguration in Flink is still pause-and-restart: the job stops, workers load state from the most recent checkpoint, and processing resumes once all partitions are loaded. What changes is the loading step - new ForSt instances attach to linked copies of checkpoint files already resident on the DFS, so no bytes move from remote storage to local disks. Flink 1.20 must instead download and rebuild roughly 290 GB, and double that for scale-out because each of the 32 new workers fetches data stored by one of the 16 original workers, which is why the scale-out gap is the widest of the three scenarios. The residual cost in Flink 2.0 is metadata: rebuilding a ForSt instance still requires loading LSM metadata, which is why OSS trails HDFS by 10 to 20 seconds and why the authors plan to merge metadata reads at startup.

Tiered cache and remote compaction dispatch

ForSt caches in two tiers on the compute node: a conventional block-based LRU cache in memory, and a file-based secondary cache on local disk that replicates whole SSTable files from remote storage. The secondary cache runs a History-Based Policy - LRU for eviction across the currently cached files, but frequency-driven loading, where files whose access frequency over the preceding minute exceeds a threshold are periodically pulled back in, which is what mitigates cache thrashing; the policy is the default in Flink 2.0 and in Alibaba's service, and is pluggable. Remote compaction reuses Flink 1.x's compaction triggering but offloads the work: the ForSt backend sends a compaction request carrying metadata to the service, a scheduling node assigns it to a compactor using round-robin, and the backend updates its LSM metadata once notified of completion. Compactors are stateless and can be deployed inside the DFS cluster's intranet, since compaction is I/O-intensive and indifferent to where the compute nodes sit.

What the paper showed — measurements and proofs

On a 290 GB replica of the production logistics job, running on six ecs.g7.8xlarge nodes against a three-node HDFS cluster, Flink 1.20 needs at least 16 compute units to keep state per CU under the 20 GB disk limit, while Flink 2.0 is memory-bound and sustains the same daily traffic on 8, cutting the monthly bill from $688 to $344, with no extra HDFS cost because both versions already store checkpoints there.
Across five hours and 300 checkpoints at a one-minute interval with incremental checkpoints enabled, every Flink 2.0 checkpoint completed within 3 seconds regardless of size, whereas Flink 1.20 - averaging 1.89 GB per incremental checkpoint - exceeded 30 seconds for 19.7% of checkpoints and 50 seconds for over 1.5%; repeating the run on OSS instead of HDFS kept Flink 2.0 within 4 seconds while Flink 1.20 fluctuated with OSS long-tail transfer latency, which the abstract summarizes as up to a 94% reduction in checkpoint duration.
For three production reconfiguration scenarios on HDFS, Flink 2.0 restarts within tens of seconds and is 16 times faster for failure recovery, 12 times faster for scale-in from 32 to 16, and 49 times faster for scale-out from 16 to 32; with OSS the conclusion holds but each operation takes roughly 10 to 20 seconds longer, attributed to slower metadata lookups and small random I/O on object storage.
Across the Nexmark queries (all but Q6, unsupported by Flink SQL), ForSt on local disk with async disabled matches Flink 1.20's throughput almost exactly, disaggregated state on HDFS with no cache costs 48% throughput on average for queries with heavy I/O, and adding only a 1 GB local disk cache puts Flink 2.0 4% above the local-state configurations on average.
On the five I/O-intensive Nexmark queries, disaggregated state with synchronous access degrades severely, asynchronous execution alone recovers roughly 2 times throughput, and the 1 GB cache adds up to 3.7 times - even though 1 GB cannot hold the state of any of them (q7 2.25 GB, q9 4.48 GB, q18 1.05 GB, q19 1.52 GB, q20 2.95 GB).
The async model costs a 30% average increase in CPU utilization for stateful operators, attributed to context switching for asynchronous access (30%), AEC intra-task scheduling (20%), batching I/O classification and parallel execution for remote access (20%), and extra garbage collection for Future objects (30%); at the job level, however, Flink 1.20 consumed 9% more CPU than Flink 2.0 on the logistics workload because it needed 16 task managers instead of 8.

Limits and trade-offs — conceded and discovered

The paper concedes the CPU tax: stateful operators use about 30% more CPU under async execution, and for workloads with minimal or no state I/O the overhead of async dispatch and DFS access can outweigh the benefit, which is why Flink 2.0 keeps both sync and async modes and leaves the choice to the user.
The paper concedes that disaggregation is not free at steady state - heavy-I/O Nexmark queries lose 48% throughput without a cache - so the headline result depends on a local cache and on ample intra-datacenter bandwidth, for which the authors cite OSS's default 5 Gbps; queries whose state fits in the block cache gain nothing from async threading or caching because dispatch overhead exceeds the memory-access gain.
The paper concedes residual costs inside its own mechanisms: draining prolongs checkpoints and is only kept tolerable by the default 6000 in-flight record limit, reconfiguration on object storage is 10 to 20 seconds slower than on HDFS because ForSt metadata must still be read at startup, and remote compaction is an experimental feature living in a branch rather than a shipped default.
Not conceded but visible in the setup: the evaluation is a two-system comparison against Flink 1.20 on a single 290 GB replica of one production job plus Nexmark, with no measurement against the disaggregated streaming systems the related-work section names, including RisingWave, which the authors themselves call the closest architecture to Flink 2.0.
Also outside the paper's discussion: the benefits reach an operator only after it has been rewritten against the async state APIs - the acknowledgments credit a separate effort to rewrite and adapt the asynchronous SQL operators - so custom operators and third-party libraries stay on the synchronous path until they are ported.

What it became — the systems that inherited it

Flink 2.0, released in 2025 as the project's first major version in a decade, ships this architecture as its headline feature; the artifact is the apache/flink release-2.0 branch, ForSt is open source at ververica/ForSt, and the design had already run for over two years inside Alibaba Cloud's Realtime Compute service, through Double 11 peaks, before being contributed upstream. The most transferable piece is not the storage layout but the asynchronous execution model: it shows how to keep per-key FIFO order, exactly-once checkpoints and watermark completeness intact while records execute out of order, using key accounting, draining and epoch queues - a recipe any streaming runtime that moves state off the node has to reproduce. It also completes the streaming side of a decade-long disaggregation trend that had already reshaped analytical and transactional databases in Snowflake, Amazon Aurora, Socrates and PolarFS, and converges on the same conclusions RisingWave reached independently, which the authors name as the closest architecture: remote storage as the source of truth plus remote compaction. Against earlier external-state stream processors it is a direct answer - MillWheel put state in BigTable and Spanner but read it synchronously on the record's critical path, and Meta's ZippyDB-backed pipelines avoided full state reloads only for monoid-style append-only updates - whereas Flink 2.0 keeps a general keyed-state API and an LSM store while still hiding the remote latency. Finally, the UFS trick of turning checkpointing into reference counting over files that already live in the checkpoint's file system generalizes beyond Flink: it is the same insight as zero-copy cloning in modern lakehouse and cloud-database storage layers, applied to streaming snapshots.

In the paper’s words — verbatim

“Flink 2.0 relies on a remote distributed file system (DFS) for primary state storage and uses local disks as a secondary cache, with state updates streamed continuously and directly to the DFS.”

Abstract

“The AEC enforces a single active in-flight computation per key, within each task, at a time.”

§4.3

“This transforms checkpointing from a data-intensive operation to a lightweight reference creation.”

§5.2

Vocabulary — as this paper uses it

Disaggregated state management: Flink 2.0's architecture in which a remote DFS is the primary store for active working state while local memory and disk act only as a cache, as opposed to Flink 1.x's embedded backend where the local disk holds the authoritative state.
ForSt (For Streaming): Flink 2.0's disaggregated state store: an LSM-tree engine whose files are written to and read from a distributed file system through the Unified File System layer, with tiered local caching, checkpoint file sharing, and optional remote compaction.
Unified File System (UFS): The layer inside ForSt that presents one logical file view over HDFS, OSS and S3, hides their differences in visibility semantics, and maintains logical-to-physical mappings with reference counts so links and moves need no data copying.
Asynchronous Execution Controller (AEC): The per-task component that schedules asynchronous state access, permitting only one in-flight computation per key at a time and buffering conflicting records, so out-of-order execution still yields Flink's per-key FIFO order.
Key Accounting Unit: The AEC's bookkeeping structure that tracks the keys of all in-progress computations - non-state executions, state accesses and callbacks - and diverts any record whose key is already in flight into the blocking buffer.
Event-time epoch: The period between two consecutive watermarks, used as the unit of asynchronous progress tracking; it is OPEN while accepting records, CLOSED once the next watermark seals it, and FINISHED once all of its records have been fully processed.
Async draining: The step that restores exactly-once under async execution: on an aligned checkpoint marker the task blocks later input and waits for all preceding state accesses and callbacks to complete before taking the local checkpoint and forwarding the marker.
Compute Unit (CU): The bundled billing unit of containerized Flink services; in Alibaba Cloud one CU is one CPU core with 4 GB of memory and 20 GB of disk at roughly $43 per month, so a state-heavy job under Flink 1.x buys cores it does not need.
Remote compaction: Compaction-as-a-service in ForSt, where Flink tasks trigger compaction but stateless compactor workers read and rewrite the LSM files directly on the shared DFS, removing compaction CPU and I/O from the task managers.

On the timeline — where this sits in the story

Year2025

MilestoneApache Flink 2.0 disaggregated state management

Problem solvedStateful stream processors struggled with local-state scaling, snapshots, rescaling, and resource spikes.

How it solved itFlink 2.0 decoupled compute from state storage using remote storage plus local caching.

Why it matteredImproved scalability and operational flexibility for stateful real-time processing.

DDIACh. 12: Stream Processing

View on the timeline