StoatFlow
Announcement·

Kafka Streams 4.4.0: transactional state stores, headers-aware partitioning and queries

What Apache Kafka 4.4.0 brings to Kafka Streams: opt-in transactional state stores (KIP-892), headers in the partitioner and IQv2, the GlobalKTable joiner fix, and static membership and plugins for the Streams Rebalance Protocol — plus where StoatFlow stands.
Author avatar Hartmut Armbruster

Kafka Streams 4.4.0 delivers a change that exactly-once users have been waiting years for: opt-in transactional state stores (KIP-892). Turn them on, and an error under exactly-once no longer means wiping local state and rebuilding it from the changelog. The work traces back to 2021, and 4.3 laid the groundwork by moving changelog offsets into the stores themselves (KIP-1035; see our deep dive). Around it, record headers reach the partitioner and IQv2, a long-standing GlobalKTable joiner bug gets a proper fix, and the Streams Rebalance Protocol gains static membership and broker-side plugins.

In StoatFlow we already ship the partitioner, joiner, close and config changes; transactional-store parity, IQv2 and the headers-aware queries are planned for 1.0.0. Where StoatFlow behaves differently, or a change does not apply to a single-instance engine, a short note says so.

TL;DR

  • KIP-892: opt-in transactional state stores — no more state wipe after an error under exactly-once — plus an uncommitted-bytes budget and isolation levels for interactive queries.
  • KIP-1321 + KIP-1356: StreamPartitioner and IQv2 become headers-aware. Windowed sinks with a header-sensitive key serialiser can move partitions on upgrade.
  • KIP-1340: stream–GlobalKTable joiners can see the join key, not just the stream key.
  • Streams Rebalance Protocol (KIP-1071): static membership, topology descriptions (KIP-1331) and broker-side custom assignors (KIP-1357).
  • Smaller changes: GroupMembershipOperation.DEFAULT (KIP-1284), application.server validated at startup (KIP-1245), clearer cleanUp() (KIP-1283).
  • StoatFlow: the partitioner, joiner, close and config changes ship today; the KIP-892 APIs, IQv2 and KIP-1356 are planned for 1.0.0.

Transactional state stores (KIP-892)

Set enable.transactional.statestores=true and state stores buffer their writes until the corresponding changelog transaction commits (KIP-892). The consequence matters most under exactly-once. Without them, an error there means the store may hold uncommitted writes, so Kafka Streams wipes it and restores it from the changelog before processing can resume — on a large store, the most expensive recovery path there is. With transactional stores, local state stays consistent with the changelog, and under either guarantee only the writes since the last successful commit are replayed.

The buffer is bounded by statestore.uncommitted.max.bytes: 64 MiB per instance by default, split across the stream threads (and the global thread, when the topology has global stores); -1 disables the bound. Crossing it triggers an early commit, whatever commit.interval.ms says. Each store reports its share through the new uncommitted-bytes metric and StateStore#approximateNumUncommittedBytes().

Interactive queries gain an isolation level. READ_UNCOMMITTED (the default) sees buffered writes; READ_COMMITTED sees only what the changelog has committed. default.interactive.query.isolation.level sets it for IQv1, and IQv2 can override it per query with StateQueryRequest#withIsolationLevel(...). Store views come from the new readOnly(IsolationLevel) on the key-value, window, session and versioned store interfaces.

The flag is off by default, so nothing changes until you opt in. The 4.4 upgrade guide does not cover it yet; the KIP and the Streams config reference are the documentation.

StoatFlow: every store the DSL selects already buffers each epoch's writes until the commit barrier's transaction commits. The 4.4 API surface — the two config keys, isolation levels, readOnly(IsolationLevel) and the uncommitted-bytes gauge — is planned for 1.0.0.

Headers, end to end (KIP-1321, KIP-1356)

Headers carry the metadata a payload shouldn't — a schema id, a trace context, a tenant tag — so every place a stream processor drops them is a place lineage breaks. Kafka Streams 4.3 let state stores persist them (KIP-1271; see our deep dive). 4.4 closes two more of the gaps.

