A Kafka Streams application can restart quickly and still spend a long time rebuilding its local state before it resumes processing. If that state represents hours of aggregations or a large join table the recovery can keep consumer lag growing long after the process has returned.
Kafka 4.4.0 introduces transactional state stores through KIP-892 by Nick Telford. The change lets supported stores keep committed local state after errors that previously made Streams discard it under exactly-once processing. I’d be interested in measuring the recovery time here as much as the application’s normal throughput.
This is Part 3 of our Kafka 4.4.0 series. The earlier articles cover broker changes and share-group dead-letter queues.
Why State Needed Rebuilding
Consider an application that calculates a running total of card purchases for each account. It stores the totals locally and writes state changes to a Kafka changelog topic so another instance can rebuild them when necessary.
Under exactly-once processing the Kafka transaction coordinates input progress with output records and changelog records. Before transactional state stores the local store could already contain updates from a transaction that had not committed to Kafka. After an error Streams could not safely assume that the local data matched committed processing and had to discard and restore the affected state.
That recovery work can involve reading a substantial changelog even when only a small amount of recent processing failed. KIP-892 changes when updates become part of the committed local store so those recent writes can be discarded separately.
Transactional State Stores
The new approach buffers uncommitted writes and applies them to the local store after the corresponding Kafka transaction commits. The RocksDB implementation commits store data together with its recovery position so the persisted state can be matched to the changelog.
If the application fails before the transaction commits it can reuse that committed state and replay the work after the last successful commit. A failure after Kafka commits but before the local store finishes committing may still require catching the store up from the committed changelog. The change avoids treating every such error as a reason to rebuild everything.
It doesn’t eliminate restoration when a task moves to a machine without the state or when the local files are missing or corrupt. Initial bootstrap still has to obtain the data too.
Enable It Explicitly
Transactional state stores are off by default. These are example application properties for evaluating the feature with exactly-once processing.
processing.guarantee=exactly_once_v2
enable.transactional.statestores=true
statestore.uncommitted.max.bytes=67108864
default.interactive.query.isolation.level=READ_COMMITTEDThe final two values need to be chosen deliberately. The buffer threshold is 64 MiB by default and READ_COMMITTED is an explicit choice in this example because the default query isolation remains READ_UNCOMMITTED.
| Setting | Effect |
|---|---|
enable.transactional.statestores | Enables transactional buffering in store implementations that support it |
statestore.uncommitted.max.bytes | Requests an early commit when uncommitted data reaches the threshold |
default.interactive.query.isolation.level | Controls whether interactive queries can see staged writes |
The threshold applies across an application instance and is divided between its stream threads and a global state thread when present. It is not a hard process-memory limit. Work performed while processing a record can exceed it before Streams gets to commit and the application still needs memory for RocksDB and caches as well as its JVM heap. The 4.4.0 configuration definitions describe that early-commit behaviour.
Custom state stores must implement the necessary support. Setting the flag does not make an arbitrary store implementation transactional.
Query Visibility
Interactive queries read application state directly rather than consuming an output topic. Before choosing the isolation level decide whether callers are allowed to see a result that could be rolled back after an error.
| Isolation | What the caller can read | Consequence |
|---|---|---|
READ_UNCOMMITTED | Committed data plus staged updates | A recent result can include work that later fails |
READ_COMMITTED | Committed store data | Results exclude staged updates and reflect progress only after commit |
For the running-total example an uncommitted read could return 132 while the last committed total is still 120. A committed read stays at 120 until the new work commits. That delay is part of choosing committed visibility and should be included in application latency measurements.
IQv1 uses the configured default while IQv2 can override the isolation level for an individual query. This does not provide one atomic snapshot across every state store and Kafka partition in a distributed application.
Recovery Tests
I wouldn’t claim a recovery speed-up from enabling the setting without testing the application’s state size and deployment behaviour. A retained local directory and a new container with an empty volume are very different recovery cases.
| Test | Compare before and after enabling transactional stores |
|---|---|
| Failure during uncommitted processing | Time until processing resumes and the number of changelog records restored |
| Restart with the same local volume | Whether committed state is reused |
| Task moves to an instance with no local state | Restore traffic and time to catch up |
| Normal processing at the same input rate | Throughput and p50/p99 processing latency alongside CPU and total memory |
| Frequent early commits | Commit rate and transaction latency alongside buffer usage |
| Interactive queries during processing and recovery | Returned values and the delay before committed updates become visible |
The extra buffering and commit work can affect steady-state performance even if recovery improves. No general reduction in CPU or memory follows from this feature and we haven’t benchmarked it for this article.
Other Streams Changes
Kafka 4.4.0 also makes it easier to inspect and extend applications using the Streams rebalance protocol.
- KIP-1331 allows a broker-side plugin to store full topology descriptions sent by Streams clients. With that plugin configured the topology can be retrieved through the Admin API or the Streams group command. It isn’t a built-in graphical topology viewer.
- KIP-1357 adds broker-side custom task assignors selected per Streams group. This is separate from choosing an assignor for a conventional consumer group.
- KIP-1356 adds IQv2 queries that return headers from headers-aware state stores.
We’re grateful to Nick Telford and the reviewers and contributors who worked through the state-store changes. Retaining committed local data should make a substantial difference for applications whose recovery is dominated by unnecessary restores but the useful number will be the recovery time measured with your topology and storage.
Next is Part 4 on protecting Kafka Connect’s internal topic setup. For the surrounding broker measurements see the AxonOps Kafka monitoring guide.