Kafka Streams Upgrade Guide for Confluent Platform

Upgrading a Kafka Streams application to Confluent Platform 8.4.0 (Kafka Streams 4.4.0) is supported from any earlier Kafka Streams version, either online by using two rolling bounces with the upgrade.from configuration or offline by stopping all instances and swapping the JAR files.

Upgrade from older versions

Upgrading from any earlier Kafka Streams version to Confluent Platform 8.4.0 is supported.

If you’re upgrading from Confluent Platform 7.4.x (Kafka Streams 3.4) or earlier, you must do two rolling bounces.

1. During the first rolling bounce, set the upgrade.from="<older_version>" configuration. Possible values are 2.4 through 3.4.

2. During the second bounce, remove the upgrade.from config.

This is required to handle two changes safely:

• A change in foreign-key join serialization format.

• A change in the serialization format for an internal repartition topic. For more information, see KIP-904.

1. Prepare your application instances for a rolling bounce, and ensure that the upgrade.from config is set to the version it is being upgraded from.

2. Bounce each instance of your application once.

3. Prepare your newly deployed Confluent Platform 8.4.0 (Kafka Streams 4.4.0) application instances for a second round of rolling bounces. Be sure to remove the value for the upgrade.from config.

4. Bounce each instance of your application once more to complete the upgrade.

As an alternative, an offline upgrade is also possible. Upgrading from any versions as old as Confluent Platform 3.0.x (Kafka Streams 0.10.x) to Confluent Platform 8.4.0 (Kafka Streams 4.4.0) in offline mode requires the following steps:

1. Stop all old application instances, for example, Confluent Platform 3.0.x.

2. Update your code and swap old code and JAR files with new code and new JAR files.

3. Restart all new Confluent Platform 8.4.0 (Kafka Streams 4.4.0) application instances.

Upgrade to Confluent Platform 8.4.0 from Confluent Platform 7.1.x or later

Compatibility

Kafka Streams applications built with Confluent Platform 8.4.0 are forward and backward compatible with certain Kafka clusters.

Forward-compatible to newer clusters up to Confluent Platform 8.4.0:

Existing Kafka Streams applications built with Confluent Platform 5.1.x (Kafka Streams 2.1) and later work with upgraded Kafka clusters running Confluent Platform 8.4.0.

Backward-compatible to older clusters down to Confluent Platform 5.1.x:

• New Kafka Streams applications built with Confluent Platform 8.4.0 work with older Kafka clusters running Confluent Platform 5.1.x (Kafka 2.1) or later. Exactly-once semantics require clusters running Confluent Platform 5.5.x (Kafka 2.5) or later.

• Kafka clusters running Confluent Platform 5.0.x or earlier are not compatible with new Confluent Platform 8.4.0 Kafka Streams applications.

• Static membership with group.protocol=streams requires Kafka clusters running Confluent Platform 8.4.x or later.

Compatibility matrix

The following table shows the versions of the Kafka Streams API that are compatible with various Kafka broker versions. For versions earlier than Confluent Platform 7.4.x (Kafka Streams 3.4.x), see Legacy Streams Upgrade Guide.

Kafka Streams API compatibility with Kafka broker versions

Kafka Streams API version	Broker Confluent Platform 5.4.x - 8.0.x (Kafka Streams 2.4.x - 4.0.x)	Broker Confluent Platform 8.1.x - 8.4.x (Kafka Streams 4.1.x - 4.4.x)
Confluent Platform 5.4.x - 5.5.x (Kafka Streams 2.4.x - 2.5.x)	compatible	compatible
Confluent Platform 6.0.x - 8.4.x (Kafka Streams 2.6.x - 4.4.x)	compatible; enabling exactly-once v2 requires broker Confluent Platform 5.5.x / Kafka Streams 2.5.x or higher	compatible; with Kafka Streams 4.4.x clients, static membership with group.protocol=streams requires broker Confluent Platform 8.4.x / Kafka Streams 4.4.x or later

Starting with Confluent Platform 8.0.x (Kafka Streams 4.0), the Streams API isn’t compatible with Kafka clusters running versions earlier than Confluent Platform 5.1.x (Kafka 2.1).

RocksDB compatibility matrix

The following table shows which versions of RocksDB are included with corresponding versions of Kafka Streams. This is important when you use state stores backed by RocksDB, as mismatched versions may lead to runtime errors or data incompatibility.

Kafka Streams version	Confluent Platform version	RocksDB version	Notes
4.4.x	8.4.x	10.1.3
4.3.x	8.3.x	10.1.3	New offsets column family (KIP-1035); refer to KIP-1035 downgrade guidance.
4.2.x	8.2.x	10.1.3
4.1.x	8.1.x	9.7.3
4.0.x	8.0.x	9.7.3	Significant API changes, see Upgraded RocksDB dependency.
3.5.x – 3.9.x	7.5.x – 7.9.x	7.9.2	Requires newer GCC; not compatible with RHEL 7.
3.0.x – 3.4.x	7.0.x – 7.4.x	7.1.2	Downgrading from 3.0.x or newer to 2.8.x or older version requires special attention due to on-disk format change.
2.6.x – 2.8.x	6.0.x – 6.2.x	5.18.4
2.3.x – 2.5.x	5.3.x – 5.5.x	5.18.3	Can specify more RocksDB configurations, which helps to limit RocksDB off-heap memory usage.
2.2.x	5.2.x	5.15.10
2.1.x	5.1.x	5.14.2
2.0.x	5.0.x	5.7.3

The exact RocksDB version is a transitive dependency of the Kafka Streams artifact and is managed automatically if you use Maven or Gradle, so updating Kafka Streams typically updates RocksDB.

If you are customizing or overriding the RocksDB version, for advanced tuning or bugfixes, be careful to match major versions and API compatibility, especially after the significant upgrade in Kafka Streams 4.0.0.

For more information, see:

• Kafka Streams Memory Management (RocksDB)

• RocksDB Releases

Upgrade your Kafka Streams applications to Confluent Platform 8.4.0 (Kafka Streams version 4.4.0)

To use Confluent Platform 8.4.0, update the Kafka Streams version your application depends on to use the version number 4.4.0. You may need to make minor code changes, detailed below, and recompile your application.

For example, in your pom.xml file:

 <dependency>
     <groupId>org.apache.kafka</groupId>
     <artifactId>kafka-streams</artifactId>
     <!-- update version to 4.4.0 -->
     <version>4.4.0</version>
 </dependency>

Streams API changes in Confluent Platform 8.4.x (Kafka Streams version 4.4.x)

The following sections describe the Kafka Streams API changes in Confluent Platform 8.4.x.

KIP-1321: Headers-aware StreamPartitioner

StreamPartitioner has a new partitions(topic, key, value, headers, numPartitions) method that receives the record headers. It’s a default method that delegates to the existing partitions(topic, key, value, numPartitions) method, which is deprecated and planned for removal in Kafka 5.0.

The built-in partitioners now pass the actual record headers to the key serializer, instead of empty headers.

Important

For windowed keys that you write without a custom StreamPartitioner, a key serializer whose output depends on record headers can produce different partitions than before the upgrade. This breaks co-partitioning with data written before the upgrade. Before you upgrade, check whether your key serializers read record headers.

KafkaStreams#queryMetadataForKey also has new overloads that accept Headers. If your partitioner or key serializer uses headers, use these overloads, so that the returned metadata matches where the key is stored.

For more information, see Discover and access application instances and their local state stores and KIP-1321.

KIP-892: Transactional state stores

Kafka Streams now supports transactional state stores. To enable them, set the new enable.transactional.statestores configuration to true. The default is false.

When transactional state stores are enabled:

• Under exactly-once processing, Kafka Streams no longer wipes and rebuilds state stores when an error occurs.

• Under exactly-once or at-least-once processing, only the writes since the last successful commit are lost when an error occurs, and Kafka Streams replays them through the topology.

Two more configurations control transactional state stores:

• statestore.uncommitted.max.bytes sets the maximum memory for buffering uncommitted state store records. The default is 64 MiB (67108864 bytes). When the buffer exceeds this limit, Kafka Streams requests a task commit.

