Apache Fluss is a streaming storage layer that closes the semantic gap between a durable log broker (like Apache Kafka) and a mutable streaming table. It provides built-in KV serving (= point lookups on current key state), changelog generation, server-side column pruning, and lakehouse tiering — all as first-class storage-layer features rather than application-level workarounds.
Fluss entered the Apache Incubator in late 2024 and is not yet a Top-Level Project as of June 2026. Latest release: 0.9.1 (May 2026). Production deployment exists at Alibaba/Taobao.
Prerequisites
- Apache Kafka — log-based messaging (partitions, ISR, replication)
- Introduction to Flink — streaming SQL, checkpointing, state backends
- Familiarity with LSM trees (= Log-Structured Merge trees — the write-optimised data structure behind RocksDB, LevelDB, and Cassandra)
Why Fluss Exists — The Gap It Fills
Fluss sits between sub-second producers and minute-scale lakehouse queries, with a Tiering Service bridging the two layers.
When you run Flink against Kafka, four structural gaps emerge:
| Gap | Mechanism |
|---|---|
| No native upsert | Kafka appends every message. A downstream Flink job wanting a keyed table must keep all history in Flink state (expensive) or deduplicate at read time. Neither is built into the broker. |
| No queryable surface | Kafka topics are opaque byte streams. Point lookups require a separate materialisation job, adding a latency tier. |
| Backfill instability | Reading large historical ranges competes with live I/O on the same broker segments, causing instability — even with Kafka’s newer Tiered Storage. |
| Column-blind network | Consumers fetch entire batches. If your job needs 3 of 40 columns, it still pays the full network bill. |
The deeper problem: with Kafka + Flink, mutable state lives entirely inside Flink’s state backend (RocksDB or Heap). Flink checkpoints carry the full KV state, downstream changelog observation requires explicit materialisation to a sink, and there is no single durable place where “the current value of key X” is queryable outside a running Flink job.
Fluss moves the mutable KV state and its changelog into the storage tier itself, making Flink jobs stateless or near-stateless with respect to table state.
Architecture
Three TabletServers host bucket replicas with dual stores (LogStore + KvStore). The CoordinatorServer manages metadata and assignments via ZooKeeper. Clients route directly to the leader TabletServer for reads/writes.
CoordinatorServer
The cluster’s single logical control plane (can be made HA = High Availability via standby instances). Responsibilities:
- Metadata management — owns schemas for all databases, tables, and partitions. Persists to ZooKeeper (
/metadata/databases,/metadata/tables). - Tablet allocation — decides which TabletServer hosts which bucket-replica, writes assignments to
/buckets/assignments. - Leader election — selects the leader for each bucket replica from the current ISR (= In-Sync Replicas, the subset of replicas confirmed to be up-to-date with the leader) set; tracks epoch counters in
/buckets/leader_and_isr. - Rebalancing — reacts to TabletServer join/leave events, initiates reassignment.
The CoordinatorServer is not in the data path — it is purely control-plane. Clients fetch routing metadata from it, then communicate directly with TabletServers for reads and writes.
TabletServer
The data-plane worker. Each TabletServer hosts a set of tablets (one tablet = one bucket-replica of one table). A TabletServer exposes two internal stores per tablet:
| Sub-store | What it is | Used by |
|---|---|---|
| LogStore | Append-only, segmented log (conceptually similar to a Kafka partition log) | LogTable + changelog side of PrimaryKeyTable |
| KvStore | Embedded RocksDB instance for point lookups | PrimaryKeyTable only |
ZooKeeper (current) → KvStore + Raft (planned)
ZooKeeper currently stores:
/metadata/*— database/table schemas/buckets/assignments— tablet placement/buckets/leader_and_isr— ISR set and leader epoch per bucket/servers— live server registrations (heartbeat ephemeral nodes)/kv_snapshots— pointers to remote KV snapshot files/remote_log_manifests— pointers to remote log segment metadata
The official roadmap plans to replace ZooKeeper with an internal KvStore + Raft consensus (similar to Kafka’s KRaft migration).
Partitioning Model
A Fluss table is divided into buckets (analogous to Kafka partitions). The number of buckets is set at table creation ('bucket.num' = 'N'). For partitioned tables (e.g., date-partitioned), each partition independently has N buckets. Each bucket has a configurable replication factor (default 3).
Bucketing assignment for writes:
- LogTable: round-robin or hash on a user-specified bucket key.
- PrimaryKeyTable: hash of the primary key — all updates to the same key always land on the same bucket.
Replication
Fluss uses a leader-follower + ISR protocol — the same conceptual model as Kafka’s ISR — re-implemented within the Fluss codebase:
- Client writes to the leader replica of the target bucket.
- Leader appends to its local LogStore and returns acknowledgment based on
acksconfig (0,1, orall). - Follower TabletServers pull new log entries from the leader (pull model, same as Kafka followers).
- Leader tracks each follower’s replication lag; followers falling too far behind are removed from the ISR.
- ISR membership is persisted by the CoordinatorServer on leader notification.
- On leader failure, the CoordinatorServer elects a new leader from ISR members and increments the leader epoch to fence zombie writers.
Dual Storage Model — LogTable vs. PrimaryKeyTable
LogTable
- Storage: purely a LogStore — append-only, ordered segments on local disk, older segments eligible for tiering to remote storage.
- Semantics: insert-only; no update, no delete.
- On-disk format: records written in a columnar (Apache Arrow-based) binary format, enabling server-side column pruning on reads.
- Kafka comparison: functionally similar to a Kafka topic-partition, but with native column projection, Arrow encoding, and a table-level schema enforced at the broker.
PrimaryKeyTable
This is where Fluss architecturally diverges from Kafka.
Left: the seven-step write lifecycle. Right: the dual-store internals of a leader TabletServer — LogStore acts as WAL (Write-Ahead Log) / changelog, KvStore provides O(log n) point lookups.
A write to a PrimaryKeyTable triggers two operations atomically on the leader:
- Append the change record to the LogStore (the WAL and changelog — durable, replicatable).
- Apply the upsert/delete to the KvStore (RocksDB in-memory MemTable).
RocksDB then handles the LSM lifecycle: flush MemTable → SST files (Sorted String Tables — immutable on-disk files containing key-value pairs in sorted order) → periodic compaction merges SSTables. The LogStore here is a write-ahead log and changelog, not a compacted representation of key state. Key-level deduplication happens inside RocksDB’s LSM tree, not by log-level compaction.
How This Differs from Kafka Log Compaction
| Dimension | Kafka Log Compaction | Fluss PrimaryKeyTable |
|---|---|---|
| Mechanism | Background log-cleaning thread rewrites segments in-place, removing superseded keys | Separate RocksDB KV engine maintains current state; log retains full ordered changelog |
| Read for latest value | Must scan log or read compacted tail | Direct RocksDB point lookup — O(log n) |
| Change feed | Compacted topic is the feed | LogStore changelog is the feed (separate from KV) |
| Recovery | Replay compacted log | Restore from KV snapshot + replay only delta changelog since snapshot |
| Data model | Single log with compaction | Physically separate log + KV engine per bucket |
KV Snapshot Mechanics
Snapshots are taken by hard-linking current RocksDB SST files into a staging directory, then uploading that staged directory to object storage (S3/OSS/HDFS). A hard link is another directory entry for the same inode, not a symlink. Creating one is cheap because it adds a name for existing file bytes instead of copying the bytes.
This is safe because RocksDB SST files (Sorted String Tables — immutable on-disk files containing key-value pairs in sorted order) are immutable after creation. If compaction later supersedes an SST, RocksDB creates new SST files and removes the old live directory entry. The snapshot staging directory still has its own hard link, so the old file blocks stay alive until the staged snapshot is uploaded and cleaned up.
The hard-link step is only the local snapshot optimization. Uploading to object storage still copies bytes, and mutable RocksDB metadata (CURRENT, MANIFEST, and required WAL files) must be copied or anchored so the remote snapshot represents one consistent database point. ZooKeeper records the snapshot pointer (/kv_snapshots).
On recovery: Fluss fetches the latest remote snapshot and replays the LogStore changelog from the snapshot’s log offset forward — avoiding full log replay from the beginning.
Concrete Write Path (End-to-End)
For a PrimaryKeyTable upsert, all seven steps in sequence:
Step 1 — Client routing. Client fetches bucket metadata from CoordinatorServer. Hashes primary key → bucket B → identifies leader TabletServer T₁.
Step 2 — LogStore write (WAL). Client sends upsert record to T₁ over Fluss’s binary RPC protocol. T₁ appends the record to its LogStore segment for bucket B. The write is now durable on the leader.
Step 3 — KvStore write. T₁ applies the upsert to RocksDB’s in-memory MemTable for bucket B. The LogStore write acts as the WAL for the KvStore — both happen atomically within the leader.
Step 4 — Follower replication. Followers T₂, T₃ issue fetch requests to T₁. T₁ returns new log entries. Followers append to their own LogStores and apply changes to their own RocksDB KvStores.
Step 5 — High Watermark advance. When all ISR followers confirm (with acks=all), T₁ advances the High Watermark (HW). Records at or below the HW become consumer-readable — typically within low milliseconds of the original write.
Step 6 — KV Snapshot (periodic). T₁ hard-links current immutable SST files into a staging directory, copies/anchors the mutable RocksDB metadata needed for a consistent checkpoint, and uploads the staged snapshot to object storage. Pointer written to ZooKeeper. Old log segments up to the snapshot’s anchor offset become eligible for local deletion (but may be retained for changelog consumers).
Step 7 — Tiering to lakehouse (minutes later). The Tiering Service Flink job reads from T₁’s LogStore changelog, converts to Paimon/Iceberg format, commits on checkpoint, and reports the tiered offset back to Fluss. Fluss expires log segments whose data is confirmed tiered.
Integration with Flink
Fluss ships a Flink connector (fluss-connector-flink) that registers Fluss tables in Flink’s Table/SQL API via a Flink Catalog. From a Flink job’s perspective:
- Reading a LogTable = unbounded streaming source (like a Kafka source), but with column pruning pushed down to the server.
- Reading a PrimaryKeyTable = changelog stream (
INSERT/UPDATE_BEFORE/UPDATE_AFTER/DELETErows), backed by the LogStore. - Lookup join on a PrimaryKeyTable = direct RocksDB point lookups against the KvStore on the leader TabletServer.
The lookup join is the key architectural advantage: Flink jobs performing dimension-enrichment lookups offload state entirely to Fluss, dramatically shrinking Flink checkpoint sizes. The mutable state lives in Fluss’s KvStore, not in Flink’s state backend.
Tiering to Paimon / Iceberg
Apache Paimon and Apache Iceberg are lakehouse table formats: they turn columnar files such as Parquet or ORC (Optimised Row Columnar) in object storage into transactional tables with snapshots, manifests, schema evolution, and time travel. They are cold/warm storage; Fluss is hot storage. Together they form a single logical table.
Paimon is Iceberg-like in the broad category sense, but it is more streaming/update-oriented in its storage model. A Paimon primary-key table stores bucket-local sorted runs plus optional changelog files; the Paimon writer, reader, and compaction code know how to merge those files into an updateable table. That makes Paimon a natural lakehouse target for Fluss PrimaryKeyTables. Iceberg remains useful for broad ecosystem interoperability and large analytic tables, but its core abstraction is an analytic table over immutable data/delete files rather than a lake storage engine with primary-key merge semantics.
Fluss-created Paimon/Iceberg tables include stream-position system columns such as __bucket, __offset, and __timestamp. These columns are what let Fluss later say: “the lake contains data through this Fluss offset; read fresh data from Fluss after that boundary.”
The Tiering Service
The Tiering Service is a long-running Flink job submitted by the cluster operator — not a background thread inside Fluss daemons.
Internal mechanics:
- TieringSource operator — uses Flink’s Source V2 API. Continuously polls the CoordinatorServer for bucket splits with untiered data. Each split maps to a (table, partition, bucket, offset-range) tuple.
- Data ingestion — reads raw log segments or KV changelog records from the relevant TabletServer.
- Format conversion — for PrimaryKeyTables, generates Paimon-format changelogs (native Paimon changelog files referenced by Paimon manifests) or merge-on-read files for Iceberg. For LogTables, converts Arrow-encoded records to Parquet/ORC.
- Lake commit — on each Flink checkpoint boundary, commits the lakehouse files and manifests, creating a new lake snapshot. Fluss annotates the lake snapshot with offset metadata such as
fluss-offsets, a JSON mapping from Fluss bucket/partition to the offset reached by that snapshot. - Fluss progress commit — after the lake snapshot exists, updates Fluss tiering state with the new snapshot ID and tiered offset. The ordering matters: committing Fluss progress before the lake snapshot would risk readers skipping data.
Freshness control: table.datalake.freshness (e.g., '30s') controls how frequently the tiering job checkpoints and commits — effectively the staleness bound of the lakehouse copy.
Read Modes After Tiering
| Query mode | How it works | Latency |
|---|---|---|
SELECT * FROM t (default) | Flink reads lake data up to the snapshot’s recorded fluss-offsets, then unions Fluss hot data above that offset | Seconds |
SELECT * FROM t$lake | Reads only the Paimon-tiered portion | Minutes (OLAP-optimised) |
| Streaming CDC (Change Data Capture) | Reads Fluss LogStore changelog directly | Sub-second |
The no-gap/no-double-count invariant is: data at or below the lake snapshot boundary comes from the lake; data above the boundary comes from Fluss. If the tiering job fails before the lake commit, the next run retries the same missing offset range. If it fails after the lake commit but before updating Fluss progress, the lake snapshot still contains the Fluss offset metadata, so recovery can rediscover the committed boundary instead of starting from stale Fluss-side state.
Consistency and Ordering Guarantees
Per-bucket ordering. All writes to a given bucket go through a single leader, and the LogStore is append-only. Strict ordering within a bucket; no cross-bucket ordering guarantee (same as Kafka per-partition semantics).
Read visibility. Records are visible only after the High Watermark advances (= all ISR members confirmed replication, when acks=all). No consumer sees uncommitted data.
Exactly-once for Flink sinks:
- LogTables: the connector tracks last written offset in Flink state; duplicate detection via producer epoch/sequence numbers (similar to Kafka’s idempotent producer).
- PrimaryKeyTables: upsert semantics are naturally idempotent — replaying the same upsert is a no-op on the key’s final value.
Leader Election and Replica Catch-Up
- Failure detected — TabletServer T₁ stops heartbeating → ZooKeeper ephemeral node disappears.
- CoordinatorServer reacts — removes T₁ from all ISR sets; triggers leader election among remaining ISR members.
- New leader elected — ISR member with highest log end offset preferred. CoordinatorServer increments leader epoch and writes to
/buckets/leader_and_isr. - Epoch fencing — any late write from the old leader carrying a stale epoch is rejected (prevents split-brain).
- Replica catch-up — when T₁ recovers, it rejoins as follower. It truncates its local log to the HW at failure time (removing uncommitted entries), then fetches from the new leader. Once within the lag threshold, it re-enters the ISR.
Current Status (June 2026)
| Dimension | Detail |
|---|---|
| Apache status | Incubating — not yet TLP (Top-Level Project) |
| Latest release | 0.9.1 (May 2026) |
| Planned 1.0 | Targeted June 2026 |
| Rust client | fluss-rust 0.1.0 (April 2026) |
| Community | 21 committers, 14 PPMC (Podling Project Management Committee) members |
| Production | Alibaba/Taobao — A/B testing for search/recommendation; handled tens of millions of requests during 618 Grand Promotion (June 2025) with sub-second latency, ~30% resource reduction, >100 TB state storage savings |
| Iceberg integration | 40 GB/s throughput, 3 PB storage, second-level CDC (Change Data Capture) latency (Alibaba) |
| ZooKeeper dependency | Present; roadmap targets replacement with internal KvStore + Raft |
| Supported lake formats | Apache Paimon, Apache Iceberg, Lance (vector format for AI use cases) |
See Also
- Introduction to Flink — Fluss’s primary query engine
- Flink Architecture — state backends and checkpointing that Fluss offloads
- Apache Paimon — the streaming/update-oriented lakehouse format Fluss can tier into
- Introduction to Apache Iceberg — one of the lakehouse formats Fluss tiers into
- Apache Kafka — the log broker Fluss complements (not replaces)
- Hard links and symbolic links — filesystem mechanism behind cheap RocksDB SST snapshots