Partitioning (KIP-1321). StreamPartitioner gains partitions(topic, key, value, headers, numPartitions). The old four-argument method is deprecated for removal in 5.0, but it stays the interface's single abstract method, so existing lambdas still compile. The built-in partitioners now pass the record's headers to the key serialiser, and KafkaStreams#queryMetadataForKey has matching Headers overloads so interactive-query routing agrees with where the key was written.

Querying (KIP-1356). Four new IQv2 query types — TimestampedKeyWithHeadersQuery, TimestampedRangeWithHeadersQuery, TimestampedWindowKeyWithHeadersQuery and TimestampedWindowRangeWithHeadersQuery — return headers-aware stores' contents as ReadOnlyRecords: key, value, timestamp and read-only headers. Record now implements ReadOnlyRecord. Existing query types against headers-aware stores keep working and simply strip the headers.

StoatFlow: the partitioning half ships today — the five-argument partitions, header-aware serialisation at sinks and in in-memory repartitioning, and the Headers overloads of queryMetadataForKey. IQv2, and with it the four headers-aware query types, is planned for 1.0.0.

The GlobalKTable joiner fix (KIP-1340)

In a stream–GlobalKTable join, the join key is computed by your KeyValueMapper, yet ValueJoinerWithKey always received the stream record's key. The types lined up, so nothing flagged it — the kind of bug that survives code review. 4.4 fixes it with a small API addition: ValueJoinerWithStreamAndMappedKey (KIP-1340), whose apply(streamKey, mappedKey, streamValue, tableValue) gets both keys, through four new join/leftJoin overloads. The ValueJoinerWithKey global-table overloads are deprecated; their behaviour is unchanged. In a left join, a mapper that returns null still emits: the joiner sees a null mapped key and a null table value. Stream records with a null value are still dropped.

StoatFlow: the same fix and the same deprecations, plus Kotlin-lambda overloads.

The Streams Rebalance Protocol (KIP-1071, KIP-1331, KIP-1357)

Some context first. Kafka 4.0 made KIP-848, the Next Generation of the Consumer Rebalance Protocol, generally available. It moves group coordination to the broker and says goodbye to "stop-the-world" rebalances: instead of every member pausing while the group renegotiates, the coordinator reconciles assignments incrementally and members keep working on the partitions they retain.

The Streams Rebalance Protocol (KIP-1071) builds on it and brings the same model to Kafka Streams. Task assignment, internal-topic creation and the topology metadata the group needs all move from the client into the group coordinator, which makes Streams tasks a first-class citizen of the Kafka protocol. It arrived as early access in 4.1, became generally available for a core feature set in 4.2, and is enabled with group.protocol=streams.

4.4 fills in more of the picture. Static membership arrives: group.instance.id now works with the Streams Rebalance Protocol, so a restarting instance can keep its tasks instead of triggering a reassignment. It needs 4.4 on both clients and brokers — an older broker rejects it with INVALID_REQUEST, which shuts the client down, so remove group.instance.id before downgrading a broker. For topologies without persistent state stores, Kafka Streams generates a new process ID on every restart, which triggers a reassignment and largely cancels the benefit.

Two new KIPs extend the protocol on the operations side. KIP-1331 lets clients push their topology description to the group coordinator, where a pluggable backend stores it — so an operator can see what a group runs without access to its source code. Admin#describeStreamsGroups returns it with DescribeStreamsGroupsOptions#includeTopologyDescription(true), as does kafka-streams-groups.sh --describe --topology. Nothing is pushed until the broker sets group.streams.topology.description.plugin.class; clients can opt out with topology.description.push.enabled (default true). KIP-1357 makes the broker-side task assignor pluggable: the broker registers assignors in group.streams.assignors (default sticky), and a group picks one with streams.assignor.name.

