<a id="streams-upgrade-guide"></a>

# Kafka Streams Upgrade Guide for Confluent Platform

Upgrading a Kafka Streams application to Confluent Platform 8.3.1 (Kafka Streams
4.3.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.
To upgrade from Confluent Platform versions earlier than 7.1.x, see
[Legacy Streams Upgrade Guide](https://docs.confluent.io/platform/7.0/streams/upgrade-guide.html).

## Upgrade from older versions

Upgrading from any earlier Kafka Streams version to Confluent Platform 8.3.1 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 *0.10.0* through *3.4*.
2. During the second bounce, remove the `upgrade.from` config.

This is required to handle three changes safely:

- The introduction of the cooperative rebalancing protocol of the embedded
  consumer. For more information, see
  [KIP-429](https://cwiki.apache.org/confluence/x/vAclBg).
- 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, once the entire group is on Confluent Platform 8.3.1
(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` 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.3.1 (Kafka Streams 4.3.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.3.1
(Kafka Streams 4.3.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.3.1 (Kafka Streams 4.3.0)
   application instances.

## Upgrade to Confluent Platform 8.3.1 from Confluent Platform 7.1.x or later

<a id="streams-upgrade-guide-compatibility"></a>

### Compatibility

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

Forward-compatible to newer clusters up to Confluent Platform 8.3.1:
: Existing Kafka Streams applications built with Confluent Platform 3.x and later work with
  upgraded Kafka clusters running Confluent Platform 8.3.1.

Backward-compatible to older clusters down to Confluent Platform 3.0.x:
: - New Kafka Streams applications built with Confluent Platform 8.3.1 work with older Kafka
    clusters running Confluent Platform 6.x down to 3.0.x.
  - Kafka clusters running Confluent Platform 3.0.x, Confluent Platform 3.1.x, or Confluent Platform 3.2.x are *not*
    compatible with new Confluent Platform 8.3.1 Kafka Streams applications.

### 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](https://docs.confluent.io/platform/7.9/streams/upgrade-guide.html).

#### 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.3.x (Kafka Streams 4.1.x - 4.3.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.3.x (Kafka Streams 2.6.x - 4.3.x) | compatible; enabling exactly-once v2 requires broker Confluent Platform 5.5.x / Kafka Streams 2.5.x or higher | compatible                                                              |

The Streams API is not compatible with Kafka clusters running older Kafka versions (0.7, 0.8, 0.9).

<a id="streams-upgrade-guide-rocksdb-compatibility"></a>

### 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.3.x                   | 8.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                   | 8.2.x                        | 10.1.3            |                                                                                                                            |
| 4.1.x                   | 8.1.x                        | 10.1.3            |                                                                                                                            |
| 4.0.x                   | 8.0.x                        | 9.7.3             | Significant API changes, see [Upgraded RocksDB dependency](#streams-upgrade-guide-rocksdb).                                |
| 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             |                                                                                                                            |
<!-- https://github.com/apache/kafka/blob/trunk/gradle/dependencies.gradle -->
<!-- 4.1.x              8.1.x         9.7.3 -->
<!-- trunk                            10.1.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)](developer-guide/memory-mgmt.md#rocksdb-mem-mgmt)
- [RocksDB Releases](https://github.com/facebook/rocksdb/releases)

<a id="upgrade-streams-cp-release"></a>

## Upgrade your Kafka Streams applications to Confluent Platform 8.3.1 (Kafka Streams version 4.3.0)

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

For example, in your `pom.xml` file:

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

<a id="streams-upgrade-guide-api-changes"></a>

## 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](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 the upcoming Confluent Platform 8.3.2 release. As an
alternative, the fix is available now in Kafka 4.3.1.

### 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](developer-guide/config-streams.md#streams-developer-guide-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`.
For configuration details, refer to
[processing.exception.handler.global.enabled](developer-guide/config-streams.md#streams-developer-guide-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](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](../schema-registry/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");
```

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](developer-guide/config-streams.md#streams-developer-guide-dsl-store-format). 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 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.

<a id="migrate-default-headers"></a>

#### 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.

<a id="streams-upgrade-guide-sr-schema-id-in-header"></a>

### 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](../schema-registry/fundamentals/serdes-develop/index.md#messages-wire-format-schema-id-in-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](developer-guide/config-streams.md#streams-developer-guide-dsl-store-format) and in the
[Migration from DEFAULT to HEADERS](#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 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 ID
  is in the header is resolved from the header, and 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, 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.

<a id="streams-upgrade-guide-sr-schema-id-header-dsl"></a>

#### 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-upgrade-guide-sr-schema-id-header-papi).

1. Enable headers-aware state stores by setting `dsl.store.format=HEADERS` and
   complete the `DEFAULT` to `HEADERS` migration, as described in the
   preceding section.
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.

<a id="streams-upgrade-guide-sr-schema-id-header-papi"></a>

#### 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](#streams-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 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.

<a id="streams-upgrade-guide-kip-1244"></a>

### 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](developer-guide/dsl-api.md#streams-developer-guide-dsl) and the
[Kafka Streams tutorial](quickstart.md#streams-quick-start) cover the equivalent Java
DSL patterns.

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.

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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1035%3A+StateStore+Managed+Changelog+Offsets)
for the API details.

<a id="streams-upgrade-guide-kip-1035-downgrade"></a>

#### 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](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 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](https://issues.apache.org/jira/browse/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](developer-guide/streams-rebalance-protocol.md#streams-developer-rebalance-protocol-not-included).

- To start using the Streams Rebalance Protocol, see
  [Enable the protocol](developer-guide/streams-rebalance-protocol.md#streams-developer-rebalance-protocol-enable).
- To migrate to the Streams Rebalance Protocol, see
  [Migration from classic protocol](developer-guide/streams-rebalance-protocol.md#streams-developer-rebalance-protocol-migrate).

For more information, see [Streams Rebalance Protocol](developer-guide/streams-rebalance-protocol.md#streams-developer-rebalance-protocol).

### KIP-1034: Dead Letter Queue for production exceptions

Kafka Streams now supports Dead Letter Queue (DLQ). A new configuration, named
`errors.deadletterqueue.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.deadletterqueue.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](https://cwiki.apache.org/confluence/x/HwviEQ).

### 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](https://cwiki.apache.org/confluence/x/QAq9F).

### 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](https://cwiki.apache.org/confluence/x/jgl3Fw).

### 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](https://cwiki.apache.org/confluence/x/8AxJFg).

### 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)](developer-guide/streams-rebalance-protocol.md#streams-developer-rebalance-protocol)
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](https://cwiki.apache.org/confluence/x/ywnxFg).

### 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](https://cwiki.apache.org/confluence/x/jQobFw).

## 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](https://issues.apache.org/jira/browse/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](https://issues.apache.org/jira/browse/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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1071%3A+Streams+Rebalance+Protocol).
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.
- **CLI Integration:** You can list, describe, and delete streams groups via
  the `kafka-streams-groups.sh` script.

#### Not included in Early Access

- **Static membership:** Setting a client `instance.id` is rejected.
- **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](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 every internal resource is
named before the Kafka Streams application is deployed, which is essential for
upgrading your topology. You can enable this feature via `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](https://cwiki.apache.org/confluence/x/lAtYEQ).

## 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](https://issues.apache.org/jira/browse/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](https://issues.apache.org/jira/browse/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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-896%3A+Remove+old+client+protocol+API+versions+in+Kafka+4.0).
- 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](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=181308223)
- Brokers, Connect, and tools now require Java 17. For more information, see
  [KIP-1013](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=284789510)
- 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](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)

If you have the “merge repartition topics” optimization enabled, migrating
from `KStream#transformValues()` to `KStream#processValues()` might not be
safe, due to
[KAFKA-19668](https://issues.apache.org/jira/browse/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](developer-guide/dsl-api.md#streams-developer-guide-dsl-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](https://cwiki.apache.org/confluence/x/Y41yEg).

### 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](https://cwiki.apache.org/confluence/x/sxCTEg).

### 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](https://cwiki.apache.org/confluence/x/eA-OEg).

### 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](https://cwiki.apache.org/confluence/x/hg-OEg).

### 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](https://cwiki.apache.org/confluence/x/hYz9Eg).

### 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](https://cwiki.apache.org/confluence/x/Vo39Eg).

### KIP-1091: Improved Kafka Streams operator metrics

Because string metrics can’t be collected on the broker side
([KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability)),
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](https://cwiki.apache.org/confluence/x/IgstEw).

### 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](https://cwiki.apache.org/confluence/x/gIuMEw).

### 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](https://cwiki.apache.org/confluence/x/NIyMEw).

### 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](https://cwiki.apache.org/confluence/x/TZCMEw).

<a id="streams-upgrade-guide-rocksdb"></a>

### 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](https://github.com/facebook/rocksdb/releases/tag/v9.7.3)
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](https://cwiki.apache.org/confluence/x/2xRRCg#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, 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](https://cwiki.apache.org/confluence/x/LQ6TEg).

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

Kafka Streams metrics can be collected broker-side by using the
[KIP-714](https://cwiki.apache.org/confluence/display/KAFKA/KIP-714%3A+Client+metrics+and+observability)
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](https://cwiki.apache.org/confluence/x/XA-OEg).

## 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](https://cwiki.apache.org/confluence/x/xQniEQ)

### 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](https://cwiki.apache.org/confluence/x/fwpeEg).

## 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](https://cwiki.apache.org/confluence/x/PxU0Dw).

### 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/x/9KCzDw)
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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-925%3A+Rack+aware+task+assignment+in+Kafka+Streams),
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](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.
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](https://cwiki.apache.org/confluence/x/f5CzDw)
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.

```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<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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-889%3A+Versioned+State+Stores)),
but it was not possible to query the new stores.
[KIP-960](https://cwiki.apache.org/confluence/x/qo_zDw) and
[KIP-968](https://cwiki.apache.org/confluence/x/WpSzDw)
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](https://cwiki.apache.org/confluence/x/eKCzDw)
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/x/yqCzDw)
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](https://cwiki.apache.org/confluence/x/TYxEE)
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](https://cwiki.apache.org/confluence/x/lAs0Dw)
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](https://cwiki.apache.org/confluence/x/CQ40Dw).
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](developer-guide/config-streams.md#streams-developer-guide-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/x/_Rk0Dw) 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 Confluent Platform 7.5.0

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](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/x/R4nQBQ)
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](https://cwiki.apache.org/confluence/x/AZfGDQ)
adds a new config, `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/x/AIwODg)
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](developer-guide/dsl-api.md#streams-developer-guide-dsl-timestamp-based-semantics).

In addition to KIP-889,
[KIP-914](https://cwiki.apache.org/confluence/x/QorFDg)
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](https://cwiki.apache.org/confluence/x/P5VbDg)
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](https://cwiki.apache.org/confluence/x/pZpbDg)
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](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=186878390)
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](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 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](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/resume KafkaStreams 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/x/yKbkCw)
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/x/KZvkCw)
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](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 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-791%3A+Add+Record+Metadata+to+State+Store+Context)
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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-796%3A+Interactive+Query+v2).

### 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](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. 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](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 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](https://cwiki.apache.org/confluence/display/KAFKA/KIP-633%3A+Deprecate+24-hour+Default+Grace+Period+for+Windowed+Operations+in+Streams).

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](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 [Kafka Streams metrics](monitoring.md#streams-monitoring).

### Interactive Query improvements

Interactive Queries were improved by
[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).
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](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 Confluent Platform 7.1.x (Kafka Streams 3.1)

For upgrade guidance on Confluent Platform versions earlier than 7.1.0, see
[Legacy Streams Upgrade Guide](https://docs.confluent.io/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).