• default.interactive.query.isolation.level sets the default isolation level for interactive queries. The default is READ_UNCOMMITTED. With READ_COMMITTED, writes become visible to interactive queries only after they’re committed.

For more information, see enable.transactional.statestores and KIP-892.

KIP-1284: DEFAULT group membership operation on close

CloseOptions.GroupMembershipOperation adds a DEFAULT value, which is now the default for CloseOptions and for the KafkaStreams#close() and KafkaStreams#close(Duration) methods. DEFAULT adapts to the group protocol and membership type:

• With group.protocol=classic, the consumer remains in the group on close. This matches the existing behavior.

• With group.protocol=streams, dynamic members leave the group on close. Static members, which set group.instance.id, remain in the group until the session timeout expires.

LEAVE_GROUP and REMAIN_IN_GROUP force the corresponding behavior regardless of the protocol. LEAVE_GROUP always leaves the group, and REMAIN_IN_GROUP keeps the member in the group without sending a leave-group request. The limitation introduced in Confluent Platform 7.3.0 (Kafka Streams 3.3.0), in which CloseOptions worked only for static members, no longer applies.

If your application sets REMAIN_IN_GROUP explicitly and uses group.protocol=streams, review your close logic, because the option is now honored. This also applies if your application calls the deprecated KafkaStreams#close(KafkaStreams.CloseOptions) method without leaveGroup(true). That method maps the default leaveGroup(false) to REMAIN_IN_GROUP, so dynamic members no longer leave the group on close, and the rebalance waits for the session timeout.

For an example, see closing an application instance. For more information, see KIP-1284.

Static membership with the Streams Rebalance Protocol

You can now use static membership with the Streams Rebalance Protocol. Applications that use group.protocol=streams can set group.instance.id.

Important

Static membership with group.protocol=streams requires both clients and brokers to run Confluent Platform 8.4.x (Kafka Streams 4.4.x) or later. Brokers on earlier versions reject group.instance.id on the streams group heartbeat with an INVALID_REQUEST error, which is fatal and shuts down the Kafka Streams client.

Before you downgrade brokers to a version earlier than Confluent Platform 8.4.x, remove group.instance.id from every Kafka Streams application that uses group.protocol=streams. If you’ve already downgraded the brokers, remove it now. Then restart the applications.

For topologies without persistent state stores, Kafka Streams generates a new process ID on each restart. The broker then recomputes the group assignment, which cancels the benefit of static membership across restarts.

To keep static members in the group across a restart, raise the group-level streams.session.timeout.ms configuration. For more information, see group.instance.id.

Rack-aware standby task assignment with the Streams Rebalance Protocol

For applications that use group.protocol=streams, you now configure the client tag keys for rack-aware standby task assignment on the broker, as part of the Streams Rebalance Protocol:

• The broker configuration group.streams.rack.aware.assignment.tags sets the default for all streams groups.

• The group configuration streams.rack.aware.assignment.tags overrides the default for one group. For example:

  ${CONFLUENT_HOME}/bin/kafka-configs --bootstrap-server localhost:9092 \
    --alter --entity-type groups --entity-name my-group \
    --add-config streams.rack.aware.assignment.tags=[zone,cluster]
  

Under this protocol, Kafka Streams ignores the client-side rack.aware.assignment.tags configuration, but each instance still needs a client.tag.<key> setting for every listed key. When the tags are set, the broker returns the MISSING_CLIENT_TAGS status to members that lack any of them, and Kafka Streams logs it as a warning. Clients earlier than Confluent Platform 8.4.x (Kafka Streams 4.4) don’t receive this status.

The built-in sticky assignor doesn’t use these tags.

For more information, see KIP-708 and KIP-1071.

KIP-1340: Stream key and mapped key in KStream-GlobalKTable joins

KStream-GlobalKTable joins can now access both the KStream record key and the mapped key that your KeyValueMapper computes. You don’t need to recompute the mapped key inside the joiner. The new ValueJoinerWithStreamAndMappedKey functional interface receives the stream key, the mapped key, the stream value, and the table value, in that order.

Four new KStream#join and KStream#leftJoin overloads accept this joiner. The following example uses the new join overload:

KStream<String, EnrichedOrder> enriched = orders.join(
    customers,
    (orderId, order) -> order.customerId(),
    (orderId, customerId, order, customer) ->
        new EnrichedOrder(orderId, customerId, order, customer));

Starting with Confluent Platform 8.4.x (Kafka Streams 4.4.x), the four existing ValueJoinerWithKey-based overloads for KStream-GlobalKTable joins are deprecated, and may be removed in a future release. The recommended alternative is to use the new ValueJoinerWithStreamAndMappedKey-based overloads. The deprecated overloads keep their runtime behavior, and the readOnlyKey parameter remains the KStream record key, so existing applications compile and run without modification.

For more information, see KStream-GlobalKTable joins and KIP-1340.

KIP-1356: Interactive Queries v2 for headers-aware state stores

Four new Interactive Queries v2 (IQv2) query types return record headers from the headers-aware state stores introduced in Confluent Platform 8.3.x (Kafka Streams 4.3.x):

• TimestampedKeyWithHeadersQuery

• TimestampedRangeWithHeadersQuery

• TimestampedWindowKeyWithHeadersQuery

• TimestampedWindowRangeWithHeadersQuery

Results carry headers as a ReadOnlyRecord. The range and window query types return a closeable ReadOnlyRecordIterator. The @Evolving annotation marks these query types, so their API might change in future releases.

For more information, see Query header-aware state stores with IQv2 and KIP-1356.

Headers-aware suppress() buffer

The buffer that the suppress() operator uses is now headers-aware, which removes one of the KIP-1285 limitations from Confluent Platform 8.3.x.

• With dsl.store.format set to HEADERS, Kafka Streams stores each buffered value with the headers of the record it came from, so record headers survive the suppression boundary. When Kafka Streams evicts a buffered row, the emitted record carries the headers of the emitted value.

• With the default dsl.store.format=DEFAULT, behavior is unchanged: if a later record updates a buffered row before eviction, the buffer keeps only the headers of the later record.

The buffer’s changelog records keep the existing value format in both cases. With HEADERS, Kafka Streams carries the headers of the buffered old and prior values in additional Kafka record headers, vh.old and vh.prior. Earlier Kafka Streams versions ignore these headers, so the changelog remains readable after a downgrade.

KIP-1245: Earlier validation of application.server

Kafka Streams now validates the application.server configuration when you create a StreamsConfig instance. The value must be "" (empty string) or a valid endpoint from which Kafka Streams can parse both the host and port, such as host:port or protocol://host:port. Invalid values that previously failed later, during startup or assignment, now fail immediately with a ConfigException.

For more information, see KIP-1245.

KIP-1283: KafkaStreams#cleanUp() preserves process metadata

KafkaStreams#cleanUp() no longer logs a WARN message when it can’t delete the application state directory only because expected metadata files remain, such as kafka-streams-process-metadata or .lock. In this case, Kafka Streams treats the local state cleanup as successful, and the application state directory might remain.

If you need a full local reset, including the persisted process metadata, close the KafkaStreams instance. Then delete the application state directory manually.

For more information, see KIP-1283.

Subtopology grouping for shared StoreBuilder instances

Kafka Streams now groups processors that return the same StoreBuilder instance from ConnectedStoreProvider#stores() into one subtopology. For affected applications, task IDs change, and Kafka Streams restores state from the changelog on the first startup after the upgrade. Plan for the restore time when you upgrade. Processors that connect to a store by name aren’t affected.

Confluent Platform 8.3.3 and Kafka Streams 4.3.2 also include the subtopology grouping change. If your application already runs one of these versions or later, the regrouping doesn’t occur when you upgrade to Confluent Platform 8.4.x.

For more information, see KAFKA-20464.

Asynchronous bootstrap DNS resolution isn’t supported