StoatFlow: not applicable, by design. A StoatFlow application is a single instance in a classic consumer group, so there are no tasks to assign and no streams group to push a topology to. topology.description.push.enabled is accepted and ignored; the same description is always available from GET /topology/ks.

Smaller changes

  • GroupMembershipOperation.DEFAULT (KIP-1284). A new default for CloseOptions adapts to the protocol: under the classic protocol the member stays in the group; under the Streams Rebalance Protocol dynamic members leave and static members stay until the session timeout. close() and close(Duration) now use it.
  • application.server validated at startup (KIP-1245). A malformed value now fails with a ConfigException when StreamsConfig is created, not later during assignment.
  • cleanUp() semantics (KIP-1283). cleanUp() may leave the state directory behind when it holds only process metadata and the lock file. That is by design, and no longer logs a warning.
  • Async bootstrap DNS forced off (KIP-909). Kafka Streams does not support the asynchronous bootstrap resolution mode yet: it forces bootstrap.resolve.timeout.ms to 0 for every client it creates and ignores a positive value with a warning.
  • Metrics and upgrade. record-e2e-latency-* at terminal nodes now measures to the end of processing; every record in a batch gets the batch's completion time, so read the values as upper bounds. upgrade.from accepts 4.3.

StoatFlow: GroupMembershipOperation.DEFAULT resolves to the leaveGroupOnClose setting, which defaults to staying in the group, matching the classic protocol. application.server is validated when the config is built, and cleanUp() always removes the whole state directory — StoatFlow keeps no process-metadata file there. Its end-to-end latency already included processing time, and upgrade.from is accepted and ignored.

What to watch when upgrading

  • Windowed sinks with header-sensitive key serialisers can move partitions. WindowedStreamPartitioner now passes the record's headers to the key serialiser, where 4.3 passed empty headers. If your serialiser overrides serialize(topic, headers, data) and its output depends on the headers, windowed records may land on different partitions than before the upgrade, which breaks co-partitioning with existing data. Wrap the serialiser, or supply a partitioner that ignores headers. Other sinks already serialised the key with headers in 4.3.
  • Processors sharing a StoreBuilder instance are regrouped (KAFKA-20464). Processors that return the same StoreBuilder instance from ConnectedStoreProvider#stores() now land in one subtopology. Task IDs change, so state restores from the changelog on the first start after the upgrade — plan for the restore time. Linking processors to a store by name is unaffected.
  • REMAIN_IN_GROUP now means it. Under the Streams Rebalance Protocol, 4.3 members left the group on every close, whatever CloseOptions said. In 4.4 REMAIN_IN_GROUP is honoured, so a member closed with it keeps its tasks until the session timeout. Dynamic members still leave on a plain close(); static members stay.
  • Deprecation warnings on GlobalKTable joins. Move ValueJoinerWithKey joiners to ValueJoinerWithStreamAndMappedKey; ignoring the mapped key keeps today's behaviour.
  • Transactional state stores are opt-in. Enable them deliberately and size statestore.uncommitted.max.bytes for your write volume per commit interval.

Worth the upgrade

4.4.0 is a substantial Kafka Streams release, and a good one. If you run exactly-once with large stores, transactional state stores alone change what a failure costs you: try enable.transactional.statestores on a staging copy first and watch the uncommitted-bytes gauge against your commit interval. Everyone else gets a partitioner and IQv2 that see record headers, a joiner that sees the key it joined on, and a rebalance protocol that keeps getting better. Congratulations to the Apache Kafka community on the release.

The Kafka Streams upgrade guide and the KIPs linked above carry the detail. For the method-by-method status in StoatFlow, see the Kafka Streams compatibility matrix.

Three clocks: stream time, Flink watermarks, and the one watermark StoatFlow computes

Kafka Streams, Flink and StoatFlow answer "is this window done?" with three different clocks. StoatFlow runs two of them: it accepts a late record the way Kafka Streams does, on the operator's stream time, and closes the window the way Flink does, on a single global watermark. Part 1 of a four-part series on event time — including two corrections to a claim our own compatibility matrix made about grace.