Transactional state stores arrive in Kafka Streams 4.4: how KIP-892 works

TL;DR

• What shipped: Kafka Streams 4.4.0 delivers KIP-892, opt-in with enable.transactional.statestores=true. A store buffers its writes until the changelog transaction commits, so an error under exactly-once no longer means wiping local state and restoring it from scratch.
• How: each store stages its writes in an on-heap sorted map that reads overlay, and in a plain RocksDB WriteBatch. At commit the changelog offset joins the batch, and one write applies both.
• Also new: an uncommitted-bytes budget that forces an early commit, an uncommitted-bytes metric, and READ_COMMITTED interactive queries.
• The catch: it is off by default and not yet in the upgrade guide, custom stores cannot opt in, and a store's durable offset is still only as fresh as RocksDB's last memtable flush — the KIP-1035 trait that surprised 4.3 users with full restores.
• In StoatFlow: stores have been transactional from the start. The 4.4 surface — isolation levels, the gauges, the three config keys — is part of StoatFlow 1.0.0.

In June we called KIP-1035, in Kafka 4.3, the keystone for transactional state stores. Kafka Streams 4.4.0 builds the arch on it. KIP-892 — proposed by Nick Telford, with work on the problem going back to 2021 — is in, and turning it on changes the most expensive moment in an exactly-once application's life: recovering from an error.

Why exactly-once wiped the store

Under exactly-once, a task writes to its RocksDB store as records arrive, long before the Kafka transaction that covers them commits. RocksDB runs with its write-ahead log off and flushes a memtable when it fills, so the writes of an open transaction can reach disk. If the process dies, the broker aborts that transaction — but its writes may already be in the store.

Kafka Streams had no way to tell committed state from uncommitted. Under exactly-once it deleted a task's .checkpoint file once it had loaded it and wrote it again only on a clean close, so after a crash a store held data but no offset. Its own log line gives the reasoning: "under EOS it has the risk of getting uncommitted data in stores we have to treat it as a task corruption error and wipe out the local state". The task was wiped and rebuilt from the changelog, and restore time tracked total state size — at Meltwater, as Nick showed at Kafka Summit London 2024, around 45 stores and roughly 8 TiB of changelog.

(Our KIP-1035 post first described this the other way round, as an offset running ahead of the data. It now carries the corrected mechanism and diagram.)

How 4.4 implements it

With the flag on, every built-in store gets a transaction buffer, and nothing reaches RocksDB until the commit:

• The staging map is a sorted map on the heap, key to new value or tombstone. Every read goes through it: a point lookup checks it first, and range scans merge its entries over the store's own iterator, with tombstones hiding deleted keys. Interactive-query threads read it under a read lock.
• The write batch collects the same writes for RocksDB stores. At commit — after the changelog transaction has committed — the changelog offsets and the interactive-query position go into the same batch under the offsets column family, and one db.write applies it: the data and its offset together, or neither. In-memory stores have no batch; their commit copies the staging map into the store under a write lock.

The design moved on the way. Earlier drafts buffered in a RocksDB WriteBatchWithIndex, and the first implementation staged into a concurrent skip list. What shipped is a plain TreeMap behind a read-write lock: the stream thread is the map's only writer, so the skip list's concurrency cost it throughput for nothing. And because the staging map does all the reading, the batch needs no index.

What a crash costs now

One qualifier applies on every row. RocksDB still runs without a write-ahead log, so a committed write sits in a memtable until RocksDB flushes it or the store closes cleanly. After a crash, the offset that survives is the last flushed one: consistent with the data, but possibly several commits old. KIP-892 makes the restore a replay instead of a rebuild; it does not make it free.

The budget, the metric, the isolation levels

• statestore.uncommitted.max.bytes bounds the buffers: 64 MiB per application instance by default, divided equally across the stream threads (and the global thread when the topology has global stores). Crossing it triggers an early commit, whatever commit.interval.ms says; -1 disables the bound.
• uncommitted-bytes is a new store-level metric at the INFO recording level, and StateStore#approximateNumUncommittedBytes() reports the same number in code.
• Interactive queries gain an isolation level. READ_UNCOMMITTED, the default, reads through the staging map; READ_COMMITTED reads the store alone. default.interactive.query.isolation.level sets it for IQv1, IQv2 can override it per query with StateQueryRequest#withIsolationLevel(...), and readOnly(IsolationLevel) on the key-value, window, session and versioned store interfaces returns a view at either level. A committed read trails processing by up to one commit interval — 30 seconds by default under at-least-once.

Coverage, and why opt-in is right for now

Every built-in store type takes part: the RocksDB key-value stores (timestamped and headers variants included), the segmented window and session stores, versioned stores, and the in-memory key-value, window and session stores — global stores too.

Custom stores cannot. The buffer classes are package-private, and the only public hooks are approximateNumUncommittedBytes() and readOnly(...), whose defaults report nothing and return the store itself. There is a sharper edge: 4.4 skips the exactly-once wipe on the strength of the task-wide flag, not per store. As we read the 4.4 source, a custom persistent store that writes straight to disk loses the protection the wipe used to give it, and after a crash can keep writes from an aborted transaction. Keep custom stores out of transactional applications until they can buffer.