Kafka Streams doesn’t support the asynchronous mode for bootstrap DNS resolution introduced by KIP-909. Kafka Streams sets bootstrap.resolve.timeout.ms to 0 for every client that it creates internally. If you set a positive value, including with a client prefix such as consumer. or producer., Kafka Streams ignores it and logs a warning.

Streams API changes in Confluent Platform 8.3.x (Kafka Streams version 4.3.x)

The following sections describe the Kafka Streams API changes in Confluent Platform 8.3.x.

Important

Confluent Platform 8.3.0 (Kafka Streams version 4.3.0) contains a critical native memory leak in the RocksDB state store layer (KAFKA-20616 and KAFKA-20688). Under repeated state store task closes, such as rebalances or error-triggered recoveries, off-heap memory grows unbounded and can cause out-of-memory failures. This leak is fixed in Confluent Platform 8.3.2. As an alternative, the fix is available now in Kafka 4.3.1.

In Confluent Platform 8.3.3, Kafka Streams groups processors that share a StoreBuilder instance into one subtopology. For affected applications, upgrading from an earlier Confluent Platform 8.3.x release changes task IDs, and Kafka Streams restores state from the changelog on the first startup. For more information, see Subtopology grouping for shared StoreBuilder instances.

KIP-1259: Purge stale local state on startup

Kafka Streams now enables purging local state directories and checkpoint files during application startup if they have not been modified for a configurable period. This addresses scenarios where stale local state persists after changelog tombstones expire during broker-side retention (“zombie data”).

Configure the new state.cleanup.dir.max.age.ms setting to specify the maximum age threshold. State directories older than this threshold are deleted on startup, triggering a full rebuild from the changelog topic. The default value is -1 (disabled).

For more information, see KIP-1259.

KIP-1270: ProcessingExceptionHandler for global stores and GlobalKTables

Kafka Streams now supports ProcessingExceptionHandler for global store and GlobalKTable processing. Previously, the ProcessingExceptionHandler applied only to regular stream tasks. With this release, you can configure exception handling for global stores and GlobalKTables by setting the new configuration processing.exception.handler.global.enabled to true. For configuration details, refer to processing.exception.handler.global.enabled.

When enabled, the configured ProcessingExceptionHandler is invoked for exceptions occurring during global store and GlobalKTable processing.

Note

Dead Letter Queue (DLQ) support is not available for global stores and GlobalKTables. For global store and GlobalKTable exceptions, the record metadata does not include a DLQ topic.

For more information, see KIP-1270.

KIP-1285: Headers-aware state stores for DSL operators

Kafka Streams now supports opt-in headers-aware state stores for DSL operators. A new global config dsl.store.format (default: DEFAULT) controls the state store format used by all DSL operators. Setting it to HEADERS enables headers-aware stores that persist record headers alongside the value and timestamp in state stores. This builds on the headers-aware state store implementations introduced by KIP-1271. Headers-aware stores enable new Schema Registry onboarding without modifying existing data. They support the Schema Registry schema GUID in record header format, which is required to store schema identifiers in record headers.

// Enable headers-aware stores globally for all DSL operators
Properties props = new Properties();
props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, "HEADERS");

Per-operator customization of the store format is possible by providing a custom DslStoreSuppliers implementation through Materialized.withStoreType() or by supplying explicit store suppliers that return headers-aware stores.

The boolean isTimestamped constructors and isTimestamped() methods on the DslKeyValueParams, DslWindowParams, and DslSessionParams classes are deprecated in favor of the new DslStoreFormat-based constructors and storeFormat() accessors. The default behavior is unchanged, and existing applications are not affected.

For configuration details, refer to dsl.store.format. For more information, see KIP-1285.

Limitations in Confluent Platform 8.3.x

The following operators and store types do not support dsl.store.format=HEADERS. When these are used, the HEADERS setting is silently ignored and no headers are stored:

• The suppress() operator

• Left and outer stream-stream joins

• Versioned state stores

Starting in Confluent Platform 8.4.x, the suppress() buffer is headers-aware. For more information, see Headers-aware suppress() buffer.

Changelog format and restore

The changelog topic record format is not affected by the headers-aware store format. When headers-aware stores write to the changelog, the value body remains the raw value bytes only, the same as with DEFAULT stores. The timestamp and headers are carried as native Kafka record metadata (record timestamp and record headers), not embedded in the value payload. Changelog topics remain compatible regardless of the store format in use.

During state restoration from the changelog:

• Old changelog records (written before HEADERS was enabled) are restored into headers-aware stores with empty headers. The restore converter reconstructs the internal store format by reading the timestamp and headers from the Kafka record metadata.

• New changelog records (written with HEADERS enabled, carrying native Kafka record headers) can also be restored into DEFAULT stores. In that case, the headers are silently dropped and only the timestamp is prepended to the value.

No changelog topic recreation or reformatting is required when switching store formats.

Migration from DEFAULT to HEADERS

Switching dsl.store.format from DEFAULT to HEADERS does not require application downtime or manual intervention. The migration of local RocksDB state is handled automatically and incrementally:

1. On startup, each persistent state store opens with two RocksDB column families: the legacy column family (containing existing data in the old format) and a new headers-aware column family.

2. All new writes go directly to the headers-aware column family.

3. Reads check the headers-aware column family first. If the key is not found there, the store reads from the legacy column family, converts the value by prepending an empty headers prefix, writes the converted entry to the headers-aware column family, and deletes it from the legacy column family. This is a lazy, per-key migration that happens transparently during normal processing.

4. After all legacy data has been accessed, the store operates entirely from the headers-aware column family with no migration overhead.

Reverting from HEADERS back to DEFAULT requires clearing local state directories and restoring from changelog topics, because DEFAULT stores cannot read the headers-prefixed value format written by HEADERS stores.

Performance considerations

Enabling HEADERS stores introduces the following overhead compared to DEFAULT:

• Local state store size: Each RocksDB entry grows by at least 1 byte (the varint-encoded empty headers size prefix 0x00). With non-empty headers, the growth equals the varint-encoded headers size plus the serialized headers bytes. For applications that do not propagate headers into state stores, this overhead is minimal.

• Changelog topic size: The changelog record value body is unchanged. However, each changelog record carries headers as native Kafka record headers, which increases the per-record size on the wire and on disk proportional to the number and size of headers.

• Restore cost: During state restoration, the restore converter performs additional allocation and byte copying to reconstruct the internal [headersSize][headersBytes][timestamp][value] format from the raw changelog record and its metadata.

• Migration cost: During the transition period (while legacy data still exists in the old column family), each read of a legacy key involves a format conversion and a cross-column-family write. This cost is incurred only once per key and diminishes as migration completes.

For workloads where headers are small or empty, the overall impact is negligible. Applications that attach large or numerous headers to records should account for the proportional increase in state store and changelog sizes.

Upgrade Kafka Streams applications to Schema Registry schema ID in header format

Schema Registry supports two wire formats for identifying the schema of a record. In the prefix format, the schema ID is embedded in the serialized payload: a magic byte followed by the schema ID, prepended to the serialized key or value bytes. In the newer schema GUID in record header format, the schema identifier is stored in a Kafka record header instead of in the payload.

If your Confluent Platform Kafka Streams application already uses Schema Registry and you want to switch its SerDes from the prefix format to the header format, read the following guidance carefully. Upgrading values is a safe, rolling operation, but upgrading keys is not supported and breaks the application.

The header format also lets an application that does not use Schema Registry today, for example one that serializes plain Avro, JSON, or Protobuf records without Schema Registry, adopt Schema Registry without changing its existing serialized bytes, because the schema identifier travels in the record header instead of the payload. The prefix format cannot do this, because prepending the magic byte and schema ID to the payload changes the serialized bytes and breaks existing consumers of that data. The SerDes and state store guidance in this section applies whether your application already uses Schema Registry or is adopting it for the first time.

Important

The schema ID in record header format for Kafka Streams requires headers-aware state stores. Before you switch any SerDes to the header format, enable dsl.store.format=HEADERS and complete the DEFAULT to HEADERS migration described in dsl.store.format and in the Migration from DEFAULT to HEADERS section.

