KIP-1271: record headers in state stores, and the cost of a value format

TL;DR

• What: KIP-1271 (storage) and KIP-1285 (DSL opt-in) let a state store persist a record's Headers alongside its value. Shipped in Apache Kafka 4.3.0 (22 May 2026).
• Why: a schema id, a traceparent, a tenant tag, or lineage that rides on the input record now survives aggregation — readable via Interactive Queries and on the changelog.
• Kafka Streams: the core landed in 4.3.0 (timestamped key-value, windowed and session stores). Versioned stores, lazy header parsing, and some interactive-query verification are still open pull requests.
• StoatFlow 1.0.0: the full set — all four store families, versioned included — plus lazy header parsing, on the same on-disk and changelog bytes as Kafka Streams. Level on the core; a step ahead on the pieces Kafka has not merged yet.
• The engineering lesson: the on-disk change is one varint. Implementing it faithfully is an engine-wide change — because a value type is load-bearing.

A record's headers — the small key/value pairs that ride beside the key and value — carry the metadata a payload shouldn't: the schema id a Schema-Registry serialiser prepends, a traceparent for distributed tracing, a tenant or routing tag, a change-data-capture operation type. Through a stream processor they flow from input to processor to output. But at the state store boundary they have always stopped: aggregate a stream into a table, and the headers on the records that built each row are gone.

KIP-1271 and KIP-1285 close that gap, and they shipped in Apache Kafka 4.3.0. This post is three things in order: what the KIP is, where Kafka Streams has got to with it (and what is still in flight), and how StoatFlow compares — implementing the same feature, on the same bytes. The through-line is an engineering one: the on-disk change is a single varint, and implementing it faithfully is, unavoidably, an engine-wide change.

The feature: headers that survive the state store

Kafka Streams and StoatFlow alike — this section is the KIP itself.

There are two KIPs. KIP-1271 is the storage half: a state store can persist a record's Headers with the value, and you can read them back — through the Processor API, through Interactive Queries, and on the changelog. KIP-1285 is the DSL half: you opt in per store, or set an application-wide default, and the framework attaches the processing record's headers for you, with no headers API in your topology code.

It is opt-in by design — a store changes format only when you ask for it. That is a deliberate choice, not a hedge: most stores don't need headers, and the ones that do, say so.

Persisting headers enables a few things that were awkward or impossible before:

• Schema-id-aware deserialisation of stored state. If the value was written by a Schema-Registry serialiser, the schema id lives in the headers. Persist them, and the value can be deserialised correctly on the way out — including from the changelog, by a different consumer.
• Trace and lineage surviving an aggregation. The traceparent of the record that last updated a row is right there on the row.
• Routing or tenant metadata surviving a windowed aggregate, readable later through an Interactive Query without re-deriving it.

The format is small

Universal — this is the KIP's on-disk layout.

The format is the easy part. A stored value, with headers enabled, is laid out like this:

Diagram (placeholder — replace with a KSTD/Excalidraw visual): the on-disk value format.

  Stored value (headers enabled)
  ┌──────────────┬────────────────────┬──────────────────────┐
  │ headers_size │   headers block    │       payload        │
  │  (varint)    │  (key/value pairs) │  [ts(8)][value]  or  │
  │              │                    │       [value]        │
  └──────────────┴────────────────────┴──────────────────────┘

  No headers → a single 0x00 byte, then the payload unchanged.

The payload is exactly what the store held before — a timestamp and the value for timestamped stores, just the value for the others. There is no version byte; the store's location on disk is what tells the engine which format to expect. Empty headers cost one byte. That is the whole of it. Read the KIP and you would budget a day.

Why it isn't small: a value type is load-bearing

Universal — true of any KIP-1271 implementation, Kafka Streams included.

The cost is not in the format. It is in how many independent code paths hold, move, or reconstruct a value — each of which must now carry headers too, correctly, on the commit-critical path.