Three fixes landed before the release: a stream–stream outer or left join that crashed under exactly-once with transactional stores (KAFKA-20749), a RocksDB store that failed to reopen (KAFKA-20753), and putAll bypassing the buffer (KAFKA-20933). This is new code on the most sensitive path in the engine. Shipping it off by default, documented so far only in the configuration reference, is a sensible way to land it.

KIP-1035, five months on

KIP-892 rests on the groundwork 4.3 shipped, and that groundwork taught some lessons in production. If you run 4.3 or plan to move to 4.4, these are the ones to know:

• Skip 4.3.0. Its offsets column family leaked native memory on close paths, and under cascading task closes that ended in an out-of-memory kill (KAFKA-20616, fixed in 4.3.1 and 4.4.0).
• Quiet stores restore from stale offsets. With the forced flush on commit gone, a store's offset becomes durable only when its memtable fills (16 MB by default) or the store closes cleanly. A low-traffic store that dies uncleanly can come back with an offset that changelog retention has already passed, and that means a full restore — "far more often than in 4.2", windowed and segmented stores hardest (KAFKA-20663). 4.3.1 fixed two bugs that made it worse; the trait itself remains in 4.4, and KIP-892 does not change it. The documented mitigation is a smaller write_buffer_size for such stores through a RocksDBConfigSetter, and shutdowns that complete.
• State directories were cleaned too early. The per-commit checkpoint file used to refresh the task directory's timestamp; without it, an actively committing task looked stale. When a stream thread failed and its task moved to another thread on the same instance, the state-directory cleaner could delete the directory in the gap between the two locks, and the task restored from scratch (KAFKA-20805, fixed in 4.3.2 and 4.4.0). It bites only where a dirty close does not wipe — at-least-once, or exactly-once with transactional stores — so it matters more once KIP-892 is on.
• No in-place downgrade below 4.3. Older versions do not know the offsets column family and crash on startup; delete state.dir first and let the stores restore from their changelogs.
• Restores of windowed stores get cheaper. With no stored offset, a window or session store now skips changelog records that have already expired instead of replaying them (KAFKA-13499, in 4.4.0). The same optimisation for stores that do have an offset (KAFKA-20940) is in 4.3.2 and comes to 4.4 in 4.4.1.

On the 4.3 line, run 4.3.2.

In StoatFlow

StoatFlow has buffered this way from its first release, for every store the DSL selects. Each store keeps the current commit's writes in an in-memory overlay that reads see; the changelog transaction commits first, and only then does the commit reach RocksDB, with the offset never ahead of the data. A crash therefore never leaves uncommitted writes in a store, and StoatFlow has never needed a consistency wipe (crash recovery).

What 4.4 lets a Kafka Streams application observe about transactional stores is part of StoatFlow 1.0.0:

• READ_COMMITTED reads through stoatflow.state.interactive-query-isolation-level, or Kafka Streams' default.interactive.query.isolation.level, and readOnly(IsolationLevel) for a single store handle (isolation level for interactive queries).
• Two gauges: stoatflow.store.uncommitted.bytes per store, for every store type, and stoatflow.barrier.epoch.uncommitted.bytes, the application total; approximateNumUncommittedBytes() reports the per-store value in code.
• The three Kafka Streams keys. statestore.uncommitted.max.bytes sets StoatFlow's budget, default.interactive.query.isolation.level the isolation level, and enable.transactional.statestores is accepted — set to false, it warns, because StoatFlow's stores cannot be made non-transactional.

One caveat both engines share. StoatFlow also runs RocksDB without a write-ahead log, with a 16 MiB write buffer by default, so after a crash a store restores from its last flushed offset: a replay of everything after it, or a full restore if changelog retention has passed it. In-memory stores always restore in full. We once wrote that a delta restore completes in under a second whatever the state size; that was never measured, and the honest answer is that it scales with how far the last flush trails. The same mitigation applies — a smaller write buffer for quiet stores, set per store through the rocks-db-config-setter, and pods given time to shut down cleanly.

Worth enabling?

For an exactly-once application with large persistent stores and only built-in store types: yes. Turn it on in staging first, watch uncommitted-bytes against your commit interval, and size statestore.uncommitted.max.bytes for your peak writes per commit — an early commit is cheap, a wipe is not. Leave it off where custom stores are involved. At-least-once applications gain less, since they were never wiped, but their stores stop holding writes past the last commit, and READ_COMMITTED queries become possible.

And whichever engine you run, give quiet stores a smaller write buffer.

KIP-892 took five years from the first ticket to a release, and much of that was the patient work of getting KIP-1035 in first. Congratulations to Nick Telford and everyone who reviewed and tested it — it is a substantial improvement to Kafka Streams.

Read on:

• KIP-892: Transactional Semantics for StateStores and its umbrella ticket KAFKA-14412
• KIP-1035: StateStore managed changelog offsets
• Apache Kafka 4.4.0 release notes
• KIP-1035: Why Kafka Streams 4.3 lets state stores own their offsets — our groundwork post
• Kafka Streams 4.4.0: transactional state stores, headers-aware partitioning and queries — the whole release