Headers-aware state stores have limitations: left and outer stream-stream joins and versioned state stores silently ignore dsl.store.format=HEADERS and store no headers. If your topology uses any of these, values that pass through that operator or store lose their schema GUID header, and deserialization fails downstream. In Confluent Platform 8.3.x, the suppress() operator also drops headers. Starting in Confluent Platform 8.4.x, the suppress() buffer is headers-aware, so with dsl.store.format=HEADERS, a record that the buffer emits keeps its schema GUID header. For more information, see Headers-aware suppress() buffer.

Upgrade value SerDes

Switching value SerDes from the prefix format to the schema ID in record header format is a safe, rolling upgrade. Enable the header format by setting the value.schema.id.serializer Schema Registry SerDes config to io.confluent.kafka.serializers.schema.id.HeaderSchemaIdSerializer on your value SerDe. The default value deserializer (DualSchemaIdDeserializer) reads both the header and the prefix formats, so no deserializer change is needed. Existing records are not rewritten:

• The value bytes of prefix-format records stay exactly the same. The schema ID remains embedded in the value payload, so no reprocessing or republishing of existing data is required.

• Records written after the upgrade carry the schema GUID in the Kafka record header instead of in the value payload.

• Deserializers handle both formats transparently. A record whose schema GUID is in the header is resolved from the header, and a record without a header is resolved from the embedded prefix schema ID.

When prefix-format value records are read into headers-aware state stores, an empty headers entry is prepended to the stored representation, producing the same [headersSize][headersBytes][timestamp][value] format described previously, with headersSize set to 0 and no headers bytes. The value, including the embedded schema ID, is unchanged.

Upgrade key SerDes

Do not switch key SerDes from the prefix format to the schema ID in record header format. This is not supported and breaks the application.

Kafka Streams computes the partition for a record from the serialized key bytes. Moving the schema ID out of the key payload changes those bytes, which changes the computed partition. Existing keys would then hash to different partitions than the records already stored in changelog, repartition, and output topics, breaking co-partitioning, joins, aggregations, and state store lookups. Because this cannot be done without a full reprocessing of all data, upgrading keys to the header format is not supported. Keep keys in the prefix format.

This is a stricter, Kafka Streams-specific case of the general Limitations of schema GUID in header. That caution applies to any Schema Registry client that migrates a record key from prefix format to header format.

Upgrade DSL applications

The following steps apply to Kafka Streams DSL applications, which create and manage state stores for you automatically. If your application uses the Processor API instead, see Upgrade Processor API applications.

1. Enable headers-aware state stores by setting dsl.store.format=HEADERS and complete the DEFAULT to HEADERS migration described in Migration from DEFAULT to HEADERS.

2. Reconfigure your value SerDes to use the schema ID in record header format by setting the value.schema.id.serializer SerDes config to HeaderSchemaIdSerializer. Keep the key SerDes on the prefix format.

3. Deploy the updated application with a rolling restart. No downtime, reprocessing, or republishing of existing data is required.

Upgrade Processor API applications

The SerDes changes described in the preceding sections apply the same way to Processor API applications: switch value.schema.id.serializer to HeaderSchemaIdSerializer, rely on the default DualSchemaIdDeserializer to read both formats, and keep keys on the prefix format. If your application builds its own state stores with the Processor API instead of relying on the Upgrade DSL applications, you must enable headers-aware stores explicitly: the dsl.store.format=HEADERS setting only affects stores that DSL operators create, and it has no effect on stores you build with a StoreBuilder.

To hold header-format values in a Processor API state store, build a headers-aware store using the org.apache.kafka.streams.state.Stores factory, with one of these builder and supplier pairs:

• Stores.timestampedKeyValueStoreWithHeadersBuilder(...) with Stores.persistentTimestampedKeyValueStoreWithHeaders(...)

• Stores.timestampedWindowStoreWithHeadersBuilder(...) with Stores.persistentTimestampedWindowStoreWithHeaders(...)

• Stores.sessionStoreWithHeadersBuilder(...) with Stores.persistentSessionStoreWithHeaders(...)

Key-value, window, and session stores each support a direct upgrade from their existing plain store to the headers-aware variant: switch the StoreBuilder and Kafka Streams reads existing on-disk data in the old format while writing new data in the headers-aware format, without a full reprocessing of the changelog. Versioned stores have no headers-aware variant and are not covered by this upgrade path. State stores whose values are not serialized with the header format do not need to change.

KIP-1244: Deprecation of streams-scala module

The kafka-streams-scala module (org.apache.kafka.streams.scala package) is deprecated in this release and might be removed in a future release. Both the Scala 2.12 and Scala 2.13 artifacts (kafka-streams-scala_2.12 and kafka-streams-scala_2.13) are affected.

The module remains fully functional in Confluent Platform 8.3. Existing applications compile and run without changes but emit deprecation warnings at compile time.

Migration pattern

Replace imports from the Scala wrapper package with the corresponding Java DSL types from org.apache.kafka.streams.kstream. The Scala wrappers add implicit Serdes, function-to-Single Abstract Method (SAM) conversions, and operator overloads, so you typically need to:

1. Switch the dependency from kafka-streams-scala_<version> to kafka-streams (the Java DSL).

2. Change import org.apache.kafka.streams.scala._ and related package imports to import org.apache.kafka.streams.kstream._.

3. Pass Serde instances explicitly by using Consumed.with(), Produced.with(), Grouped.with(), and Materialized.with() rather than relying on the implicit Serdes object.

4. Replace SAM-converted Scala lambdas with explicit KeyValueMapper, ValueMapper, Predicate, and similar functional interfaces, or rely on Scala’s native SAM support (Scala 2.12+).

The Java DSL reference and the Kafka Streams tutorial cover the equivalent Java DSL patterns.

For more information, see KIP-1244.

KIP-1035: Self-managed state store changelog offsets

Kafka Streams now persists state store changelog offsets inside each state store rather than in a single per-task .checkpoint file. For built-in RocksDB stores, offsets are written into a dedicated offsets column family inside each RocksDB instance. This is an internal infrastructure change and is transparent to most users. Existing per-task .checkpoint files are migrated automatically on first startup, and no application or operator action is required. Exactly-once semantics (EOS) crash behavior is unchanged in this release: state stores are still wiped and fully restored from the changelog.

Keeping each store’s changelog offsets together with its data, rather than in a separate .checkpoint file, improves the atomicity of offset and state updates.

Authors of custom StateStore implementations can opt in to managing their own offsets by implementing managesOffsets(), commit(Map<TopicPartition, Long>), and committedOffset(TopicPartition). Refer to KIP-1035 for the API details.

Warning

Downgrading from Kafka Streams 4.3.x to 4.2.x or older is not supported in-flight. Older versions do not declare the new offsets column family when opening RocksDB and fail to open the store on startup. To downgrade, stop each application instance, delete the local state directory (state.dir), and then start the older version. Kafka Streams restores all state stores from their changelog topics using the older on-disk format.

Streams API changes in Confluent Platform 8.2.x (Kafka Streams version 4.2.x)

KIP-1071: Streams Rebalance Protocol

The Streams Rebalance Protocol, introduced as Early Access in Confluent Platform 8.1.x (Kafka Streams 4.1.x), is now generally available.

The Streams Rebalance Protocol is a broker-driven rebalancing system designed specifically for Kafka Streams applications. Following the pattern of KIP-848, which moved rebalance coordination of plain consumers from clients to brokers, KIP-1071 extends this model to Kafka Streams workloads. Instead of clients computing new assignments on the client during rebalance events involving all members of the group, assignments are computed continuously on the broker. Instead of using a consumer group, the Kafka Streams application registers as a Kafka Streams group with the broker, which manages and exposes all metadata required for coordination of the Kafka Streams application instances.

Warning

In Confluent Platform 8.2.0 (Kafka Streams 4.2.0), a critical broker-side bug in the offline migration code (KAFKA-20254) prevented migrating existing groups from the classic protocol to the Streams Rebalance Protocol. This bug is fixed in Confluent Platform 8.3.0 (Kafka Streams 4.3.0). Newly created streams groups were not impacted.