• There are four store families, not one. Key-value, versioned, windowed, and session stores each have their own typed wrapper, caching overlay, changelog decorator, and backends (in-memory, single-RocksDB, segmented-RocksDB). And they are not copies of each other: versioned stores keep history rather than overwriting; session stores have no per-record timestamp; windowed and session stores are split across RocksDB segments by time. The same idea has to land four different ways.
• The value serde signature changes. Reading a schema id out of the headers at store-access time means the value must be deserialised with its headers in hand — the three-argument, header-aware serde. That is a new boundary on every read and write.
• The changelog dictates where you pack. The changelog carries the headers natively — as Kafka record headers, beside an unchanged value — not buried in the value bytes. So the in-memory cache holds headers separately and folds them into the on-disk layout only when it writes to RocksDB, never on the changelog path. Get that boundary wrong and it's invisible until another consumer — or another engine — reads the changelog and finds the headers in the wrong place.
• Restoration is the inverse. State rebuilds from the changelog (unpacked value + native headers), so restoration has to re-pack each record into the on-disk format as it writes it.
• The DSL has to attach the headers for you. The useful end state isn't "a header-aware store exists"; it's that an ordinary aggregation, opted in, persists the headers of the record being processed — with no headers API in the topology. That needs the current record's headers captured per processing context and threaded into the store behind an adapter.

How load-bearing? Kafka Streams implemented this across more than forty pull requests, spanning some two dozen Jira sub-tasks, over roughly five months: a foundation layer (the byte format, the header-aware serdes), then a multi-PR series per store family, then the DSL wiring. The format is one varint; the change is the engine.

Kafka Streams vs StoatFlow

The core of KIP-1271/1285 shipped in Kafka 4.3.0. A handful of pieces are still open pull requests on Kafka's side — versioned stores, lazy header parsing, and some interactive-query verification. StoatFlow implements the full set on the same bytes, which leaves it level on the core and a step ahead on the still-open pieces.

Two rows are where StoatFlow is currently ahead, and both are simply pieces Kafka's contributors have in flight rather than design disagreements:

