# Kafka Streams Upgrade Guide for Confluent Cloud This guide explains which Kafka Streams versions are supported on Confluent Cloud, how to upgrade to 4.2.0, and which configuration changes and rolling-bounce steps are required when moving from earlier versions. To upgrade from Kafka Streams versions earlier than 3.1.x, see [Legacy Streams Upgrade Guide](https://docs.confluent.io/legacy/platform/7.0/streams/upgrade-guide.html). #### IMPORTANT Kafka Streams 4.3.0 contains a critical native memory leak in the RocksDB state store layer ([KAFKA-20616](https://issues.apache.org/jira/browse/KAFKA-20616) and [KAFKA-20688](https://issues.apache.org/jira/browse/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 4.3.1. If your applications use the 4.3.0 `kafka-streams` client, upgrade to 4.3.1. ## Kafka Streams and Confluent Cloud version support The following table lists Kafka Streams and Confluent Cloud version support. | Kafka Streams | Release Date | Business Level end of support | Premier Level end of support | |-----------------|-------------------|---------------------------------|--------------------------------| | 4.2.x | February 17, 2026 | February 17, 2028 | February 17, 2029 | | 4.1.x | October 15, 2025 | October 15, 2027 | October 15, 2028 | | 4.0.x | June 11, 2025 | June 11, 2027 | June 11, 2028 | | 3.9.x | February 19, 2025 | February 19, 2027 | February 19, 2028 | | 3.8.x | December 2, 2024 | December 2, 2026 | December 2, 2027 | | 3.7.x | July 26, 2024 | July 26, 2026 | July 26, 2027 | | 3.6.x | February 9, 2024 | February 9, 2026 | February 9, 2027 | | 3.5.x | August 25, 2023 | August 25, 2025 | August 25, 2026 | | 3.4.x | May 3, 2023 | May 3, 2025 | May 3, 2026 | | 3.3.x | November 4, 2022 | November 4, 2024 | November 4, 2025 | | 3.2.x | July 25, 2022 | July 25, 2024 | July 25, 2025 | ## Upgrade from older versions Upgrading from any earlier Kafka Streams version to 4.2.0 is supported. If you’re upgrading from Kafka Streams 3.4 or earlier, you must do two rolling bounces. 1. During the first rolling bounce, set the `upgrade.from=""` 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](https://cwiki.apache.org/confluence/x/P5VbDg). If you skip or delay the second rolling bounce, your deployment continues using the previous eager rebalancing protocol, but you can switch safely to cooperative rebalancing at any time, after the entire group is on Kafka Streams 3.4 or later, by removing the configuration value and bouncing. 1. Prepare your application instances for a rolling bounce, and ensure that the `upgrade.from` configuration is set to the version you’re upgrading from. 2. Bounce each application instance. 3. Prepare your newly deployed Kafka Streams 4.2.0 application instances for a second round of rolling bounces. Be sure to remove the value for the `upgrade.from` configuration. 4. Bounce each application instance a second time to complete the upgrade. As an alternative, an offline upgrade is also possible. Upgrading from any versions as old as Kafka Streams 0.10.x to 4.2.0 in offline mode requires the following steps: 1. Stop all old application instances, for example, Kafka Streams 0.10.x. 2. Update your code and swap old code and JAR files with new code and new JAR files. 3. Restart all new 4.2.0 application instances. ## Upgrade to Kafka Streams 4.2.0 from Kafka Streams 3.1.x or later ### RocksDB compatibility matrix The following table shows which RocksDB version ships with each Kafka Streams version. This matters when you use state stores backed by RocksDB, because mismatched versions can lead to runtime errors or data incompatibility. | Kafka Streams version | RocksDB version | Notes | |-------------------------|-------------------|----------------------------------------------------------------------------------------------------------------------------| | 4.3.x | 10.1.3 | New `offsets` column family (KIP-1035); refer to [KIP-1035 downgrade guidance](#streams-upgrade-guide-kip-1035-downgrade). | | 4.2.x | 10.1.3 | | | 4.1.x | 10.1.3 | | | 4.0.x | 9.7.3 | Significant API changes, see [Upgraded RocksDB dependency](#streams-upgrade-guide-rocksdb). | | 3.5.x – 3.9.x | 7.9.2 | Requires newer GCC; not compatible with RHEL 7. | | 3.0.x – 3.4.x | 7.1.2 | Downgrading from 3.0.x or newer to a 2.8.x or older version requires special attention because of a data format change. | | 2.6.x – 2.8.x | 5.18.4 | | | 2.3.x – 2.5.x | 5.18.3 | Can specify more RocksDB configurations, which helps to limit RocksDB off-heap memory usage. | | 2.2.x | 5.15.10 | | | 2.1.x | 5.14.2 | | | 2.0.x | 5.7.3 | | The exact RocksDB version is a transitive dependency of the Kafka Streams artifact, and Maven or Gradle manage it automatically, 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)](/platform/current/streams/developer-guide/memory-mgmt.html) - [RocksDB Releases](https://github.com/facebook/rocksdb/releases) ## Upgrade your Kafka Streams applications to version 4.2.0 To use version 4.2.0, update the Kafka Streams version your application depends on to use the version number 4.2.0. You might need to make minor code changes, detailed in the following sections, and recompile your application. For example, in your `pom.xml` file: ```xml org.apache.kafka kafka-streams 4.2.0 ``` ## Streams API changes in Kafka Streams version 4.3.x ### 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1259%3A+Add+configuration+to+wipe+Kafka+Streams+local+state+on+startup). ### 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`. 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1270%3A+Extend+ProcessExceptionalHandler+for+GlobalThread). ### 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1271%3A+Headers-Aware+State+Stores). Headers-aware stores enable new Schema Registry onboarding without modifying existing data. They support the Schema Registry [schema GUID in record header](../sr/fundamentals/serdes-develop/index.md#messages-wire-format-schema-id-in-header) format, which is required to store schema identifiers in record headers. ```java // Enable headers-aware stores globally for all DSL operators Properties props = new Properties(); props.put(StreamsConfig.DSL_STORE_FORMAT_CONFIG, "HEADERS"); ``` **Recommendation:** For new Kafka Streams applications, set `dsl.store.format=HEADERS`. This avoids a future migration if you later adopt record headers, such as the Schema Registry schema GUID in header format, or other features that depend on header-aware stores. The runtime overhead for workloads without headers is a single byte per state store entry. Existing applications can stay on `DEFAULT` and switch when needed. 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 more information, see [KIP-1285](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1285%3A+DSL+Opt-in+Support+for+Headers-Aware+State+Stores). #### Current limitations 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 #### Changelog format and restore The changelog topic record format is not affected by this change. 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: - 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. - Changelog records written with `HEADERS` enabled carry native Kafka record headers. These records can also be restored into `DEFAULT` stores, in which 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. #### KIP-1271 Processor API surface Applications that build state stores directly through the Processor API can use the header-aware interfaces and factory methods introduced by [KIP-1271](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1271%3A+Headers-Aware+State+Stores). New interfaces: - `TimestampedKeyValueStoreWithHeaders` - `TimestampedWindowStoreWithHeaders` - `SessionStoreWithHeaders` Each interface extends its non-header-aware, timestamped counterpart and adds `put` / `get` overloads that accept or return a `Headers` parameter. Records written through the non-header-aware methods are stored with empty headers. The plain (non-timestamped) key-value store and the plain window store have no headers-aware variant; use the timestamped variants instead. New factory methods on `org.apache.kafka.streams.state.Stores`: - `Stores.timestampedKeyValueStoreWithHeadersBuilder(...)` with `Stores.persistentTimestampedKeyValueStoreWithHeaders(...)` - `Stores.timestampedWindowStoreWithHeadersBuilder(...)` with `Stores.persistentTimestampedWindowStoreWithHeaders(...)` - `Stores.sessionStoreWithHeadersBuilder(...)` with `Stores.persistentSessionStoreWithHeaders(...)` For test access, `TopologyTestDriver` adds matching accessor methods: `getTimestampedKeyValueStoreWithHeaders(String)`, `getTimestampedWindowStoreWithHeaders(String)`, and `getSessionStoreWithHeaders(String)`, that return the header-aware view of a store materialized with `HEADERS` format. ### 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](../sr/fundamentals/serdes-develop/index.md#messages-wire-format-schema-id-in-header) format, the schema identifier is stored in an Kafka record header instead of in the payload. If your Confluent Cloud Kafka Streams application already uses Schema Registry and you want to switch its serialization and deserialization (SerDes) from the prefix format to the schema ID in record header format, read the following guidance carefully before making this change. 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 adopt Schema Registry without changing its existing serialized bytes. For example, an application that serializes plain Avro, JSON, or Protobuf records without Schema Registry can add schema identifiers in the record header without altering 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 the [Migration from DEFAULT to HEADERS](#streams-cc-upgrade-guide-migrate-default-headers) section. #### 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 configuration 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 ID is in the header is resolved from the header; a record without the 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. This produces the same `[headersSize][headersBytes][timestamp][value]` format described earlier, 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. #### 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](#streams-cc-upgrade-guide-sr-schema-id-header-papi). 1. Enable headers-aware state stores by setting `dsl.store.format=HEADERS`. 2. Complete the `DEFAULT` to `HEADERS` migration, as described in the preceding section. 3. Reconfigure your **value** SerDes to use the schema ID in record header format by setting the `value.schema.id.serializer` SerDes configuration to `HeaderSchemaIdSerializer`. 4. Keep the **key** SerDes on the prefix format. 5. 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. - 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](#streams-cc-upgrade-guide-sr-schema-id-header-dsl), 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 one of the key-value, window, or session `Stores` factory builder and supplier pairs described in [KIP-1271 Processor API surface](#streams-cc-upgrade-guide-kip-1271). 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 reprocessing 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 this release. Existing applications continue to 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_` 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+). For more information, see [KIP-1244](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1244%3A+Deprecate+streams-scala+module). ### 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. **Custom \`\`StateStore\`\` API additions** Authors of custom `StateStore` implementations can opt in to managing their own offsets by overriding three new default methods on `org.apache.kafka.streams.processor.StateStore`: - `boolean managesOffsets()`: Returns `true` if the store persists its consumer offset alongside its data. The default implementation returns `false`, preserving the legacy `.checkpoint` file behavior. This method is deprecated and might be removed in a future release. New implementations should return `true` and manage their own offsets. - `void commit(Map changelogOffsets)`: Atomically persists pending writes and the given changelog offsets. Replaces `flush()` for stores that manage their own offsets. - `Long committedOffset(TopicPartition partition)`: Returns the last committed changelog offset for the given partition, or `null` if no offset has been committed. A store that overrides `managesOffsets()` to return `true` must also override `commit()` and `committedOffset()` to maintain offset-data atomicity. The `flush()` method on `StateStore` is deprecated. Existing custom stores that do not override `managesOffsets()` continue to work unchanged. For more information, see [KIP-1035](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1035%3A+StateStore+Managed+Changelog+Offsets). #### WARNING Downgrading from Kafka Streams 4.3.x to 4.2.x or older is not supported as a rolling downgrade. 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 Kafka Streams version 4.2.x ### Streams Rebalance Protocol The Streams Rebalance Protocol is a broker-driven rebalancing system designed specifically for Kafka Streams applications. Following the pattern of [KIP-848](https://cwiki.apache.org/confluence/display/KAFKA/KIP-848%3A+Move+rebalance+coordination+of+plain+consumers+from+clients+to+brokers), which moved rebalance coordination of plain consumers from clients to brokers, [KIP-1071](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1071%3A+Streams+Rebalance+Protocol) extends this model to Kafka Streams workloads. Instead of clients computing new assignments during rebalance events that involve all members of the group, the broker computes assignments continuously. 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. #### IMPORTANT - Use the Streams Rebalance Protocol as the default option. - Use client-based rebalancing only if you need one of the features that are [not included in this release](#streams-upgrade-guide-streams-rebalance-protocol-not-included). - The Streams Rebalance Protocol is enabled on Dedicated, Enterprise, Standard, and Basic clusters. #### Streams Rebalance Protocol features - **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:** Provides a basic task assignment strategy that minimizes task movement during rebalances. - **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 through the `Admin` client. - **CLI Integration:** You can list, describe, and delete streams groups with the `kafka-streams-groups` script. #### Not included in the current release This release covers a subset of the functionality detailed in [KIP-1071](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1071%3A+Streams+Rebalance+Protocol). The following features are not included in this release: - **Static membership:** You can’t set a client `group.instance.id`. - **Topology updates:** Updates to existing topologies are not supported. If you change a topology significantly, for example, by adding new source topics or changing the number of sub-topologies, you must create a new streams group. - **High availability assignor:** Only the sticky assignor is supported. Warmup tasks and rack aware assignment are not supported in this release. - **Regular Expressions:** Pattern-based topic subscription is not supported. - **Online migration:** Group migration is available between the classic and new streams protocols, but only as an offline upgrade, which means that all instances must first be stopped, and the group must become empty before it can be restarted with the new protocol enabled. Rolling bounce online upgrade is not supported. #### NOTE [KAFKA-20254](https://issues.apache.org/jira/browse/KAFKA-20254) doesn’t affect Confluent Cloud. Confluent Cloud supports offline group migrations from “classic” to “streams” protocol, subject to the limitations described in this section. - **Custom Kafka clients:** The [KafkaClientSupplier interface](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/KafkaClientSupplier.html) is not fully supported. Using a custom `KafkaClientSupplier` can provide only restore/global consumer, producer, and admin client. It’s not possible to provide the “main” consumer when “streams” groups are enabled. #### Why use the Streams Rebalance Protocol? KIP-1071 delivers broker-driven rebalancing specifically optimized for Kafka Streams applications. This reduces coordination overhead and improves failure detection, leading to more stable and responsive stream processing workloads. - **Broker-driven coordination:** Centralizes task assignment logic on brokers instead of the client. This provides consistent, authoritative task assignment decisions from a single coordination point and reduces the potential for split-brain scenarios. - **Faster, more stable rebalances:** Reduces rebalance duration and impact by removing the global synchronization point. This helps minimize application downtime during membership changes or failures. - **Better observability:** Provides dedicated metrics and admin interfaces that separate streams from consumer groups, leading to clearer troubleshooting with broker-side observability. #### Enable the protocol Kafka Streams clients must be running Kafka 4.2 or later to use this protocol. - In your Kafka Streams application configuration, set `group.protocol=streams`. Online “rolling-bounce” migration between the classic consumer group protocol and the Streams Rebalance Protocol is not supported in either direction, but you can upgrade an empty group from “classic” to “streams” or downgrade from “streams” to “classic”. Also, the `application.id` can’t be in use as a `group.id` by any other consumer, either “classic” or “consumer”, or by any share-group application. You can delete a previous consumer group by using `kafka-consumer-groups` in `${CONFLUENT_HOME}/bin` before starting the application with the new protocol, but this also deletes all offsets for that group. To operate the new streams groups, you can explore the options of `kafka-streams-groups` to list, describe, and delete streams groups. In the new protocol, `session.timeout.ms`, `heartbeat.interval.ms`, `initial.rebalance.delay.ms`, and `num.standby.replicas` are *group-level* configurations. Setting `session.timeout.ms`, `heartbeat.interval.ms`, or `num.standby.replicas` on the client has no effect; Kafka Streams ignores these client-side settings. Similarly, the existing cluster configuration, `initial.rebalance.delay.ms`, doesn’t apply to streams groups. You can use the `kafka-configs` tool to set these configurations, for example: ```bash kafka-configs --bootstrap-server localhost:9092 --alter --entity-type groups --entity-name wordcount --add-config streams.num.standby.replicas=1 ``` ## Streams API changes in Kafka Streams version 4.1.x ### KIP-1071: Early Access of the Streams Rebalance Protocol The Streams Rebalance Protocol is available as an Early Access feature. This release covers a subset of the functionality detailed in [KIP-1071](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1071%3A+Streams+Rebalance+Protocol). ### KIP-1111: Enforce explicit naming for internal resources [KIP-1111](https://cwiki.apache.org/confluence/x/4Y_MEw) 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 you name every internal resource before deploying the Kafka Streams application, which is essential for upgrading your topology. You can enable this feature with `StreamsConfig` by using the `StreamsConfig#ENSURE_EXPLICIT_INTERNAL_RESOURCE_NAMING_CONFIG` parameter. When set to `true`, the application doesn’t 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](https://cwiki.apache.org/confluence/x/lAtYEQ). ## Streams API changes in Kafka Streams version 4.0.x For a full list of API changes, see [KAFKA-12822](https://issues.apache.org/jira/browse/KAFKA-12822). - Kafka Streams 4.0 removes all public API that was deprecated in Kafka Streams 3.6 or an earlier release, except for `JoinWindows.of()` and `JoinWindows#grace()`. For more information, see [KAFKA-17531](https://issues.apache.org/jira/browse/KAFKA-17531). - Ensure that your Java clients are at Kafka Streams version 2.1 or later. - Take care with Kafka clients that aren’t part of Apache Kafka®. For more information, see [KIP-896](https://cwiki.apache.org/confluence/display/KAFKA/KIP-896%3A+Remove+old+client+protocol+API+versions+in+Kafka+4.0). - Kafka Streams 4.0 raises the minimum Java version for clients and Kafka Streams applications from Java 8 to Java 11. For more information, see [KIP-750](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181308223). - Kafka Streams 4.0 no longer supports `eos-v1` (Exactly Once Semantics version 1). ### KAFKA-12822: Remove deprecated APIs of Kafka Streams in 4.0 Kafka Streams 4.0 removes all methods, classes, APIs, and configuration parameters that were deprecated up to and including Kafka Streams 3.6. The following list shows some of the important deprecated APIs. For the full list, see [KAFKA-12822](https://issues.apache.org/jira/browse/KAFKA-12822). - [builder methods for time/session/join/sliding windows](https://issues.apache.org/jira/browse/KAFKA-16332) - [kstream.KStream#branch in Java and Scala](https://issues.apache.org/jira/browse/KAFKA-12824) - [KafkaStreams#setUncaughtExceptionHandler()](https://issues.apache.org/jira/browse/KAFKA-12827) - [KStream#through() in Java and Scala](https://issues.apache.org/jira/browse/KAFKA-12823) - [Old processor APIs](https://issues.apache.org/jira/browse/KAFKA-12829) - [transformer methods and classes in Java and Scala](https://issues.apache.org/jira/browse/KAFKA-16339) ### KIP-1056: Remove `default.` prefix for exception handler StreamsConfig KIP-1056 deprecates the `default.deserialization.exception.handler` and `default.production.exception.handler` configurations because they don’t have any overwrites. Instead, use the new `deserialization.exception.handler` and `production.exception.handler` configurations. For more information, see [KIP-1056](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1056%3A+Remove+%60default.%60+prefix+for+exception+handler+StreamsConfig). ### KIP-1070: Deprecate MockProcessorContext The previous release introduced a new version of the Processor API and incrementally replaced and deprecated the old Processor API. KIP-1070 deprecates the following APIs: - `MockProcessorContext` - `Transformer` - `TransformerSupplier` - `ValueTransformer` - `ValueTransformerSupplier` For more information, see [KIP-1070](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1070%3A+deprecate+MockProcessorContext). ### KIP-1077: Deprecate ForeachProcessor and move to internal package KIP-1077 deprecates the `ForeachProcessor` class. This change improves the organization and clarity of the Kafka Streams API by keeping internal classes out of public packages. For more information, see [KIP-1077](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1077%3A+Deprecate+%60ForeachProcessor%60+and+move+to+internal+package). ### KIP-1078: Remove leaking getter methods in Joined helper class KIP-1078 deprecates the leaking getter methods in the `Joined` helper class. These methods are scheduled for future removal without a replacement, because they don’t add any value to Kafka Streams users. For more information, see [KIP-1078](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1078%3A+Remove+Leaking+Getter+Methods+in+Joined+Helper+Class). ### KIP-1085: Fix leaking \*_DOC variables in StreamsConfig To improve encapsulation and organization of configuration documentation within Kafka Streams, KIP-1085 deprecates certain public doc description variables that are used only within the `StreamsConfig` or `TopologyConfig` classes. KIP-1085 also deprecates the unused `DUMMY_THREAD_INDEX` variable. For more information, see [KIP-1085](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1085%3A+Fix+leaking+*_DOC+variables+in+StreamsConfig). ### KIP-1087: Removing intermediateTopicsOption from StreamsResetter Because Kafka Streams 4.0 removes the already-deprecated `#through` method, the `intermediateTopicsOption` of the `StreamsResetter` tool in Kafka is no longer needed; KIP-1087 deprecates it. For more information, see [KIP-1087](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1087%3A+Removing+intermediateTopicsOption+from+StreamsResetter). ### KIP-1091: Improved Kafka Streams operator metrics Because the broker can’t collect string metrics ([KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability)), this version introduces numeric counterparts that enable proper broker-side metric collection for Kafka Streams applications. KIP-1091 exposes these metrics at the `INFO` recording level and adds a thread-level metric with a `String` value for users leveraging Java Management Extensions (JMX). For more information, see [KIP-1091](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1091%3A+Improved+Kafka+Streams+operator+metrics). ### KIP-1104: Allow foreign key extraction from both key and value in KTable joins KIP-1104 introduces a new method in the Java and Scala APIs that accepts a `BiFunction` for foreign key extraction, enabling foreign key extraction from both the key and value in KTable joins. This change reduces storage overhead and improves API usability. Previously, foreign key joins in KTables allowed only extraction from the value, which led to data duplication and potential inconsistencies. KIP-1104 deprecates the existing methods but doesn’t remove them, ensuring backward compatibility. For more information, see [KIP-1104](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1104%3A+Allow+Foreign+Key+Extraction+from+Both+Key+and+Value+in+KTable+Joins). ### KIP-1106: Add duration based offset reset option for consumer clients KIP-1106 deprecates the `Topology.AutoOffsetReset` enum and replaces it with a new class, `org.apache.kafka.streams.AutoOffsetReset`, that captures the reset strategies. KIP-1106 also adds new methods to the `org.apache.kafka.streams.Topology` and `org.apache.kafka.streams.kstream.Consumed` classes to support the new reset strategy. These changes provide more flexibility and efficiency for managing offsets, especially in scenarios that involve long-term storage and infinite retention. For more information, see [KIP-1106](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1106%3A+Add+duration+based+offset+reset+option+for+consumer+clients). ### 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 configurations by using the new `StreamsConfig#PROCESSOR_WRAPPER_CLASS_CONFIG` configuration. Kafka Streams applies this configuration during the topology building phase, so the configuration doesn’t take effect unless you pass it when you create the `StreamsBuilder` (DSL) or `Topology` (PAPI) objects. You *must* use the `StreamsBuilder` or `Topology` constructor overload that accepts a `TopologyConfig` parameter so that Kafka Streams picks up the `StreamsConfig#PROCESSOR_WRAPPER_CLASS_CONFIG`. For more information, see [KIP-1112](https://cwiki.apache.org/confluence/x/TZCMEw). ### 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. - RocksDB 9.7.3 removes the `org.rocksdb.AccessHint` class and its associated methods. - RocksDB 9.7.3 removes several methods related to compressed block cache configuration in the `BlockBasedTableConfig` class, including `blockCacheCompressedNumShardBits`, `blockCacheCompressedSize`, and their corresponding setters. The `cache` option now consolidates this functionality; configure your compressed block cache by using the `setCache` method instead. - RocksDB 9.7.3 removes the `NO_FILE_CLOSES` field from the `org.rocksdb.TickerType` enum. As a result, the `number-open-files` metric doesn’t work as expected. The `number-open-files` metric returns the constant `-1` until it is officially removed. - The `org.rocksdb.Options.setLogger()` method now accepts a `LoggerInterface` parameter instead of the previous `Logger`. RocksDB 9.7.3 also modifies some data types in its Java API. These changes, along with the removed class, field, and new methods, are primarily relevant to users implementing custom RocksDB configurations. These changes are largely transparent to most Kafka Streams users. However, if you use advanced RocksDB customizations in your Streams applications, particularly through the `rocksdb.config.setter`, consult the [RocksDB 9.7.3 changelog](https://github.com/facebook/rocksdb/releases/tag/v9.7.4) 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. ### KIP-714: Client metrics and observability In this release, the `ClientInstanceIds` instance stores the global consumer `Uuid` for the [KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability#KIP714:Clientmetricsandobservability-Clientidentificationandtheclientinstanceid) 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, Kafka Streams didn’t invoke the `ProductionExceptionHandler` on a (retryable) `TimeoutException`. In Kafka Streams 4.0, Kafka Streams calls the handler, and the default handler returns `RETRY` to preserve existing behavior. A custom handler can break the infinite retry loop by returning either `CONTINUE` or `FAIL`. For more information, see [KIP-1065](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=311627309). ### KIP-1076: Metrics for client applications KIP-714 extension You can collect Kafka Streams metrics broker-side by using the [KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability) broker plugin. The plugin collects metrics for the internally used clients of a Kafka Streams application and for the Kafka Streams runtime itself. For more information, see [KIP-1076](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1076%3A++Metrics+for+client+applications+KIP-714+extension). ## Streams API changes in Kafka Streams 3.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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1033%3A+Add+Kafka+Streams+exception+handler+for+exceptions+occurring+during+processing) ### KIP-1049: Customize logging interval Kafka Streams now enables you to customize the logging interval of the stream-thread runtime summary by using the newly added `log.summary.interval.ms` configuration. By default, Kafka Streams logs the summary every 2 minutes. For more information, see [KIP-1049](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1049%3A+Add+config+log.summary.interval.ms+to+Kafka+Streams). ## Streams API changes in Kafka Streams 3.8.0 ### KIP-924: Customizable task assignment for Streams Kafka Streams now supports customizable task assignment strategies through 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` configuration, switch to using the new `task.assignor.class` configuration instead. The internal configuration is deprecated and is scheduled for removal in a future release. If you were previously plugging in the `StickyTaskAssignor` through the legacy `internal.task.assignor.class` configuration, you must ensure that you are importing the new `org.apache.kafka.streams.processor.assignment.StickTaskAssignor` 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-924%3A+customizable+task+assignment+for+Streams). ### KIP-813: Shareable state stores The Processor API now supports so-called read-only state stores, added in [KIP-813](https://cwiki.apache.org/confluence/display/KAFKA/KIP-813%3A+Shareable+State+Stores). 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-989%3A+Improved+StateStore+Iterator+metrics+for+detecting+leaks) 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 Kafka Streams 3.7.0 ### KIP-925: Rack-aware task assignment in Kafka Streams (part 2) Part one of [KIP-925](https://cwiki.apache.org/confluence/display/KAFKA/KIP-925%3A+Rack+aware+task+assignment+in+Kafka+Streams) added the `min_traffic` assignment strategy for Kafka Streams. 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-954%3A+expand+default+DSL+store+configuration+to+custom+types) builds on [KIP-591](https://cwiki.apache.org/confluence/display/KAFKA/KIP-591%3A+Add+Kafka+Streams+config+to+set+default+state+store) and enables you to provide a default state store provider for your custom stores. KIP-954 provides a new interface, along with default support for RocksDB and in-memory state stores. ### KIP-962: Relax non-null key requirement in Kafka Streams Previously, Kafka Streams treated records with null-keys as invalid input for joins and dropped them. [KIP-962](https://cwiki.apache.org/confluence/display/KAFKA/KIP-962%3A+Relax+non-null+key+requirement+in+Kafka+Streams) relaxes this behavior for various left-joins so that the join can process null-key records 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. ```java //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 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 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 The Kafka 3.5 release added versioned state stores ([KIP-889](https://cwiki.apache.org/confluence/display/KAFKA/KIP-889%3A+Versioned+State+Stores)), but you couldn’t query the new stores. [KIP-960](https://cwiki.apache.org/confluence/display/KAFKA/KIP-960%3A+Support+single-key_single-timestamp+interactive+queries+%28IQv2%29+for+versioned+state+stores) and [KIP-968](https://cwiki.apache.org/confluence/display/KAFKA/KIP-968%3A+Support+single-key_multi-timestamp+interactive+queries+%28IQv2%29+for+versioned+state+stores) close this gap by adding two new IQv2 query types, `VersionedKeyQuery` and `MultiVersionedKeyQuery`. Both queries enable you to look up a single key and 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-985%3A+Add+reverseRange+and+reverseAll+query+over+kv-store+in+IQv2) 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-988%3A+Streams+Standby+Update+Listener) 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: Add `TimestampedKeyQuery` and `TimestampedRangeQuery` to IQv2 [KIP-992](https://cwiki.apache.org/confluence/display/KAFKA/KIP-992%3A+Proposal+to+introduce+IQv2+Query+Types%3A+TimestampedKeyQuery+and+TimestampedRangeQuery) 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` configuration 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 Kafka Streams 3.6.0 ### KIP-923: Add a grace period to stream-table join [KIP-923](https://cwiki.apache.org/confluence/display/KAFKA/KIP-923%3A+Add+A+Grace+Period+to+Stream+Table+Join) 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 [KIP-925](https://cwiki.apache.org/confluence/display/KAFKA/KIP-925%3A+Rack+aware+task+assignment+in+Kafka+Streams) introduces rack-aware task assignment. You can enable rack-aware task assignment for `StickyTaskAssignor` or `HighAvailabilityTaskAssignor` to compute task assignments that can minimize cross-rack traffic under certain conditions. For more information, including how to enable and configure this feature, see [rack.aware.assignment.strategy](/platform/current/streams/developer-guide/config-streams.html#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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-941%3A+Range+queries+to+accept+null+lower+and+upper+bounds) 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 in Kafka Streams 3.5.0 Downgrading from Kafka Streams 3.5.x or later to Kafka Streams 3.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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-904%3A+Kafka+Streams+-+Guarantee+subtractor+is+called+before+adder+if+key+has+not+changed). 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-399%3A+Extend+ProductionExceptionHandler+to+cover+serialization+exceptions) adds a method, `handleSerializationException()`, to the `ProductionExceptionHandler` interface to handle any serialization errors encountered while producing records. ### KIP-884: Add configuration to configure KafkaClientSupplier [KIP-884](https://cwiki.apache.org/confluence/display/KAFKA/KIP-884%3A+Add+config+to+configure+KafkaClientSupplier+in+Kafka+Streams) adds a new configuration, `default.client.supplier` that enables using a custom `KafkaClientSupplier` without any code changes. ### KIP-889: Versioned state stores [KIP-889](https://cwiki.apache.org/confluence/display/KAFKA/KIP-889%3A+Versioned+State+Stores) introduces versioned state stores to improve the accuracy of joins when out-of-order records are processed. For more information, see [dsl.timestamp.based.semantics](/platform/current/streams/developer-guide/dsl-api.html#dsl-timestamp-based-semantics). In addition to KIP-889, [KIP-914](https://cwiki.apache.org/confluence/display/KAFKA/KIP-914%3A+DSL+Processor+Semantics+for+Versioned+Stores) 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 might 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-904%3A+Kafka+Streams+-+Guarantee+subtractor+is+called+before+adder+if+key+has+not+changed) 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, Kafka Streams applied two independent updates to the same row, resulting in spurious intermediate results. KIP-904 detects this case and applies only a single update, which avoids spurious intermediate results. ### KIP-907: Add Boolean serde to public interface Kafka Streams includes built-in Serdes for most primitive types. [KIP-907](https://cwiki.apache.org/confluence/display/KAFKA/KIP-907%3A+Add+Boolean+Serde+to+public+interface) adds a new one for booleans. ## Streams API changes in Kafka Streams 3.4.0 ### KIP-770: Replace cache.max.bytes.buffering with statestore.cache.max.bytes [KIP-770](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=186878390) deprecates the existing `cache.max.bytes.buffering` configuration and introduces a new `statestore.cache.max.bytes` configuration to replace it. The semantics and default value of the cache size configuration are unchanged. This KIP also adds a new `cache.size` metric at the DEBUG level so users can monitor the actual size of the Kafka Streams cache. ### KIP-837: Allow multicasting a result record [KIP-837](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=211883356) 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](https://cwiki.apache.org/confluence/x/UY9rDQ) 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 Kafka Streams 3.3.0 ### Source/sink node metrics for consumed/produced throughput in Kafka Streams Starting with Kafka Streams 3.3.0, 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](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=211886093) 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 and resume Kafka Streams topologies [KIP-834](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=211882832) 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-820%3A+Extend+KStream+process+with+new+Processor+API) 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-812%3A+Introduce+another+form+of+the+%60KafkaStreams.close%28%29%60+API+that+forces+the+member+to+leave+the+consumer+group) 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 Kafka Streams 3.2.0 ### Rack awareness for Kafka Streams Starting with Kafka Streams 3.2.0, Kafka Streams can distribute its standby replicas over distinct “racks” with [KIP-708](https://cwiki.apache.org/confluence/display/KAFKA/KIP-708%3A+Rack+awareness+for+Kafka+Streams). 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 run in. Users can specify the tags 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 across different task dimensions. Rack-aware standby assignment improves fault tolerance in case of the failure of an entire “rack”. For example, you can use this feature to ensure that Kafka Streams distributes replicas across different availability zones in a cloud hosting provider. ### Add record metadata to state store context [KIP-791](https://cwiki.apache.org/confluence/display/KAFKA/KIP-791%3A+Add+Record+Metadata+to+State+Store+Context) adds the `recordMetadata()` method to the `StateStoreContext` and provides access to the topic, partition, and offset of the current record. Exposing the current context this way enables state stores to track their current offset in each input partition and to implement the consistency mechanisms that [KIP-796](https://cwiki.apache.org/confluence/display/KAFKA/KIP-796%3A+Interactive+Query+v2) introduces. ### Interactive Query v2 preview Kafka Streams 3.2.0 introduces Interactive Queries v2 (IQv2). IQv2 is a preview feature, and the IQv2 interfaces are marked as `@Evolving`, which means they might break compatibility in minor releases without a deprecation period if preview users find significant flaws in the current API. - [KIP-796](https://cwiki.apache.org/confluence/display/KAFKA/KIP-796%3A+Interactive+Query+v2) 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. You can add specific query types to Interactive Query v2 by implementing the `Query` interface. KIP-976 also defines the `KeyQuery` class so that users can evaluate a key/value lookup by using IQv2. - [KIP-805](https://cwiki.apache.org/confluence/display/KAFKA/KIP-805%3A+Add+range+and+scan+query+over+kv-store+in+IQv2) 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-806%3A+Add+session+and+window+query+over+kv-store+in+IQv2) 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 Kafka Streams 3.1.0 ### Java 17 support In Kafka Streams 3.1.0, Kafka Streams supports Java 17. ### Improved left/outer stream-stream join semantics [KIP-633](https://cwiki.apache.org/confluence/display/KAFKA/KIP-633%3A+Deprecate+24-hour+Default+Grace+Period+for+Windowed+Operations+in+Streams) improves the semantics of left/outer stream-stream joins. Previously, a left/outer stream-stream join might have emitted so-called spurious left/outer results because of an eager-emit strategy. KIP-633 changes the implementation to emit left/outer join result records only after the join window closes. KIP-633 deprecates the old API for specifying the join window, `JoinWindows.of()`, in favor of the `JoinWindows.ofTimeDifferenceAndGrace()` and `JoinWindows.ofTimeDifferencWithNoGrace()` methods. The new semantics apply only if you use the new join window builders. KIP-633 also makes setting a grace period mandatory for windowed aggregations: `TimeWindows` (hopping/tumbling), `SessionWindows`, and `SlidingWindows`. KIP-633 deprecates the corresponding `.of(...)` builder methods in favor of the new `.ofTimeDifferenceAndGrace()` and `.ofTimeDifferencWithNoGrace()` methods. ### New metrics to track blocking times [KIP-761](https://cwiki.apache.org/confluence/display/KAFKA/KIP-761%3A+Add+Total+Blocked+Time+Metric+to+Streams) adds new metrics that enable tracking blocking times on the underlying consumer and producer clients. For more information, see [Monitor Kafka Streams Applications](/platform/current/streams/monitoring.html). ### Interactive Query improvements [KIP-763](https://cwiki.apache.org/confluence/display/KAFKA/KIP-763%3A+Range+queries+with+open+endpoints) and [KIP-766](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=186876596) improve Interactive Queries. 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 through [KIP-775](https://cwiki.apache.org/confluence/display/KAFKA/KIP-775%3A+Custom+partitioners+in+foreign+key+joins). 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. ## Upgrade guide for versions earlier than Kafka Streams 3.1.x For upgrade guidance on Kafka Streams versions earlier than 3.1.0, see [Legacy Streams Upgrade Guide](https://docs.confluent.io/legacy/platform/7.0/streams/upgrade-guide.html). #### NOTE This website includes content developed at the [Apache Software Foundation](https://www.apache.org/) under the terms of the [Apache License v2](https://www.apache.org/licenses/LICENSE-2.0.html).