Important

The Streams Rebalance Protocol is the preferred default option.

Use client-based rebalancing only if you need one of the features that are not included in this release.

• To start using the Streams Rebalance Protocol, see Enable the protocol.

• To migrate to the Streams Rebalance Protocol, see Migration from classic protocol.

For more information, see Streams Rebalance Protocol.

KIP-1034: Dead Letter Queue for production exceptions

Kafka Streams now supports Dead Letter Queue (DLQ). A new configuration, named errors.dead.letter.queue.topic.name, enables specifying the name of the DLQ topic. When this configuration is set and DefaultProductionExceptionHandler is used, records that cause exceptions are forwarded to the DLQ topic. If a custom exception handler is used, it is up to the custom handler to build DLQ records to send. Depending on the implementation, the errors.dead.letter.queue.topic.name configuration may be ignored.

• The org.apache.kafka.streams.errors.ProductionExceptionHandler$ProductionExceptionHandlerResponse class is deprecated and replaced with org.apache.kafka.streams.errors.ProductionExceptionHandler$Response class.

• Methods handle and handleSerializationException in org.apache.kafka.streams.errors.ProductionExceptionHandler are deprecated and replaced by handleError and handleSerializationError, respectively, in order to use the new Response class.

For more information, see KIP-1034.

KIP-1153: Refactor Kafka Streams CloseOptions to fluent API style

The org.apache.kafka.streams.CloseOptions class replaces the existing org.apache.kafka.streams.KafkaStreams$CloseOptions class. The latter is deprecated and is scheduled for removal in the next major release.

The CloseOptions class enables specifying a close timeout and group membership operation: whether the consumer needs to leave the group or remain in the group. For more information, see KIP-1153.

KIP-1230: Add config for file system permissions

Kafka Streams now enables state store directories created by Kafka Streams to have write access for the OS group, by using the newly added configuration allow.os.group.write.access. For more information, see KIP-1230.

KIP-1195: Deprecate BrokerNotFoundException

org.apache.kafka.streams.errors.BrokerNotFoundException is deprecated and is scheduled for removal in the next major release. For more information, see KIP-1195.

KIP-1216: Streams rebalance listener metrics

Kafka Streams now provides rebalance listener metrics for streams groups to monitor the latency of rebalance callbacks.

The following metrics are available at the thread level:

• tasks-assigned-latency-avg

• tasks-assigned-latency-max

• tasks-lost-latency-avg

• tasks-lost-latency-max

• tasks-revoked-latency-avg

• tasks-revoked-latency-max

These metrics are populated only when the Streams Rebalance Protocol (KIP-1071) is enabled. If you’re migrating from the consumer rebalance listener metrics, update your monitoring dashboards and alerts to use these new streams-specific metrics. For more information, see KIP-1216.

KIP-1221: Add application-id tag to client-state metric

The application-id tag is now available for the Kafka Streams client state metric (client-state). For more information, see KIP-1221.

Streams API changes in Confluent Platform 8.1.x (Kafka Streams version 4.1.x)

Important

Kafka Streams 4.1.0 contains a critical memory leak bug (KAFKA-19748) that affects users of range scans and certain DSL operators: session windows, sliding windows, stream-stream joins, and foreign-key joins. You should upgrade directly to version 4.1.1, which includes the fix.

Confluent Platform 8.1.0 already includes the fix for KAFKA-19668, the issue with KStream#processValues() described in the Confluent Platform 8.0.x section, below. Unlike the equivalent Kafka Streams 4.1.0 release, which requires 4.1.1 for the fix, you can upgrade to Confluent Platform 8.1.x with no extra action for this issue.

Early Access of the Streams Rebalance Protocol

This Early Access release covers a subset of the functionality detailed in KIP-1071. Do not use the Early Access protocol in production. The API is subject to change in future releases.

Included in Early Access

• Core Streams Group Rebalance Protocol: The group.protocol=streams configuration enables the dedicated streams rebalance protocol. This separates streams groups from consumer groups and provides a streams-specific group membership lifecycle and metadata management on the broker.

• Sticky Task Assignor: A basic task assignment strategy that minimizes task movement during rebalances is included.

• Interactive Query Support: IQ operations are compatible with the new streams protocol.

• New Admin RPC: StreamsGroupDescribe RPC provides streams-specific metadata separate from consumer group information, with corresponding access via the Admin client.

• Command-line tool integration: You can list, describe, and delete streams groups with the kafka-streams-groups script.

Not included in Early Access

• Static membership: Setting a client group.instance.id is rejected. Starting in Confluent Platform 8.4.x (Kafka Streams 4.4.x), you can use static membership with the Streams Rebalance Protocol. For more information, see Static membership with the Streams Rebalance Protocol.

• Topology updates: If a topology is changed significantly, for example, by adding new source topics or changing the number of sub-topologies, a new streams group must be created.

• High availability assignor: Only the sticky assignor is supported.

• Regular expressions: Pattern-based topic subscription is not supported.

• Reset operations: CLI offset reset operations are not supported.

• Protocol migration: Group migration is not available between the classic and new streams protocols.

KIP-1111: Enforce explicit naming for internal resources

The introduction of KIP-1111 enables you to enforce explicit naming for all internal resources of the topology, including internal topics, like changelog and repartition topics, and their associated state stores. This ensures that every internal resource is named before the Kafka Streams application is deployed, which is essential for upgrading your topology. You can enable this feature with StreamsConfig using the StreamsConfig#ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG parameter. When set to true, the application does not start if any internal resource has an auto-generated name.

KIP-1020: Move window.size.ms and windowed.inner.class.serde

The window.size.ms and windowed.inner.class.serde configurations are now defined in TimeWindowed and SessionWindowed SerDes. For more information, see KIP-1020.

Streams API changes in Confluent Platform 8.0.x (Kafka Streams version 4.0.x)

For a full list of API changes, see KAFKA-12822.

• All public API, deprecated in Confluent Platform 7.6.x (Kafka 3.6) or an earlier release, have been removed, with the exception of JoinWindows.of() and JoinWindows#grace(). For more information, see KAFKA-17531.

• Old protocol API versions have been removed. You should ensure brokers are at Confluent Platform version 5.1.x (Kafka Streams 2.1) or later before upgrading the Java clients to 4.0, including Connect and Kafka Streams that use the clients internally.

• Similarly, ensure that your Java clients, including Connect and Kafka Streams, are at Confluent Platform version 5.1.x (Kafka Streams 2.1) or later before upgrading the brokers to Confluent Platform 8.x (Kafka Streams 4.0).

• Care needs to be taken with Kafka clients that are not part of Apache Kafka®. For more information, see KIP-896.

• The minimum Java version required by clients and Kafka Streams applications has been increased from Java 8 to Java 11. For more information, see KIP-750

• Brokers, Connect, and tools now require Java 17. For more information, see KIP-1013

• In this release, eos-v1 (Exactly Once Semantics version 1) is no longer supported. To use eos-v2, brokers must be running Confluent Platform version 5.5.x (Kafka Streams 2.5) or later.

KAFKA-12822: Remove deprecated APIs of Kafka Streams in 4.0

All deprecated methods, classes, APIs, and config parameters up to and including Confluent Platform 7.6.x (Kafka Streams 3.6) have been removed.

The following list shows some of the important deprecated APIs. For the full list, see KAFKA-12822.

• builder methods for time/session/join/sliding windows

• kstream.KStream#branch in Java and Scala

• KafkaStreams#setUncaughtExceptionHandler()

• KStream#through() in Java and Scala

• Old processor APIs

• transformer methods and classes in Java and Scala

If you have the “merge repartition topics” optimization enabled, migrating from KStream#transformValues() to KStream#processValues() might not be safe, due to KAFKA-19668. Do not upgrade to Confluent Platform 8.0.0 in this case. Instead, upgrade directly to Confluent Platform 8.0.1, which includes the fix. For more information, see Transformers removal and migration to processors.