• The versioned store family — both the store itself and versioned KTables in the DSL. On the same lazy dual-column-family migration as the rest.
• Lazy header parsing. A Headers wrapper that defers parsing the header block until something actually reads it. A header-agnostic value serde — or a DSL read that only wants the value and strips the headers — never pays for the parse; a Schema-Registry serde that reaches for the schema id triggers it on first access. It is the biggest win for versioned stores, which have no caching layer in front of them. (Kafka's own version of this is KAFKA-20155, still open.)

Everywhere else the two line up — including where they both stop. Headers on join buffers, non-timestamped header-aware stores, and headers-specific metrics are all outside both KIPs, and StoatFlow draws those lines in the same place Kafka does.

StoatFlow's take: an upgrade that costs nothing

StoatFlow-specific.

Here is the operational question that outranks the rest for anyone running a stateful application: turning headers on changes the value format, so what happens to the gigabytes of state already on disk in the old format?

There are two honest answers. Eager migration rewrites every entry on the first boot with headers enabled — simple, but a migration pause and an I/O spike proportional to your state size, before the app processes a record. Lazy migration rewrites nothing up front: keep the old data where it is, write new data in the new format, reconcile the two on read — using two RocksDB column families, independent keyspaces inside one database. Kafka Streams takes the lazy path; so does StoatFlow.

Diagram (placeholder — replace with a KSTD/Excalidraw visual): the store, the moment headers are enabled.

  One RocksDB database, two column families
  ┌──────────────────────────────────────────────────────────┐
  │  ┌──── legacy CF (old format) ───┐  ┌── headers CF ─────┐  │
  │  │  key → value                  │  │ key → headers ++  │  │
  │  │  (everything already on disk) │  │        value      │  │
  │  └───────────────────────────────┘  └───────────────────┘  │
  │                                                            │
  │   read   →  check both; convert a legacy hit on the fly    │
  │   write  →  land in the headers CF, remove from legacy     │
  └──────────────────────────────────────────────────────────┘

  Enable the flag → nothing is rewritten up front.
  Each key migrates itself the first time it is written.

You flip the flag and restart, and not a byte of existing state is rewritten. A read checks both keyspaces and converts an old-format hit on the way out — old entries read back with empty headers, which is exactly right, because they predate the feature. A write lands in the new keyspace and clears the old copy, so each key upgrades itself the first time it's touched; the legacy keyspace drains to nothing over time. With the flag off, the store is byte-for-byte what it was — fully reversible.

There is an honest nuance worth stating, because it cuts against our own grain. Kafka Streams went lazy for a specific reason: in a horizontally-scaled deployment, an eager migration would block a partition's recovery during a rebalance, and a rebalance is exactly when you cannot afford a long pause. StoatFlow has no rebalances — it runs as a single instance by design — and opens each store once at startup. That argument for lazy doesn't apply to us; eager would have worked. We chose lazy anyway, because a frictionless opt-in — flip a flag, restart, no pause, no I/O spike, instantly reversible — is worth more than the column family it saves, and because it keeps us byte-for-byte on Kafka's on-disk layout. Choosing the more involved implementation to preserve that compatibility is the trade we keep making.

It covers all four families. Key-value and versioned stores carry the two column families directly; windowed and session stores are split into time segments, so each segment gets its own pair and upgrades independently as records land in it. The versioned store is the one variation worth a footnote: it keeps every version of a key rather than overwriting, so old versions are legitimate history surfaced through the merge, not stale copies to be drained.

The lazy path has sharp edges you only meet once you commit to it. A delete must remove the key from both keyspaces — miss the legacy copy and the old value resurrects through the read-merge, a tombstone that didn't take. And you cannot delete from the legacy keyspace any data the current transaction hasn't committed, or an aborted transaction takes real data with it. We learned the restoration corner from a real-Kafka integration test: a unit test restored happily because it exercised the typed wrapper, but the end-to-end test — write, wipe local disk, restore from an actual changelog — came back with malformed values, because the re-pack step was keyed off the wrong object on the restore path. Only restoring against a real broker surfaced it. These are the price of not rewriting your state up front, and they're exactly the kind of thing a faithful implementation has to get right out of sight.

Scope, and what it costs

The scope lines are universal (they're the KIP's); the costs are universal effects.

The feature draws two deliberate lines, and StoatFlow draws them where Kafka does:

• Timestamped stores only. There are no non-timestamped header-aware stores — the key-value-with-headers store is always timestamped. That is the KIP's scope, not a missing feature.
• Joins are out of scope. A KStream-KStream join buffers records in a plain window store, and neither KIP wires headers through it. Headers on join state would be a step beyond the KIPs, with its own design questions.

A few behaviours the KIP explicitly leaves for later — making stored headers visible to user functions, attaching them to DSL results, downgrade — are deferred in both Kafka Streams and StoatFlow.

And it has costs, which are the expected effects of doing what it says, not drawbacks:

• Headers take space. Every stored value grows by the header block (empty headers: one byte). That is the storage cost of keeping them — on disk, in the cache, and on the changelog.
• A transient read-merge. While a store is mid-upgrade, a read that misses the new keyspace does a second lookup in the legacy one and converts on the fly. It drains to a single keyspace as keys are rewritten; steady state is a single lookup.

The lesson

Universal.

The size of a change is set by how load-bearing the thing you change is, not by the size of its specification. A serialised value is about as load-bearing as it gets in a stateful stream processor: it lives in the cache, on the changelog, in the restoration path, behind the query API, and in the on-disk format — every one of them on the commit-critical path. So a one-varint value-format change is, unavoidably, a state-layer change. Kafka Streams measured it in dozens of pull requests; StoatFlow's effort was comparably large.

And the part a user actually experiences is the smallest slice of that work: a flag that turns the feature on and lets state already on disk keep working, untouched, until it upgrades itself one key at a time. The reason that flag is cheap to use is the same reason the rest was expensive — staying faithful to a format down to the byte, in every path that touches a value.

This sits directly on top of an earlier foundation: KIP-1035, which moved changelog offsets into a RocksDB column family of their own. The column-family mechanism that makes lazy header migration possible is the same one that holds those offsets. The on-disk layout keeps earning its keep.

Read on:

• KIP-1271: Allow to Store Record Headers in State Stores
• KIP-1285: DSL support for headers in state stores
• KIP-1035: why Kafka Streams 4.3 lets state stores own their offsets
• Apache Kafka 4.3.0 release announcement