KIP-1056: Remove default. prefix for exception handler StreamsConfig

The configs default.deserialization.exception.handler and default.production.exception.handler are deprecated, because they don’t have any overwrites. Instead, use the new configs: deserialization.exception.handler and production.exception.handler. For more information, see KIP-1056.

KIP-1070: Deprecate MockProcessorContext

In the previous release, a new version of the Processor API was introduced, and the old Processor API was incrementally replaced and deprecated.

These APIs follow this path and are deprecated:

• MockProcessorContext

• Transformer

• TransformerSupplier

• ValueTransformer

• ValueTransformerSupplier

For more information, see KIP-1070.

KIP-1077: Deprecate ForeachProcessor and move to internal package

The ForeachProcessor class is deprecated. This change is aimed at improving the organization and clarity of the Kafka Streams API by ensuring that internal classes are not exposed in public packages. For more information, see KIP-1077.

KIP-1078: Remove leaking getter methods in Joined helper class

The leaking getter methods in the Joined helper class are deprecated. These methods are deprecated without a replacement for future removal, because they don’t add any value to Kafka Streams users. For more information, see KIP-1078.

KIP-1085: Fix leaking *_DOC variables in StreamsConfig

To ensure better encapsulation and organization of configuration documentation within Kafka Streams, certain public doc description variables that are only used within the StreamsConfig or TopologyConfig classes are deprecated. Also, the unused variable DUMMY_THREAD_INDEX is deprecated. For more information, see KIP-1085.

KIP-1087: Removing intermediateTopicsOption from StreamsResetter

Due to the removal of the already deprecated #through method in Kafka Streams, the intermediateTopicsOption of StreamsResetter tool in Kafka is not needed any more and is deprecated. For more information, see KIP-1087.

KIP-1091: Improved Kafka Streams operator metrics

Because string metrics can’t be collected on the broker side (KIP-714), this version introduces numeric counterparts to enable proper broker-side metric collection for Kafka Streams applications. These metrics are available at the INFO recording level, and a thread-level metric with a String value is available for users leveraging Java Management Extensions (JMX). For more information, see KIP-1091.

KIP-1104: Allow foreign key extraction from both key and value in KTable joins

To reduce storage overhead and improve API usability, a new method in the Java and Scala APIs that accepts a BiFunction for foreign key extraction is introduced, which enables foreign key extraction from both the key and value in KTable joins.

Previously, foreign key joins in KTables allowed only extraction from the value, which led to data duplication and potential inconsistencies.

This enhancement introduces a new method in the Java and Scala APIs that accepts a BiFunction for foreign key extraction, enabling more intuitive and efficient joins.

The existing methods are deprecated but not removed, ensuring backward compatibility. This change aims to reduce storage overhead and improve API usability.

For more information, see KIP-1104.

KIP-1106: Add duration based offset reset option for consumer clients

The Topology.AutoOffsetReset enum is deprecated and replaced with a new class, org.apache.kafka.streams.AutoOffsetReset to capture the reset strategies. New methods are added to the org.apache.kafka.streams.Topology and org.apache.kafka.streams.kstream.Consumed classes to support the new reset strategy. These changes aim to provide more flexibility and efficiency in managing offsets, especially in scenarios involving long-term storage and infinite retention. For more information, see KIP-1106.

KIP-1112: Allow custom processor wrapping

You can now configure your topology with a ProcessorWrapper, which enables accessing and optionally wrapping and replacing any processor in the topology by injecting an alternative ProcessorSupplier in its place. You can use this to peek records and access the processor context even for DSL operators, for example, to implement a logging or tracing framework, or to aid in testing or debugging scenarios.

You must implement the ProcessorWrapper interface and pass the class or class name into the configs via the new StreamsConfig#PROCESSOR_WRAPPER_CLASS_CONFIG config.

This config is applied during the topology building phase, and therefore doesn’t take effect unless the config is passed in when creating the StreamsBuilder (DSL) or Topology (PAPI) objects. You must use the StreamsBuilder or Topology constructor overload that accepts a TopologyConfig parameter for the StreamsConfig#PROCESSOR_WRAPPER_CLASS_CONFIG to be picked up.

For more information, see KIP-1112.

Upgraded RocksDB dependency

This version upgrades the RocksDB dependency to version 9.7.3, from 7.9.2. This upgrade incorporates various improvements and optimizations within RocksDB. However, it also introduces some API changes.

• The org.rocksdb.AccessHint class, along with its associated methods, has been removed.

• Several methods related to compressed block cache configuration in the BlockBasedTableConfig class have been removed, including blockCacheCompressedNumShardBits, blockCacheCompressedSize, and their corresponding setters. These functionalities are now consolidated under the cache option, and you should configure your compressed block cache by using the setCache method instead.

• The NO_FILE_CLOSES field has been removed from the org.rocksdb.TickerType enum. As a result, the number-open-files metric does not work as expected. Metric number-open-files returns constant -1 from now on, until it is officially removed.

• The org.rocksdb.Options.setLogger() method now accepts a LoggerInterface as a parameter, instead of the previous Logger.

Some data types used in RocksDB’s Java API have been modified. These changes, along with the removed class, field, and new methods, are primarily relevant to users implementing custom RocksDB configurations.

These changes are expected to be largely transparent to most Kafka Streams users. However, if you’re using advanced RocksDB customizations within your Streams applications, particularly through the rocksdb.config.setter, you are advised to consult the detailed RocksDB 9.7.3 changelog to ensure a smooth transition and adapt your configurations as needed. Specifically, if you’re leveraging the removed AccessHint class, the removed methods from the BlockBasedTableConfig class, the NO_FILE_CLOSES field from TickerType, or you’re relying on the previous signature of setLogger(), you must update your implementations.

Kafka Streams depends on a RocksDB version that requires MacOS 10.14 or higher.

Downgrading from Confluent Platform 8.0.x (Kafka Streams version 4.0.x) or later to a version earlier than Kafka Streams version 4.0.0 needs special attention: Since the 4.0.0 release, Kafka Streams upgraded RocksDB from version 7.9.2 to 9.7.3. This upgrade introduces a RocksDB file format version change from version 5 to version 6, introduced in RocksDB 8.6. While the newer RocksDB version 9.7.3 can read state stores written in the old version 5 format, the older RocksDB versions cannot read state stores written in the new version 6 format. Therefore, you cannot downgrade from Kafka Streams 4.0.x or newer to versions older than 4.0.0 unless you first delete the local RocksDB state stores written by Kafka Streams 4.0.x or newer. After deleting the local state stores, the older version restores them from the changelogs using the old file format.

KIP-714: Client metrics and observability

In this release, the ClientInstanceIds instance stores the global consumer Uuid for the KIP-714 id with a key of global stream-thread name appended with -global-consumer, where before it was only the global stream-thread name.

KIP-1065: Add “retry” return-option to ProductionExceptionHandler

Previously, the ProductionExceptionHandler was not invoked on a (retryable) TimeoutException. With Kafka Streams 4.0, the handler is called, and the default handler returns RETRY, to not change existing behavior. Now, a custom handler can decide to break the infinite retry loop by returning either CONTINUE or FAIL. For more information, see KIP-1065.

KIP-1076: Metrics for client applications KIP-714 extension

Kafka Streams metrics can be collected broker-side by using the KIP-714 broker plugin. This enables collecting the metrics of the internally used clients of a Kafka Streams application by using a broker-side plugin. Also, it enables collecting the metrics of the Kafka Streams runtime itself. For more information, see KIP-1076.

Streams API changes in Confluent Platform 7.9.0

KIP-1033: Improve exception handling

You can provide a processing exception handler to manage exceptions during the processing of a record, rather than throwing the exception all the way out of your Kafka Streams application. Provide the configs by using the StreamsConfig as StreamsConfig#PROCESSING_EXCEPTION_HANDLER_CLASS_CONFIG. The specified handler must implement the org.apache.kafka.streams.errors.ProcessingExceptionHandler interface. For more information, see KIP-1033

KIP-1049: Customize logging interval

Kafka Streams now enables you to customize the logging interval of stream-thread runtime summary, by using the newly added configuration, log.summary.interval.ms. By default, the summary is logged every 2 minutes. For more information, see KIP-1049.

Streams API changes in Confluent Platform 7.8.0

KIP-924: Customizable task assignment for Streams

Kafka Streams now supports customizable task assignment strategies via the task.assignor.class configuration. You can set the configuration to the fully qualified class name of a custom task assignor implementation that extends the new org.apache.kafka.streams.processor.assignment.TaskAssignor interface.

The new configuration also enables bringing back the behavior of the old task assignor, StickyTaskAssignor, that was used before the introduction of the HighAvailabilityTaskAssignor. If no custom task assignor is configured, the default task assignor, HighAvailabilityTaskAssignor, is used.

If you were using the internal.task.assignor.class config, you should switch to using the new task.assignor.class config instead, because the internal config will be removed in a future release. If you were previously plugging in the StickyTaskAssignor via the legacy internal.task.assignor.class config, you must ensure that you are importing the new org.apache.kafka.streams.processor.assignment.StickyTaskAssignor when you switch to the new task.assignor.class config, which is a version of the StickyTaskAssignor that implements the new public TaskAssignor interface. For more information, see the public interface section of KIP-924.

KIP-813: Shareable State Stores

The Processor API now supports so-called read-only state stores, added in KIP-813. These stores don’t have a dedicated changelog topic, but instead use their source topic for fault-tolerance, similar to KTables with source-topic optimization enabled.

KIP-989: Improved StateStore Iterator metrics for detecting leaks

To improve detection of leaked state store iterators, KIP-989 adds new store-level metrics to track the number and age of open iterators. The new metrics are num-open-iterators, iterator-duration-avg, iterator-duration-max, and oldest-iterator-open-since-ms. These metrics are available for all state stores, including RocksDB, in-memory, and custom

Streams API changes in Confluent Platform 7.7.0

KIP-925: Rack aware task assignment in Kafka Streams (Part 2)

In part one of KIP-925, the min_traffic assignment strategy for Kafka Streams was added. Part Two finishes the KIP by introducing the second rack-aware assignment strategy: balanced_subtopology.

KIP-954: Expand default DSL store configuration to custom types

KIP-954 builds on KIP-591 and enables you to provide a default state store provider for your custom stores. As part of this change, a new interface has been provided along with default support for RocksDB and in-memory state stores.

KIP-962: Relax non-null key requirement in Kafka Streams

Kafka Streams treated records with null-keys as invalid input for joins and dropped them. KIP-962 relaxes this behavior for various left-joins, allowing null-key records to be processed successfully.

The behavior of the following operators changed.

• left join KStream-KStream: no longer drop left records with null-key and call ValueJoiner with null for the right value.

• outer join KStream-KStream: no longer drop left/right records with null-key and call ValueJoiner with null for the right/left value.

• left-foreign-key join KTable-KTable: no longer drop left records with null-foreign-key returned by the ForeignKeyExtractor and call ValueJoiner with null for the right value.

• left join KStream-KTable: no longer drop left records with null-key and call ValueJoiner with null for the right value.

• left join KStream-GlobalTable: no longer drop records when KeyValueMapper returns null and call ValueJoiner with null for right value.

Stream-DSL users who want to keep the current behavior can prepend a .filter() operator to the previously listed operators and filter accordingly. The following snippets illustrate how to keep the pre-7.7.0 behavior.

//left join KStream-KStream
leftStream
.filter((key, value) -> key != null)
.leftJoin(rightStream, (leftValue, rightValue) -> join(leftValue, rightValue), windows);

//outer join KStream-KStream
rightStream
.filter((key, value) -> key != null);
leftStream
.filter((key, value) -> key != null)
.outerJoin(rightStream, (leftValue, rightValue) -> join(leftValue, rightValue), windows);

//left-foreign-key join KTable-KTable
Function<String, String> foreignKeyExtractor = leftValue -> ...
leftTable
.filter((key, value) -> foreignKeyExtractor.apply(value) != null)
.leftJoin(rightTable, foreignKeyExtractor, (leftValue, rightValue) -> join(leftValue, rightValue), Named.as("left-foreign-key-table-join"));

//left join KStream-KTable
leftStream
.filter((key, value) -> key != null)
.leftJoin(kTable, (k, leftValue, rightValue) -> join(leftValue, rightValue));

//left join KStream-GlobalTable
KeyValueMapper<String, String, String> keyValueMapper = (key, value) -> ...;
leftStream
.filter((key, value) -> keyValueMapper.apply(key,value) != null)
.leftJoin(globalTable, keyValueMapper, (leftValue, rightValue) -> join(leftValue, rightValue));

KIP-960 / KIP-968: IQ support for Versioned State Stores

Versioned state stores were added in the Kafka 3.5 release (KIP-889), but it was not possible to query the new stores. KIP-960 and KIP-968 close this gap by adding new query types for IQv2, namely VersionedKeyQuery and MultiVersionedKeyQuery, respectively. Both queries enable you to do lookups for a single key, to ask for the most recent value, a historic value, or a range of historic values for the provided key.

KIP-985: Add reverseRange and reverseAll query over kv-store in IQv2

IQv2 supports RangeQuery and enables you to query for a range of keys and specify unbounded, bounded, or half-open key-ranges. It returns data in ascending (byte[]-lexicographical) order (per partition). KIP-985 extends this functionality by adding the .withDescendingKeys() method to enable receiving data in descending order, so you can request the result to be ordered (per partition) in either ascending or descending order, or to leave the order unspecified.

KIP-988: Streams Standby Update Listener

KIP-988 adds a new interface for handling cases where standby tasks have their state stores registered, load a batch of records, and stop updates.

KIP-992: Introduce IQv2 Query Types: TimestampedKeyQuery and TimestampedRangeQuery

KIP-992 adds new timestamped-key and timestamped-range interactive queries for timestamped key-value state stores. This change improves the type safety of the IQv2 API. The existing RangeQuery now always returns only the value if issued against a timestamped key-value store.

default.dsl.store config deprecated

Instead, use the dsl.store.suppliers.class. If you currently specify default.dsl.store=ROCKS_DB or default.dsl.store=IN_MEMORY, replace these configurations with dsl.store.suppliers.class=BuiltInDslStoreSuppliers.RocksDBDslStoreSuppliers.class and dsl.stores.suppliers.class=BuiltInDslStoreSuppliers.InMemoryDslStoreSuppliers.class, respectively.

Streams API changes in Confluent Platform 7.6.0

KIP-923: Add A Grace Period to Stream Table Join

KIP-923 adds a grace period to stream-table joins to improve table-side out-of-order data handling. The joined object has a new method named withGracePeriod that causes the table-side lookup to happen only after the grace period has passed.

KIP-925: Rack aware task assignment in Kafka Streams

Rack aware task assignment was introduced in KIP-925. Rack aware task assignment can be enabled for StickyTaskAssignor or HighAvailabilityTaskAssignor to compute task assignments, which can minimize cross-rack traffic under certain conditions. For more information, including how it can be enabled and further configured, see rack.aware.assignment.strategy.

KIP-941: Range queries to accept null lower and upper bounds

Previously, RangeQuery did not support null to specify “no upper/lower bound”. KIP-941 allows users to pass null into withRange(...) for lower/upper bounds to specify a full or half-open range:

• withRange(null, null) == withNoBounds()

• withRange(lower, null) == withLowerBound(lower)

• withRange(null, upper) == withUpperBound(upper)

Streams API changes

Downgrading from Confluent Platform 7.5.x (Kafka Streams 3.5.x) or later to Confluent Platform 7.4.x or earlier requires special attention: starting in the 3.5.0 release, Kafka Streams uses a new serialization format for repartition topics. This means that older versions of Kafka Streams don’t recognize the bytes written by newer versions, so it’s harder to downgrade Kafka Streams with version 3.5.0 or later to older versions in-flight. For more information, see KIP-904.

For a downgrade, first switch the config from upgrade.from to the version you’re downgrading to. This disables writing the new serialization format in your application. It’s important to wait in this state long enough to ensure that the application has finished processing any “in-flight” messages written into the repartition topics in the new serialization format. Afterward, you can downgrade your application to a pre-3.5.x version.

KIP-399: Extend ProductionExceptionHandler to cover serialization exceptions

KIP-399 adds a method, handleSerializationException(), to the ProductionExceptionHandler interface to handle any serialization errors encountered while producing records.

KIP-884: Add config to configure KafkaClientSupplier

KIP-884 adds a new config, default.client.supplier that enables using a custom KafkaClientSupplier without any code changes.

KIP-889: Versioned state stores

KIP-889 introduces versioned state stores to improve the accuracy of joins when out-of-order records are processed. For more information, see Timestamp-based semantics for table processors.

In addition to KIP-889, KIP-914 updates DSL processing semantics if a user opts-in to use the new versioned key-value stores. Using the new versioned key-value stores, DSL processing can better handle out-of-order data. For example, a late record may be dropped and stream-table joins can do a timestamp-based lookup into the table. Table aggregations and primary/foreign-key table-table joins are also improved. Versioned key-value stores are not supported for global-KTable, and they don’t work with suppress().

KIP-904: Guarantee subtractor is called before adder if key has not changed

KIP-904 improves the implementation of KTable aggregations. In general, an input KTable update triggers a result refinement for two rows, but prior to KIP-904, if both refinements happened to the same result row, two independent updates to the same row are applied, resulting in spurious intermediate results. KIP-904 enables detecting this case and applies only a single update, avoiding spurious intermediate results.

KIP-907: Add Boolean serde to public interface

Kafka Streams includes built-in Serdes for most primitive types. KIP-907 adds a new one for booleans.

Streams API changes in Confluent Platform 7.4.0

KIP-770: Replace cache.max.bytes.buffering with statestore.cache.max.bytes

KIP-770 deprecates the existing cache.max.bytes.buffering config and introduces a new statestore.cache.max.bytes config to replace it. The semantics and default value of the cache size config is unchanged. This KIP also adds a new cache.size metric at the DEBUG level for users to monitor the actual size of the Kafka Streams cache.

KIP-837: Allow MultiCasting a Result Record

KIP-837 enables you to multicast result records to every partition of downstream sink topics and adds functionality for choosing to drop result records without sending.

KIP-865: Support “–bootstrap-server” in kafka-streams-application-reset

KIP-865 updates the Kafka Streams application reset tool’s server parameter name to conform to the other Kafka tooling by deprecating the --bootstrap-servers parameter and introducing a new --bootstrap-server parameter in its place.

Streams API changes in Confluent Platform 7.3.0

Source/sink node metrics for consumed/produced throughput in Kafka Streams

Source and sink node metrics for consumed and produced throughput are available in Kafka Streams.

Previously, with the metrics available in the plain consumer you could derive the consumed throughput of your applications at the subtopology level, but the same was not true for the produced throughput.

KIP-846 fills this gap and gives you a way to compute the production rate of each subtopology by introducing two new metrics for the throughput at sink nodes. Even though it’s possible to derive the consumed throughput with existing client-level metrics, KIP-846 also adds two new metrics for the throughput at source nodes, to provide an equally fine-grained metrics scope as for the newly added metrics at the sink nodes, and to simplify the user experience.

Pause/resume KafkaStreams topologies

KIP-834 adds the ability to pause and resume topologies. You can use this feature to reduce resources used or modify data pipelines. Paused topologies skip processing, punctuation, and standby tasks. For distributed Kafka Streams applications, each instance must be paused and resumed separately.

Consolidate KStream transform() and process() methods

KIP-820 generalizes the Kafka Streams API to consolidate Transformers, which could forward results, and Processors, which could not. The change makes use of the new type-safe Processor API, which simplifies Kafka Streams, making it easier to use and learn.

New KafkaStreams.close() API

KIP-812 introduces another form of the KafkaStreams.close() API that forces the member to leave the consumer group. This new method efficiently closes the stream permanently by forcing the member to leave the consumer group.

Streams API changes in Confluent Platform 7.2.0

Rack awareness for Kafka Streams

Starting with Confluent Platform 7.2.0, Kafka Streams can distribute its standby replicas over distinct “racks” with KIP-708. To form a “rack”, Kafka Streams uses tags in the application configuration. For example, Kafka Streams clients might be tagged with the cluster or the cloud region they are running in. Users can specify the tags that should be used for the rack-aware distribution of the standby replicas by setting the rack.aware.assignment.tags configuration. During task assignment, Kafka Streams tries to distribute the standby replicas over different task dimensions. Rack-aware standby assignment improves fault tolerance in case of the failure of an entire “rack”. This can be used, for example, to ensure that replicas are distributed over different availability zones in a cloud hosting provider.

Add record metadata to state store context

KIP-791 adds the recordMetadata() method to the StateStoreContext, providing access to the topic, partition, and offset of the record currently being processed. Exposing the current context in this way enables state stores to track their current offset in each input partition, allowing them to implement the consistency mechanisms introduced in KIP-796.

Interactive Query v2 preview

Confluent Platform 7.2.0 introduces Interactive Queries v2 in Kafka Streams (IQv2). IQv2 is a preview feature, and the interfaces of IQv2 are marked as @Evolving, which means that they may break compatibility in minor releases without a deprecation period if preview users find significant flaws in the current API.

• KIP-796 specifies an improved interface for Interactive Queries in Kafka Streams (IQv2). The new interface makes querying the state store simpler and faster and reduces the maintenance cost when modifying existing state stores and adding new state stores. KIP-796 describes the generic interface for querying state stores with Interactive Queries. Specific query types can be added to Interactive Query v2 by implementing the Query interface. KIP-976 also defines the KeyQuery class to enable users to evaluate a key/value lookup by using IQv2.

• KIP-805 adds the RangeQuery class to IQv2. The RangeQuery class is an implementation of the Query interface that enables querying state stores over a range specified by upper or lower key bounds or by scanning all records of a state store when no bounds are provided.

• KIP-806 adds two implementations of the Query interface.

  • The WindowKeyQuery class enables scanning over windows with a given key within a specified time range.

  • The WindowRangeQuery class enables scanning over windows within a given time range independently of the windows’ keys.

Streams API changes in Confluent Platform 7.1.0

Java 17 support

In Confluent Platform 7.1.0, Kafka Streams supports Java 17.

Improved left/outer stream-stream join semantics

The semantics of left/outer stream-stream join were improved by KIP-633.

Previously, a left-/outer stream-stream join might have emitted so-called spurious left/outer results, due to an eager-emit strategy. The implementation was changed to emit left/outer join result records only after the join window is closed. The old API to specify the join window, JoinWindows.of(), that enables the eager-emit strategy, was deprecated in favor of the JoinWindows.ofTimeDifferenceAndGrace() and JoinWindows.ofTimeDifferenceWithNoGrace() methods. The new semantics are enabled only if you use the new join window builders.

Additionally, KIP-633 makes setting a grace period mandatory for windowed aggregations, that is, TimeWindows (hopping/tumbling), SessionWindows, and SlidingWindows. The corresponding builder methods .of(...) were deprecated in favor of the new .ofTimeDifferenceAndGrace() and .ofTimeDifferenceWithNoGrace() methods.

New metrics to track blocking times

KIP-761 adds new metrics that enable tracking blocking times on the underlying consumer and producer clients. For more information, see Kafka Streams metrics.

Interactive Query improvements

Interactive Queries were improved by KIP-763 and KIP-766. Range queries now accept null as a lower/upper key-range bound to indicate an open-ended lower/upper bound.

Custom partitioners for foreign-key table-table joins

Foreign-key table-table joins now support custom partitioners via KIP-775. Previously, if an input table was partitioned by a non-default partitioner, joining records might fail. With KIP-775, you now can pass a custom StreamPartitioner into the join using the newly added TableJoined object.

Note

This website includes content developed at the Apache Software Foundation under the terms of the Apache License v2.