# Kafka Streams Domain Specific Language for Confluent Platform The Kafka Streams Domain Specific Language (DSL) is a high-level, declarative API for building stream processing applications with operations such as `map`, `filter`, `join`, and `aggregate`. Built on the Streams Processor API, the DSL works well for most users, especially beginners, and expresses most data processing operations in a few lines of code. ## Overview In comparison to the [Processor API](processor-api.md#streams-developer-guide-processor-api), only the DSL supports: * Built-in abstractions for [streams and tables](../concepts.md#streams-concepts-duality) in the form of [KStream](../concepts.md#streams-concepts-kstream), [KTable](../concepts.md#streams-concepts-ktable), and [GlobalKTable](../concepts.md#streams-concepts-globalktable). Having first-class support for streams and tables is crucial because, in practice, most use cases require not just either streams or databases/tables, but a combination of both. For example, if your use case is to create a customer 360-degree view that is updated in real-time, what your application will be doing is transforming many input *streams* of customer-related events into an output *table* that contains a continuously updated 360-degree view of your customers. * Declarative, functional programming style with [stateless transformations](#streams-developer-guide-dsl-transformations-stateless) (e.g., `map` and `filter`) as well as [stateful transformations](#streams-developer-guide-dsl-transformations-stateful) such as [aggregations](#streams-developer-guide-dsl-aggregating) (e.g., `count` and `reduce`), [joins](#streams-developer-guide-dsl-joins) (e.g., `leftJoin`), and [windowing](#streams-developer-guide-dsl-windowing) (e.g., [session windows](#windowing-session)). With the DSL, you can define [processor topologies](../concepts.md#streams-concepts-processor-topology) (that is, the logical processing plan) in your application. The steps to accomplish this are: 1. Specify [one or more input streams that are read from Kafka topics](#streams-developer-guide-dsl-sources). 2. Compose [transformations](#streams-developer-guide-dsl-transformations) on these streams. 3. Write the [resulting output streams back to Kafka topics](#streams-developer-guide-dsl-destinations), or expose the processing results of your application directly to other applications through [Kafka Streams Interactive Queries for Confluent Platform](interactive-queries.md#streams-developer-guide-interactive-queries), for example, through a REST API. After the application is run, the defined processor topologies are continuously executed (that is, the processing plan is put into action). A step-by-step guide for writing a stream processing application using the DSL is provided below. After you build your Kafka Streams application with the DSL, you can view the underlying `Topology` by first executing `StreamsBuilder#build()` which returns the `Topology` object. Then, to view the `Topology`, call `Topology#describe()`. Full details on describing a `Topology` can be found in [describing a topology](processor-api.md#streams-developer-guide-describing-a-topology). For a complete list of available API functionality, see also the [Kafka Streams Javadocs](../javadocs.md#streams-javadocs). ## Creating source streams from Kafka You can easily read data from Apache Kafka® topics into your application. The following operations are supported. | Reading from Kafka | Description | |----------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | **Stream**

- *input topics* → KStream | Creates a [KStream](../concepts.md#streams-concepts-kstream) from the specified Kafka input topics and interprets the data
as a [record stream](../concepts.md#streams-concepts-kstream).
A `KStream` represents a *partitioned* record stream.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/StreamsBuilder.html#stream(java.lang.String))

In the case of a KStream, the local KStream instance of every application instance will
be populated with data from only **a subset** of the partitions of the input topic. Collectively, across
all application instances, all input topic partitions are read and processed.

```java
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;

StreamsBuilder builder = new StreamsBuilder();

KStream wordCounts = builder.stream(
"word-counts-input-topic", /* input topic */
Consumed.with(
Serdes.String(), /* key serde */
Serdes.Long() /* value serde */
)
);
```

If you do not specify Serdes explicitly, the default Serdes from the
[configuration](config-streams.md#streams-developer-guide-configuration) are used.

You **must specify Serdes explicitly** if the key or value types of the records in the Kafka input
topics do not match the configured default Serdes. For information about configuring default Serdes, available
Serdes, and implementing your own custom Serdes see [Kafka Streams Data Types and Serialization for Confluent Platform](datatypes.md#streams-developer-guide-serdes).

Several variants of `stream` exist. For example, you can specify a regex pattern for input topics to read from.
Note that all matching topics will be part of the same input topic group, and the work will not be parallelized
for different topics if subscribed to in this way.

Kafka Streams assumes that input topics are already partitioned by key, which means that `builder.stream()` creates a stream assuming that
the source topic is partitioned by key. | | **Table**

- *input topic* → KTable | Reads the specified Kafka input topic into a [KTable](../concepts.md#streams-concepts-ktable). The topic is
interpreted as a changelog stream, where records with the same key are interpreted as UPSERT (that is, INSERT/UPDATE)
(when the record value is not `null`) or as DELETE (when the value is `null`) for that key.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/StreamsBuilder.html#table-java.lang.String(java.lang.String))

In the case of a KTable, the local KTable instance of every application instance will
be populated with data from only **a subset** of the partitions of the input topic. Collectively, across
all application instances, all input topic partitions are read and processed.

You must provide a name for the table (more precisely, for the internal
[state store](../architecture.md#streams-architecture-state) that backs the table). This is required for
supporting [Kafka Streams Interactive Queries for Confluent Platform](interactive-queries.md#streams-developer-guide-interactive-queries) against the table. When a
name is not provided, the table is not queryable, and an internal name is provided for the state store.

If you do not specify Serdes explicitly, the default Serdes from the
[configuration](config-streams.md#streams-developer-guide-configuration) are used.

You **must specify Serdes explicitly** if the key or value types of the records in the Kafka input
topics do not match the configured default Serdes. For information about configuring default Serdes, available
Serdes, and implementing your own custom Serdes see [Kafka Streams Data Types and Serialization for Confluent Platform](datatypes.md#streams-developer-guide-serdes).

Several variants of `table` exist, for example to specify the `auto.offset.reset` policy to be used when
reading from the input topic. | | **Global Table**

- *input topic* → GlobalKTable | Reads the specified Kafka input topic into a [GlobalKTable](../concepts.md#streams-concepts-globalktable). The topic is
interpreted as a changelog stream, where records with the same key are interpreted as UPSERT (that is, INSERT/UPDATE)
(when the record value is not `null`) or as DELETE (when the value is `null`) for that key.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/StreamsBuilder.html#globalTable-java.lang.String(java.lang.String))

In the case of a GlobalKTable, the local GlobalKTable instance of every application instance will
be populated with data from all input topic partitions. Collectively, across
all application instances, all input topic partitions are consumed by all instances of the application.

You must provide a name for the table (more precisely, for the internal
[state store](../architecture.md#streams-architecture-state) that backs the table). This is required for
supporting [Kafka Streams Interactive Queries for Confluent Platform](interactive-queries.md#streams-developer-guide-interactive-queries) against the table. When a
name is not provided, the table is not queryable, and an internal name is provided for the state store.

```java
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.GlobalKTable;

StreamsBuilder builder = new StreamsBuilder();

GlobalKTable wordCounts = builder.globalTable(
"word-counts-input-topic",
Materialized.>as(
"word-counts-global-store" /* table/store name */)
.withKeySerde(Serdes.String()) /* key serde */
.withValueSerde(Serdes.Long()) /* value serde */
);
```

You **must specify Serdes explicitly** if the key or value types of the records in the Kafka input
topics do not match the configured default Serdes. For information about configuring default Serdes, available
Serdes, and implementing your own custom Serdes see [Kafka Streams Data Types and Serialization for Confluent Platform](datatypes.md#streams-developer-guide-serdes).

Several variants of `globalTable` exist, for example, to specify explicit Serdes. | ## Transform a stream The KStream and KTable interfaces support a variety of transformation operations. Each of these operations can be translated into one or more connected processors into the underlying processor topology. Because KStream and KTable are strongly typed, all of these transformation operations are defined as generic functions where you can specify the input and output data types. Some KStream transformations can generate one or more KStream objects, for example: - `filter` and `map` on a KStream generate another KStream - `split` on a KStream can generate multiple KStreams Some others can generate a KTable object, for example an aggregation of a KStream also yields a KTable. This allows Kafka Streams to continuously update the computed value upon arrivals of [out-of-order records](../concepts.md#streams-concepts-aggregations) after it has already been produced to the downstream transformation operators. All KTable transformation operations can only generate another KTable. However, the Kafka Streams DSL does provide a special function that converts a KTable representation into a KStream. All of these transformation methods can be chained together to compose a complex processor topology. These transformation operations are described in the following subsections: - [Stateless transformations](#streams-developer-guide-dsl-transformations-stateless) - [Stateful transformations](#streams-developer-guide-dsl-transformations-stateful) ### Stateless transformations Stateless transformations do not require state for processing and they do not require a state store associated with the stream processor. Kafka 0.11.0 and later allow you to materialize the result from a stateless `KTable` transformation. This allows the result to be queried through [Kafka Streams Interactive Queries for Confluent Platform](interactive-queries.md#streams-developer-guide-interactive-queries). To materialize a `KTable`, each of the below stateless operations [can be augmented](interactive-queries.md#streams-developer-guide-interactive-queries-local-key-value-stores) with an optional `queryableStoreName` argument. | Transformation | Description | |----------------------------------------------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | **Branch**

- KStream → KStream[] | Branch (or split) a `KStream` based on the supplied predicates into one or more `KStream` instances.
([details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#split-org.apache.kafka.streams.kstream.Predicate...-))

Predicates are evaluated in order. A record is placed to one and only one output stream on the first match:
if the n-th predicate evaluates to true, the record is placed to n-th stream. If no predicate matches, the
record is dropped.

Branching is useful, for example, to route records to different downstream topics.

```java
KStream stream = ...;
Map> branches =
stream.split(Named.as("Branch-"))
.branch((key, value) -> key.startsWith("A"), /* first predicate */
Branched.as("A"))
.branch((key, value) -> key.startsWith("B"), /* second predicate */
Branched.as("B"))
.defaultBranch(Branched.as("C"));

// KStream branches.get("Branch-A") contains all records whose keys start with "A"
// KStream branches.get("Branch-B") contains all records whose keys start with "B"
// KStream branches.get("Branch-C") contains all other records
``` | | **Broadcast/Multicast**

- no operator | Broadcast a `KStream` into multiple downstream operators.

A record is sent to more than one operator by applying multiple operators to the same `KStream` instance.

```java
KStream stream = ...;
KStream<...> stream1 = stream.map(...);
KStream<...> stream2 = stream.mapValues(...);
KStream<...> stream3 = stream.flatMap(...);
```

Multicast a `KStream` into multiple downstream operators.

In contrast to branching, which sends each record to at most one downstream branch, a multicast may send a record to any number of downstream `KStream` instances.

A multicast is implemented as a broadcast plus filters.

```java
KStream stream = ...;

// contains all records whose keys start with "A"
KStream<...> stream1 = stream.filter((key, value) -> key.startsWith("A"));

// contains all records whose keys start with "AB" (subset of stream1)
KStream<...> stream2 = stream.filter((key, value) -> key.startsWith("AB"));

// contains all records whose keys contains a "B" (superset of stream2)
KStream<...> stream3 = stream.filter((key, value) -> key.contains("B"));
``` | | **Filter**

- KStream → KStream
- KTable → KTable | Evaluates a boolean function for each element and retains those for which the function returns true.
([KStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#filter-org.apache.kafka.streams.kstream.Predicate-),
[KTable details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KTable.html#filter-org.apache.kafka.streams.kstream.Predicate-))

```java
KStream stream = ...;

// A filter that selects (keeps) only positive numbers
// Java example, using lambda expressions
KStream onlyPositives = stream.filter((key, value) -> value > 0);
``` | | **Inverse Filter**

- KStream → KStream
- KTable → KTable | Evaluates a boolean function for each element and drops those for which the function returns true.
([KStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#filterNot-org.apache.kafka.streams.kstream.Predicate-),
[KTable details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KTable.html#filterNot-org.apache.kafka.streams.kstream.Predicate-))

```java
KStream stream = ...;

// An inverse filter that discards any negative numbers or zero
// Java example, using lambda expressions
KStream onlyPositives = stream.filterNot((key, value) -> value <= 0);
``` | | **FlatMap**

- KStream → KStream | Takes one record and produces zero, one, or more records. You can modify the record keys and values, including
their types.
([details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#flatMap-org.apache.kafka.streams.kstream.KeyValueMapper-))

**Marks the stream for data re-partitioning:**
Applying a grouping or a join after `flatMap` will result in re-partitioning of the records.
If possible use `flatMapValues` instead, which will not cause data re-partitioning.

```java
KStream stream = ...;
KStream transformed = stream.flatMap(
// Here, we generate two output records for each input record.
// We also change the key and value types.
// Example: (345L, "Hello") -> ("HELLO", 1000), ("hello", 9000)
(key, value) -> {
List> result = new LinkedList<>();
result.add(KeyValue.pair(value.toUpperCase(), 1000));
result.add(KeyValue.pair(value.toLowerCase(), 9000));
return result;
}
);
``` | | **FlatMap (values only)**

- KStream → KStream | Takes one record and produces zero, one, or more records, while retaining the key of the original record.
You can modify the record values and the value type.
([details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#flatMapValues-org.apache.kafka.streams.kstream.ValueMapper-))

`flatMapValues` is preferable to `flatMap` because it will not cause data re-partitioning. However, you
cannot modify the key or key type like `flatMap` does.

```java
// Split a sentence into words.
KStream sentences = ...;
KStream words = sentences.flatMapValues(value -> Arrays.asList(value.split("\\s+")));
``` | | **Foreach**

- KStream → void
- KStream → void
- KTable → void | **Terminal operation.** Performs a stateless action on each record.
([details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#foreach-org.apache.kafka.streams.kstream.ForeachAction-))

You would use `foreach` to cause *side effects* based on the input data (similar to `peek`) and then *stop*
*further processing* of the input data (unlike `peek`, which is not a terminal operation).

**Note on processing guarantees:** Any side effects of an action (such as writing to external systems) are not
trackable by Kafka, which means they will typically not benefit from Kafka’s processing guarantees.

```java
KStream stream = ...;

// Print the contents of the KStream to the local console.
// Java example, using lambda expressions
stream.foreach((key, value) -> System.out.println(key + " => " + value));
``` | | **GroupByKey**

- KStream → KGroupedStream | Groups the records by the existing key.
([details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#groupByKey--))

Grouping is a prerequisite for [aggregating a stream or a table](#streams-developer-guide-dsl-aggregating)
and ensures that data is properly partitioned (“keyed”) for subsequent operations.

**When to set explicit Serdes:**
Variants of `groupByKey` exist to override the configured default Serdes of your application, which **you**
**must do** if the key and/or value types of the resulting `KGroupedStream` do not match the configured default
Serdes.

**Grouping vs. Windowing:**
A related operation is [windowing](#streams-developer-guide-dsl-windowing), which lets you control how to
“sub-group” the grouped records *of the same key* into so-called *windows* for stateful operations such as
windowed [aggregations](#streams-developer-guide-dsl-aggregating) or
windowed [joins](#streams-developer-guide-dsl-joins).

**Causes data re-partitioning if and only if the stream was marked for re-partitioning.**
`groupByKey` is preferable to `groupBy` because it re-partitions data only if the stream was already marked
for re-partitioning. However, `groupByKey` does not allow you to modify the key or key type like `groupBy`
does.

```java
KStream stream = ...;

// Group by the existing key, using the application's configured
// default serdes for keys and values.
KGroupedStream groupedStream = stream.groupByKey();

// When the key and/or value types do not match the configured
// default serdes, we must explicitly specify serdes.
KGroupedStream groupedStream = stream.groupByKey(
Grouped.with(
Serdes.ByteArray(), /* key */
Serdes.String()) /* value */
);
``` | | **GroupBy**

- KStream → KGroupedStream
- KTable → KGroupedTable | Groups the records by a *new* key, which may be of a different key type.
When grouping a table, you may also specify a new value and value type.
`groupBy` is a shorthand for `selectKey(...).groupByKey()`.
([KStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#groupBy-org.apache.kafka.streams.kstream.KeyValueMapper-),
[KTable details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KTable.html#groupBy-org.apache.kafka.streams.kstream.KeyValueMapper-))

Grouping is a prerequisite for [aggregating a stream or a table](#streams-developer-guide-dsl-aggregating)
and ensures that data is properly partitioned (“keyed”) for subsequent operations.

**When to set explicit Serdes:**
Variants of `groupBy` exist to override the configured default Serdes of your application, which **you must**
**do** if the key and/or value types of the resulting `KGroupedStream` or `KGroupedTable` do not match the
configured default Serdes.

**Grouping vs. Windowing:**
A related operation is [windowing](#streams-developer-guide-dsl-windowing), which lets you control how to
“sub-group” the grouped records *of the same key* into so-called *windows* for stateful operations such as
windowed [aggregations](#streams-developer-guide-dsl-aggregating) or
windowed [joins](#streams-developer-guide-dsl-joins).

**Always causes data re-partitioning:** `groupBy` always causes data re-partitioning.
If possible use `groupByKey` instead, which will re-partition data only if required.

```java
KStream stream = ...;
KTable table = ...;

// Java examples, using lambda expressions

// Group the stream by a new key and key type
KGroupedStream groupedStream = stream.groupBy(
(key, value) -> value,
Grouped.with(
Serdes.String(), /* key (note: type was modified) */
Serdes.String()) /* value */
);

// Group the table by a new key and key type, and also modify the value and value type.
KGroupedTable groupedTable = table.groupBy(
(key, value) -> KeyValue.pair(value, value.length()),
Grouped.with(
Serdes.String(), /* key (note: type was modified) */
Serdes.Integer()) /* value (note: type was modified) */
);
``` | | **Cogroup**

- KGroupedStream → CogroupedKStream
- CogroupedKStream → CogroupedKStream | Cogrouping enables aggregating multiple input streams in a single operation.
The different (already grouped) input streams must have the same key type and may have different values types.
`KStream#cogroup()` creates a new cogrouped stream with a single input stream, while
`CogroupedKStream#cogroup()` adds a grouped stream to an existing cogrouped stream.

Because each `KGroupedStream` may have a different value type, an individual “adder” aggregator must be
provided via `cogroup()`; those aggregators will be used by the downstream
[aggregate()](#streams-developer-guide-dsl-aggregating) operator.
A `CogroupedKStream` may be [windowed](#streams-developer-guide-dsl-windowing) before it is aggregated.

Cogroup does not cause a repartition as it has the prerequisite that the input streams are grouped.
In the process of creating these groups they will have already been repartitioned if the stream was already
marked for repartitioning.

```java
KGroupedStream groupedStreamOne = ...;
KGroupedStream groupedStreamTwo = ...;

// Java examples, using lambda expressions

// Create new cogroup from the first stream (the value type of the CogroupedKStream is the value type of the final aggregation result)
CogroupedKStream cogroupedStream = groupedStreamOne.cogroup(
(aggKey, newValue, aggValue) -> aggValue + Integer.parseInt(newValue) /* adder for first stream */
);
// Add the second stream to the existing cogroup (note, that the second input stream has a different value type than the first input stream)
cogroupedStream = cogroupedStream.cogroup(
groupedStreamTwo,
(aggKey, newValue, aggValue) -> aggValue + newValue.intValue() /* adder for second stream */
);
// Aggregate all streams of the cogroup
KTable aggregatedTable = cogroupedStream.aggregate(
() -> 0, /* initializer */
Materialized.as("aggregated-stream-store") /* state store name */
.withValueSerde(Serdes.Integer()) /* serde for aggregate value */
);
``` | | **Map**

- KStream → KStream | Takes one record and produces one record. You can modify the record key and value, including their types.
([details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#map-org.apache.kafka.streams.kstream.KeyValueMapper-))

**Marks the stream for data re-partitioning:**
Applying a grouping or a join after `map` will result in re-partitioning of the records.
If possible use `mapValues` instead, which will not cause data re-partitioning.

```java
KStream stream = ...;

// Java example, using lambda expressions
// Note how we change the key and the key type (similar to `selectKey`)
// as well as the value and the value type.
KStream transformed = stream.map(
(key, value) -> KeyValue.pair(value.toLowerCase(), value.length()));
``` | | **Map (values only)**

- KStream → KStream
- KTable → KTable | Takes one record and produces one record, while retaining the key of the original record.
You can modify the record value and the value type.
([KStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#mapValues-org.apache.kafka.streams.kstream.ValueMapper-),
[KTable details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KTable.html#mapValues-org.apache.kafka.streams.kstream.ValueMapper-))

`mapValues` is preferable to `map` because it will not cause data re-partitioning.
However, it does not allow you to modify the key or key type like `map` does.
Note that it is possible though to get read-only access to the input record key
if you use `ValueMapperWithKey` instead of `ValueMapper`.

```java
KStream stream = ...;

// Java example, using lambda expressions
KStream uppercased = stream.mapValues(value -> value.toUpperCase());
``` | | **Merge**

- KStream → KStream | Merges records of two streams into one larger stream.
([details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#merge-org.apache.kafka.streams.kstream.KStream-))

There is no ordering guarantee between records from different streams in the merged stream. Relative order
is preserved within each input stream though (ie, records within the same input stream are processed in order).

```java
KStream stream1 = ...;

KStream stream2 = ...;

KStream merged = stream1.merge(stream2);
``` | | **Peek**

- KStream → KStream | Performs a stateless action on each record, and returns an unchanged stream.
([details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#peek-org.apache.kafka.streams.kstream.ForeachAction-))

You would use `peek` to cause *side effects* based on the input data (similar to `foreach`) and *continue*
*processing* the input data (unlike `foreach`, which is a terminal operation). `peek` returns the input
stream as-is; if you need to modify the input stream, use `map` or `mapValues` instead.

`peek` is helpful for use cases such as logging or tracking metrics or for debugging and troubleshooting.

**Note on processing guarantees:** Any side effects of an action (such as writing to external systems) are not
trackable by Kafka, which means they will typically not benefit from Kafka’s processing guarantees.

```java
KStream stream = ...;

// Java example, using lambda expressions
KStream unmodifiedStream = stream.peek(
(key, value) -> System.out.println("key=" + key + ", value=" + value));
``` | | **Print**

- KStream → void | **Terminal operation.** Prints the records to `System.out` or into a file.
([details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#print-org.apache.kafka.streams.kstream.Printed-))

Calling `print(Printed.toSysOut())` is the same as calling
`foreach((key, value) -> System.out.println(key + ", " + value))`

```java
KStream stream = ...;
// print to sysout
stream.print(Printed.toSysOut());

// print to file with a custom label
stream.print(Printed.toFile("streams.out").withLabel("streams"));
``` | | **Repartition**

- KStream → KStream | Manually trigger repartitioning of the stream with the specified number of partitions.
([details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#repartition--))

Kafka Streams manages the topic for `repartition()`.
The generated topic is treated as an internal topic, so data is purged automatically, as with any other internal
repartition topic. You can specify the number of partitions, which enables scaling downstream sub-topologies in
and out. The repartition operation always triggers repartitioning of the stream, so you can use it with embedded
Processor API methods, like `process()`, that don’t trigger auto repartitioning when a key-changing operation
is performed beforehand.

```java
KStream stream = ... ;

KStream repartitionedStream = stream.repartition(Repartitioned.numberOfPartitions(10));
``` | | **SelectKey**

- KStream → KStream | Assigns a new key – possibly of a new key type – to each record.
([details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#selectKey-org.apache.kafka.streams.kstream.KeyValueMapper-))

Calling `selectKey(mapper)` is the same as calling `map((key, value) -> mapper(key, value), value)`.

**Marks the stream for data re-partitioning:**
Applying a grouping or a join after `selectKey` will result in re-partitioning of the records.

```java
KStream stream = ...;

// Derive a new record key from the record's value. Note how the key type changes, too.
// Java example, using lambda expressions
KStream rekeyed = stream.selectKey((key, value) -> value.split(" ")[0]);
``` | | **Stream to Table**

- KStream → KTable | Convert an event stream into a table or a changelog stream.
([details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#toTable--))

```java
KStream stream = ...;

KTable table = stream.toTable();
``` | | **Table to Stream**

- KTable → KStream | Get the changelog stream of this table.
([details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KTable.html#toStream--))

```java
KTable table = ...;

// Also, a variant of `toStream` exists that allows you
// to select a new key for the resulting stream.
KStream stream = table.toStream();
``` | ### Stateful transformations Stateful transformations depend on state for processing inputs and producing outputs and require a [state store](../architecture.md#streams-architecture-state) associated with the stream processor. For example, in aggregating operations, a windowing state store is used to collect the latest aggregation results per window. In join operations, a windowing state store is used to collect all of the records received so far within the defined window boundary. State stores are fault-tolerant. In case of failure, Kafka Streams guarantees to fully restore all state stores prior to resuming the processing. For more information, see [Fault tolerance](../architecture.md#streams-architecture-fault-tolerance). Available stateful transformations in the DSL include: * [Aggregating](#streams-developer-guide-dsl-aggregating) * [Joining](#streams-developer-guide-dsl-joins) * [Windowing](#streams-developer-guide-dsl-windowing) (as part of aggregations and joins) * [Applying custom processors and transformers](#streams-developer-guide-dsl-process), which may be stateful, for Processor API integration The following diagram shows their relationships:  Here is an example of a stateful application: the WordCount algorithm. WordCount example (see [here](https://github.com/confluentinc/demo-scene/tree/master/kafka-streams-interactive-queries) for the full code): ```java // Assume the record values represent lines of text. For the sake of this example, you can ignore // whatever may be stored in the record keys. KStream textLines = ...; KStream wordCounts = textLines // Split each text line, by whitespace, into words. The text lines are the record // values, that is, you can ignore whatever data is in the record keys and thus invoke // `flatMapValues` instead of the more generic `flatMap`. .flatMapValues(value -> Arrays.asList(value.toLowerCase().split("\\W+"))) // Group the stream by word to ensure the key of the record is the word. .groupBy((key, word) -> word) // Count the occurrences of each word (record key). // // This will change the stream type from `KGroupedStream` to // `KTable` (word -> count). .count() // Convert the `KTable` into a `KStream`. .toStream(); ``` #### Aggregating Aggregation combines records that share a key into a single result, such as a count, sum, or running reduction. After records are [grouped](#streams-developer-guide-dsl-transformations-stateless) by key via `groupByKey` or `groupBy` – and thus represented as either a `KGroupedStream` or a `KGroupedTable`, they can be aggregated via an operation such as `reduce`. Aggregations are key-based operations, which means that they always operate over records (notably record values) of the same key. You can perform aggregations on [windowed](#streams-developer-guide-dsl-windowing) or non-windowed data. #### IMPORTANT To support fault tolerance and avoid undesirable behavior, the initializer and aggregator must be stateless. The aggregation results should be passed in the return value of the initializer and aggregator. Do not use class member variables because that data can potentially get lost in case of failure. | Transformation | Description | |-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | **Aggregate**

- KGroupedStream → KTable
- CogroupedKStream → KTable
- KGroupedTable → KTable | **Rolling aggregation.** Aggregates the values of (non-windowed) records by the grouped key.
Aggregating is a generalization of `reduce` and allows, for example, the aggregate value to have a different
type than the input values.
([KGroupedStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KGroupedStream.html),
([CogroupedKStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/CogroupedKStream.html),
[KGroupedTable details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KGroupedTable.html))

When aggregating a *grouped stream*, you must provide an initializer (e.g., `aggValue = 0`) and an “adder”
aggregator (e.g., `aggValue + curValue`).
When aggregating a *cogrouped stream*, you must only provide an initializer; the corresponding “adder”
aggregators are provided in the prior `cogroup()` calls already.
When aggregating a *grouped table*, you must additionally provide an initializer, “adder”, and “subtractor”
(think: `aggValue - oldValue`).
When aggregating a *cogrouped stream*, the actual aggregators are provided for each input stream in the prior
`cogroup()` calls, so you need to provide only an initializer (e.g., `aggValue=0`).

Several variants of `aggregate` exist, see Javadocs for details.

```java
KGroupedStream groupedStream = ...;
KGroupedTable groupedTable = ...;

// Java examples, using lambda expressions

// Aggregating a KGroupedStream (note how the value type changes from String to Long)
KTable aggregatedStream = groupedStream.aggregate(
() -> 0L, /* initializer */
(aggKey, newValue, aggValue) -> aggValue + newValue.length(), /* adder */
Materialized.as("aggregated-stream-store") /* state store name */
.withValueSerde(Serdes.Long()) /* serde for aggregate value */
);
// Aggregating a KGroupedTable (note how the value type changes from String to Long)
KTable aggregatedTable = groupedTable.aggregate(
() -> 0L, /* initializer */
(aggKey, newValue, aggValue) -> aggValue + newValue.length(), /* adder */
(aggKey, oldValue, aggValue) -> aggValue - oldValue.length(), /* subtractor */
Materialized.as("aggregated-table-store") /* state store name */
.withValueSerde(Serdes.Long()) /* serde for aggregate value */
);
```

Detailed behavior of `KGroupedStream` and `CogroupedKStream`:

- Input records with `null` keys are ignored.
- When a record key is received for the first time, the initializer is called (and called before the adder).
- Whenever a record with a non-`null` value is received, the adder is called.

Detailed behavior of `KGroupedTable`:

- Input records with `null` keys are ignored.
- When a record key is received for the first time, the initializer is called (and called before the adder
and subtractor). Note that, in contrast to `KGroupedStream`, over time the initializer may be called
more than once for a key as a result of having received input tombstone records for that key (see below).
- When the first non-`null` value is received for a key (e.g., INSERT), then only the adder is called.
- When subsequent non-`null` values are received for a key (e.g., UPDATE), then (1) the subtractor is
called with the old value as stored in the table and (2) the adder is called with the new value of the
input record that was just received. The order of execution for the subtractor and adder is not defined.
- When a tombstone record – that is, a record with a `null` value – is received for a key (e.g., DELETE),
then only the subtractor is called. Note that, whenever the subtractor returns a `null` value itself,
then the corresponding key is removed from the resulting `KTable`. If that happens, any next input
record for that key will trigger the initializer again.

See the example at the bottom of this section for a visualization of the aggregation semantics. | | **Aggregate (windowed)**

- KGroupedStream → TimeWindowedStream;
TimeWindowedStream → KTable
- KGroupedStream → SessionWindowedStream;
SessionWindowedStream → KTable
- CogroupedKStream → TimeWindowedCogroupedStream;
TimeWindowedCogroupedStream → KTable
- CogroupedKStream → SessionWindowedCogroupedStream;
SessionWindowedCogroupedStream → KTable | **Windowed aggregation.**
Aggregates the values of records, [per window](#streams-developer-guide-dsl-windowing), by the grouped key.
Aggregating is a generalization of `reduce` and allows, for example, the aggregate value to have a different
type than the input values.
([TimeWindowedKStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/TimeWindowedKStream.html),
[SessionWindowedKStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/SessionWindowedKStream.html),
[TimeWindowedCogroupedKStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/TimeWindowedCogroupedKStream.html),
[SessionWindowedCogroupedKStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/SessionWindowedCogroupedKStream.html))

When aggregating a *grouped stream*, you must provide an initializer (e.g., `aggValue = 0`), “adder”
aggregator (e.g., `aggValue + curValue`), and a window.
When aggregating a *cogrouped stream*, you must only provide an initializer (e.g., `aggValue = 0`) and a
window; the corresponding “adder” aggregators are provided in the prior `cogroup()` calls already.
When windowing is based on sessions, you must additionally provide a “session merger” aggregator
(e.g., `mergedAggValue = leftAggValue + rightAggValue`).

The windowed `aggregate` turns a `TimeWindowedKStream` or `SessionWindowdKStream`
into a windowed `KTable, V>`.

Several variants of `aggregate` exist, see Javadocs for details.

```java
import java.time.Duration;

KGroupedStream groupedStream = ...;
CogroupedKStream cogroupedStream = ...;

// Java examples, using lambda expressions

// Aggregating with time-based windowing (here: with 5-minute tumbling windows)
KTable, Long> timeWindowedAggregatedStream = groupedStream.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
.aggregate(
() -> 0L, /* initializer */
(aggKey, newValue, aggValue) -> aggValue + newValue, /* adder */
Materialized.>as("time-windowed-aggregated-stream-store") /* state store name */
.withValueSerde(Serdes.Long())); /* serde for aggregate value */

// Aggregating with time-based windowing (here: with 5-minute tumbling windows)
// (note: the required "adder" aggregator is specified in the prior `cogroup()` call already)
KTable, Long> timeWindowedAggregatedStream = cogroupedStream.windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)))
.aggregate(
() -> 0L, /* initializer */
Materialized.>as("time-windowed-aggregated-stream-store") /* state store name */
.withValueSerde(Serdes.Long())); /* serde for aggregate value */

// Aggregating with time-based windowing (here: with 5-minute sliding windows and 30-minute grace period)
KTable, Long> timeWindowedAggregatedStream = groupedStream
.windowedBy(SlidingWindows.withTimeDifferenceAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(30)))
.aggregate(
() -> 0L, /* initializer */
(aggKey, newValue, aggValue) -> aggValue + newValue, /* adder */
Materialized.>as("time-windowed-aggregated-stream-store") /* state store name */
.withValueSerde(Serdes.Long())); /* serde for aggregate value */

// Aggregating with time-based windowing (here: with 5-minute sliding windows and 30-minute grace period)
// (note: the required "adder" aggregator is specified in the prior `cogroup()` call already)
KTable, Long> timeWindowedAggregatedStream = cogroupedStream
.windowedBy(SlidingWindows.withTimeDifferenceAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(30)))
.aggregate(
() -> 0L, /* initializer */
Materialized.>as("time-windowed-aggregated-stream-store") /* state store name */
.withValueSerde(Serdes.Long())); /* serde for aggregate value */

// Aggregating with session-based windowing (here: with an inactivity gap of 5 minutes)
KTable, Long> sessionizedAggregatedStream = groupedStream.windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(5)))
.aggregate(
() -> 0L, /* initializer */
(aggKey, newValue, aggValue) -> aggValue + newValue, /* adder */
(aggKey, leftAggValue, rightAggValue) -> leftAggValue + rightAggValue, /* session merger */
Materialized.>as("sessionized-aggregated-stream-store") /* state store name */
.withValueSerde(Serdes.Long())); /* serde for aggregate value */

// Aggregating with session-based windowing (here: with an inactivity gap of 5 minutes)
// (note: the required "adder" aggregator is specified in the prior `cogroup()` call already)
KTable, Long> sessionizedAggregatedStream = cogroupedStream.windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(5)))
.aggregate(
() -> 0L, /* initializer */
(aggKey, leftAggValue, rightAggValue) -> leftAggValue + rightAggValue, /* session merger */
Materialized.>as("sessionized-aggregated-stream-store") /* state store name */
.withValueSerde(Serdes.Long())); /* serde for aggregate value */
```

Detailed behavior:

- The windowed aggregate behaves similar to the rolling aggregate described above. The additional twist is that
the behavior applies *per window*.
- Input records with `null` keys are ignored in general.
- When a record key is received for the first time for a given window, the initializer is called (and called
before the adder).
- Whenever a record with a non-`null` value is received for a given window, the adder is called.
- When using session windows: the session merger is called whenever two sessions are being merged.

See the example at the bottom of this section for a visualization of the aggregation semantics. | | **Count**

- KGroupedStream → KTable
- KGroupedTable → KTable | **Rolling aggregation.** Counts the number of records by the grouped key.
([KGroupedStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KGroupedStream.html),
[KGroupedTable details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KGroupedTable.html))

Several variants of `count` exist, see Javadocs for details.

```java
KGroupedStream groupedStream = ...;
KGroupedTable groupedTable = ...;

// Counting a KGroupedStream
KTable aggregatedStream = groupedStream.count();

// Counting a KGroupedTable
KTable aggregatedTable = groupedTable.count();
```

Detailed behavior for `KGroupedStream`:

- Input records with `null` keys or values are ignored.

Detailed behavior for `KGroupedTable`:

- Input records with `null` keys are ignored. Records with `null` values are not ignored but interpreted
as “tombstones” for the corresponding key, which indicate the deletion of the key from the table. | | **Count (windowed)**

- KGroupedStream → TimeWindowedStream;
TimeWindowedStream → KTable
- KGroupedStream → SessionWindowedStream;
SessionWindowedStream → KTable | **Windowed aggregation.**
Counts the number of records, [per window](#streams-developer-guide-dsl-windowing), by the grouped key.
([TimeWindowedKStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/TimeWindowedKStream.html),
[SessionWindowedKStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/SessionWindowedKStream.html))

The windowed `count` turns a `TimeWindowedKStream` or `SessionWindowedKStream`
into a windowed `KTable, V>`.

Several variants of `count` exist, see Javadocs for details.

```java
import java.time.Duration;

KGroupedStream groupedStream = ...;

// Counting a KGroupedStream with time-based windowing (here: with 5-minute tumbling windows)
KTable, Long> aggregatedStream = groupedStream.windowedBy(
TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5))) /* time-based window */
.count();

// Counting a KGroupedStream with sliding windows time-based windowing (here: with 5-minute sliding windows and 30-minute grace period)
KTable, Long> aggregatedStream = groupedStream.windowedBy(
SlidingWindows.withTimeDifferenceAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(30))) /* time-based window */
.count();

// Counting a KGroupedStream with session-based windowing (here: with 5-minute inactivity gaps)
KTable, Long> aggregatedStream = groupedStream.windowedBy(
SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(5))) /* session window */
.count();
```

Detailed behavior:

- Input records with `null` keys or values are ignored. | | **Reduce**

- KGroupedStream → KTable
- KGroupedTable → KTable | **Rolling aggregation.** Combines the values of (non-windowed) records by the grouped key.
The current record value is combined with the last reduced value, and a new reduced value is returned.
The result value type cannot be changed, unlike `aggregate`.
([KGroupedStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KGroupedStream.html),
[KGroupedTable details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KGroupedTable.html))

When reducing a *grouped stream*, you must provide an “adder” reducer (e.g., `aggValue + curValue`).
When reducing a *grouped table*, you must additionally provide a “subtractor” reducer (e.g.,
`aggValue - oldValue`).

Several variants of `reduce` exist, see Javadocs for details.

```java
KGroupedStream groupedStream = ...;
KGroupedTable groupedTable = ...;

// Java examples, using lambda expressions

// Reducing a KGroupedStream
KTable aggregatedStream = groupedStream.reduce(
(aggValue, newValue) -> aggValue + newValue /* adder */);

// Reducing a KGroupedTable
KTable aggregatedTable = groupedTable.reduce(
(aggValue, newValue) -> aggValue + newValue, /* adder */
(aggValue, oldValue) -> aggValue - oldValue /* subtractor */);
```

Detailed behavior for `KGroupedStream`:

- Input records with `null` keys are ignored in general.
- When a record key is received for the first time, then the value of that record is used as the initial
aggregate value.
- Whenever a record with a non-`null` value is received, the adder is called.

Detailed behavior for `KGroupedTable`:

- Input records with `null` keys are ignored in general.
- When a record key is received for the first time, then the value of that record is used as the initial
aggregate value.
Note that, in contrast to `KGroupedStream`, over time this initialization step may happen more than once
for a key as a result of having received input tombstone records for that key (see below).
- When the first non-`null` value is received for a key (e.g., INSERT), then only the adder is called.
- When subsequent non-`null` values are received for a key (e.g., UPDATE), then (1) the subtractor is
called with the old value as stored in the table and (2) the adder is called with the new value of the
input record that was just received. The order of execution for the subtractor and adder is not defined.
- When a tombstone record – that is, a record with a `null` value – is received for a key (e.g., DELETE),
then only the subtractor is called. Note that, whenever the subtractor returns a `null` value itself,
then the corresponding key is removed from the resulting `KTable`. If that happens, any next input
record for that key will re-initialize its aggregate value.

See the example at the bottom of this section for a visualization of the aggregation semantics. | | **Reduce (windowed)**

- KGroupedStream → TimeWindowedStream;
TimeWindowedStream → KTable
- KGroupedStream → SessionWindowedStream;
SessionWindowedStream → KTable | **Windowed aggregation.**
Combines the values of records, [per window](#streams-developer-guide-dsl-windowing), by the grouped key.
The current record value is combined with the last reduced value, and a new reduced value is returned.
Records with `null` key or value are ignored.
The result value type cannot be changed, unlike `aggregate`.
([TimeWindowedKStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/TimeWindowedKStream.html),
[SessionWindowedKStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/SessionWindowedKStream.html))

The windowed `reduce` turns a `TimeWindowedKStream` or a `SessionWindowedKStream`
into a windowed `KTable, V>`.

Several variants of `reduce` exist, see Javadocs for details.

```java
import java.time.Duration;
KGroupedStream groupedStream = ...;

// Java examples, using lambda expressions

// Aggregating with time-based windowing (here: with 5-minute tumbling windows)
KTable, Long> timeWindowedAggregatedStream = groupedStream.windowedBy(
TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)) /* time-based window */)
.reduce(
(aggValue, newValue) -> aggValue + newValue /* adder */
);

// Aggregating with time-based windowing (here: with 5-minute sliding windows and 30-minute grace period)
KTable, Long> timeWindowedAggregatedStream = groupedStream
.windowedBy(SlidingWindows.withTimeDifferenceAndGrace(Duration.ofMinutes(5), Duration.ofMinutes(30)) /* time-based window */)
.reduce(
(aggValue, newValue) -> aggValue + newValue /* adder */
);

// Aggregating with session-based windowing (here: with an inactivity gap of 5 minutes)
KTable, Long> sessionzedAggregatedStream = groupedStream.windowedBy(
SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(5))) /* session window */
.reduce(
(aggValue, newValue) -> aggValue + newValue /* adder */
);
```

Detailed behavior:

- The windowed reduce behaves similar to the rolling reduce described above. The additional twist is that the
behavior applies *per window*.
- Input records with `null` keys are ignored in general.
- When a record key is received for the first time for a given window, then the value of that record is used as
the initial aggregate value.
- Whenever a record with a non-`null` value is received for a given window, the adder is called.

See the example at the bottom of this section for a visualization of the aggregation semantics. | Example of semantics for stream aggregations : A `KGroupedStream` → `KTable` example is shown below. The streams and the table are initially empty. Bold font is used in the column for “KTable `aggregated`” to highlight changed state. An entry such as `(hello, 1)` denotes a record with key `hello` and value `1`. To improve the readability of the semantics table you can assume that all records are processed in timestamp order. ```java // Key: word, value: count KStream wordCounts = ...; KGroupedStream groupedStream = wordCounts .groupByKey(Grouped.with(Serdes.String(), Serdes.Integer())); KTable aggregated = groupedStream.aggregate( () -> 0, /* initializer */ (aggKey, newValue, aggValue) -> aggValue + newValue, /* adder */ Materialized.>as("aggregated-stream-store" /* state store name */) .withKeySerde(Serdes.String()) /* key serde */ .withValueSerde(Serdes.Integer())); /* serde for aggregate value */ ``` #### WARNING Always start from `Materialized.as("store-name")` and chain `.withKeySerde(...)` or `.withValueSerde(...)` onto it, as shown in the preceding code example. Don’t chain the combined `Materialized#with(keySerde, valueSerde)` method onto `Materialized.as("store-name")`, because `with()` returns a new `Materialized` instance and doesn’t carry over the store name you set with `as()`. For example, `Materialized.as("store-name").with(keySerde, valueSerde)` silently produces an unnamed, generated store name, not a store named “store-name.” Impact of record caches : For illustration purposes, the column “KTable `aggregated`” below shows the table’s state changes over time in a very granular way. In practice, you would observe state changes in such a granular way only when [record caches](memory-mgmt.md#streams-developer-guide-memory-management-record-cache) are disabled (default: enabled).
When record caches are enabled, what might happen, for example, is that the output results of the rows with timestamps 4 and 5 would be [compacted](memory-mgmt.md#streams-developer-guide-memory-management-record-cache), and there would only be a single state update for the key `kafka` in the KTable (here: from `(kafka, 1)` directly to `(kafka, 3)`).
Typically, you should only disable record caches for testing or debugging purposes. Under normal circumstances, it’s better to leave record caches enabled. | | | | KStream `wordCounts` | KGroupedStream `groupedStream` | KTable `aggregated` | |-----------|--------------|--------------|------------------------|----------------------------------|------------------------------------------------------------------------------| | Timestamp | Input record | Grouping | Initializer | Adder | State | | 1 | (hello, 1) | (hello, 1) | 0 (for hello) | (hello, 0 + 1) | **(hello, 1)**

| | 2 | (kafka, 1) | (kafka, 1) | 0 (for kafka) | (kafka, 0 + 1) | (hello, 1)


**(kafka, 1)**

| | 3 | (streams, 1) | (streams, 1) | 0 (for streams) | (streams, 0 + 1) | (hello, 1)


(kafka, 1)


**(streams, 1)**

| | 4 | (kafka, 1) | (kafka, 1) | | (kafka, 1 + 1) | (hello, 1)


(kafka, **2**)


(streams, 1)

| | 5 | (kafka, 1) | (kafka, 1) | | (kafka, 2 + 1) | (hello, 1)


(kafka, **3**)


(streams, 1)

| | 6 | (streams, 1) | (streams, 1) | | (streams, 1 + 1) | (hello, 1)


(kafka, 3)


(streams, **2**)

| Example of semantics for table aggregations : A `KGroupedTable` → `KTable` example is shown below. The tables are initially empty. Bold font is used in the column for “KTable `aggregated`” to highlight changed state. An entry such as `(hello, 1)` denotes a record with key `hello` and value `1`. To improve the readability of the semantics table, you can assume that all records are processed in timestamp order. ```java // Key: username, value: user region (abbreviated to "E" for "Europe", "A" for "Asia") KTable userProfiles = ...; // Re-group `userProfiles`. Don't read too much into what the grouping does: // its prime purpose in this example is to show the *effects* of the grouping // in the subsequent aggregation. KGroupedTable groupedTable = userProfiles .groupBy((user, region) -> KeyValue.pair(region, user.length()), Serdes.String(), Serdes.Integer()); KTable aggregated = groupedTable.aggregate( () -> 0, /* initializer */ (aggKey, newValue, aggValue) -> aggValue + newValue, /* adder */ (aggKey, oldValue, aggValue) -> aggValue - oldValue, /* subtractor */ Materialized.>as("aggregated-table-store" /* state store name */) .withKeySerde(Serdes.String()) /* key serde */ .withValueSerde(Serdes.Integer())); /* serde for aggregate value */ ``` Impact of record caches : For illustration purposes, the column “KTable `aggregated`” below shows the table’s state changes over time in a very granular way. In practice, you would observe state changes in such a granular way only when [record caches](memory-mgmt.md#streams-developer-guide-memory-management-record-cache) are disabled (default: enabled).
When record caches are enabled, what might happen, for example, is that the output results of the rows with timestamps 4 and 5 would be [compacted](memory-mgmt.md#streams-developer-guide-memory-management-record-cache), and there would only be a single state update for the key `kafka` in the KTable (here: from `(kafka, 1)` directly to `(kafka, 3)`).
Typically, you should only disable record caches for testing or debugging purposes. Under normal circumstances, it’s better to leave record caches enabled. | | | | | | KTable `userProfiles` | KGroupedTable `groupedTable` | KTable `aggregated` | |-----------|-----------------|----------------|-----------|-------------|-------------------------|--------------------------------|------------------------------------------------| | Timestamp | Input record | Interpreted as | Grouping | Initializer | Adder | Subtractor | State | | 1 | (alice, E) | INSERT alice | (E, 5) | 0 (for E) | (E, 0 + 5) | | **(E, 5)**

| | 2 | (bob, A) | INSERT bob | (A, 3) | 0 (for A) | (A, 0 + 3) | | **(A, 3)**


(E, 5)

| | 3 | (charlie, A) | INSERT charlie | (A, 7) | | (A, 3 + 7) | | (A, **10**)


(E, 5)

| | 4 | (alice, A) | UPDATE alice | (A, 5) | | (A, 10 + 5) | (E, 5 - 5) | (A, **15**)


(E, **0**)

| | 5 | (charlie, null) | DELETE charlie | (null, 7) | | | (A, 15 - 7) | (A, **8**)


(E, 0)

| | 6 | (null, E) | *ignored* | | | | | (A, 8)


(E, 0)

| | 7 | (bob, E) | UPDATE bob | (E, 3) | | (E, 0 + 3) | (A, 8 - 3) | (A, **5**)


(E, **3**)

| #### Joining A join combines records from two streams or tables that share a key, producing a new record each time a match occurs. Many stream processing applications in practice are coded as streaming joins. For example, applications backing an online shop might need to access multiple, updating database tables (e.g., sales prices, inventory, customer information) in order to enrich a new data record (e.g., customer transaction) with context information. That is, scenarios where you need to perform table lookups at very large scale and with a low processing latency. Here, a popular pattern is to make the information in the databases available in Kafka through so-called *change data capture* in combination with [Kafka’s Connect API](../../connect/index.md#kafka-connect), and then implementing applications that leverage the Streams API to perform [very fast and efficient local joins](https://www.confluent.io/blog/distributed-real-time-joins-and-aggregations-on-user-activity-events-using-kafka-streams/) of such tables and streams, rather than requiring the application to make a query to a remote database over the network for each record. In this example, the KTable concept in Kafka Streams would enable you to track the latest state (e.g., snapshot) of each table in a local state store, thus greatly reducing the processing latency as well as reducing the load of the remote databases when doing such streaming joins. The following join operations are supported, see also the diagram in the [overview section](#streams-developer-guide-dsl-transformations-stateful-overview) of [Stateful Transformations](#streams-developer-guide-dsl-transformations-stateful). Depending on the operands, joins are either [windowed](#streams-developer-guide-dsl-windowing) or non-windowed. | Join operands | Type | (INNER) JOIN | LEFT JOIN | OUTER JOIN | Demo application | |------------------------------|--------------|----------------|---------------|---------------|------------------------------------------------------------------------------------------------------------------------| | KStream-to-KStream | Windowed | Supported | Supported | Supported | [KStream-KStream join](https://developer.confluent.io/confluent-tutorials/joining-stream-stream/ksql/) | | KTable-to-KTable | Non-windowed | Supported | Supported | Supported | [KTable-KTable join](https://developer.confluent.io/confluent-tutorials/joining-table-table/kstreams/) | | KTable-to-KTable Foreign-Key | Non-windowed | Supported | Supported | Not Supported | – | | KStream-to-KTable | Non-windowed | Supported | Supported | Not Supported | [KStream-KTable join](https://developer.confluent.io/confluent-tutorials/joining-stream-table/kstreams/) | | KStream-to-GlobalKTable | Non-windowed | Supported | Supported | Not Supported | [KStream-GlobalKTable join](https://developer.confluent.io/confluent-tutorials/joining-stream-global-ktable/kstreams/) | | KTable-to-GlobalKTable | N/A | Not Supported | Not Supported | Not Supported | – | Each case is explained in more detail in the subsequent sections. ##### Join co-partitioning requirements For equi-joins, input data must be co-partitioned when joining. This ensures that input records with the same key, from both sides of the join, are delivered to the same stream task during processing. **It is your responsibility to ensure data co-partitioning when joining**. Co-partitioning is not required when performing [KTable-KTable Foreign-Key](#streams-developer-guide-dsl-joins-ktable-ktable-foreign-key) joins and [GlobalKTable](../concepts.md#streams-concepts-globalktable) joins. The requirements for data co-partitioning are: * The input topics of the join (left side and right side) must have the **same number of partitions**. * All applications that *write* to the input topics must have the **same partitioning strategy** so that records with the same key are delivered to same partition number. In other words, the keyspace of the input data must be distributed across partitions in the same manner. This means that, for example, applications that use Kafka’s [Java Producer API](../../clients/overview.md#kafka-clients) must use the same partitioner (the producer setting `"partitioner.class"`, that is, `ProducerConfig.PARTITIONER_CLASS_CONFIG`), and applications that use the Kafka’s Streams API must use the same `StreamPartitioner` for operations such as `KStream#to()`. If you use the default partitioner settings across all applications, the partitioning strategy is handled for you. Why is data co-partitioning required? Because [KStream-KStream](#streams-developer-guide-dsl-joins-kstream-kstream), [KTable-KTable](#streams-developer-guide-dsl-joins-ktable-ktable), and [KStream-KTable](#streams-developer-guide-dsl-joins-kstream-ktable) joins are performed based on the keys of records, for example, `leftRecord.key == rightRecord.key`. It is required that the input streams/tables of a join are co-partitioned by key. There are two exceptions in which co-partitioning is not required. : - For [KStream-GlobalKTable](#streams-developer-guide-dsl-joins-kstream-globalktable) joins, co-partitioning is not required because *all* partitions of the `GlobalKTable`’s underlying changelog stream are made available to each `KafkaStreams` instance, so each instance has a full copy of the changelog stream. Further, a `KeyValueMapper` allows for non-key based joins from the `KStream` to the `GlobalKTable`. - [KTable-KTable Foreign-Key](#streams-developer-guide-dsl-joins-ktable-ktable-foreign-key) joins do not require co-partitioning. Kafka Streams internally ensures co-partitioning for Foreign-Key joins. Kafka Streams partly verifies the co-partitioning requirement : During the partition assignment step, that is, at runtime, Kafka Streams verifies whether the number of partitions for both sides of a join are the same. If they’re not, a `TopologyBuilderException` (runtime exception) is being thrown. Note that Kafka Streams can’t verify whether the partitioning strategy matches between the input streams/tables of a join. You must ensure that this is the case. Ensuring data co-partitioning : If the inputs of a join are not co-partitioned yet, you must ensure this manually. You can follow a procedure such as outlined below. To avoid bottlenecks, you should repartition the topic with fewer partitions to match the larger partition number. It’s also possible to repartition the topic with more partitions to match the smaller partition number. For stream-table joins, you should repartition the KStream, because repartitioning a KTable might result in a second state store. For table-table joins, consider the size of the KTables and repartition the smaller KTable. 1. Identify the input KStream or KTable in the join whose underlying Kafka topic has the smaller number of partitions. Call this stream or table “SMALLER”, and the other side of the join “LARGER”. To learn about the number of partitions of a Kafka topic you can use, for example, the CLI tool `bin/kafka-topics` with the `--describe` option. 2. Within your application, re-partition the data of “SMALLER”. You must ensure that, when repartitioning the data with repartition, the same partitioner is used as for “LARGER”. - If “SMALLER” is a KStream: `KStream#repartition(Repartitioned.numberOfPartitions(...))`. - If “SMALLER” is a KTable: `KTable#toStream#repartition(Repartitioned.numberOfPartitions(...).toTable())`. 3. Within your application, perform the join between “LARGER” and the new stream/table. ##### KStream-KStream Join This is a sliding window join, which means that all tuples that are “close” to each other in time – with the time difference up to window size – are joined. The result is a KStream. KStream-KStream joins are always [windowed](#windowing-sliding) joins, because otherwise the size of the internal state store used to perform the join – e.g., a [sliding window](#windowing-sliding) or “buffer” – would grow indefinitely. For stream-stream joins it’s important to highlight that a new input record on one side will produce a join output *for each* matching record on the other side, and there can be *multiple* such matching records in a given join window (cf. the row with timestamp 15 in the join semantics table below, for example). Join output records are effectively created as follows, leveraging the user-supplied `ValueJoiner`: ```java KeyValue leftRecord = ...; KeyValue rightRecord = ...; ValueJoiner joiner = ...; KeyValue joinOutputRecord = KeyValue.pair( leftRecord.key, /* by definition, leftRecord.key == rightRecord.key */ joiner.apply(leftRecord.value, rightRecord.value) ); ``` | Transformation | Description | |-------------------------------------------------------------------------|---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | **Inner Join (windowed)**

- (KStream, KStream)
→ KStream | Performs an INNER JOIN of this stream with another stream.
Even though this operation is windowed, the joined stream will be of type `KStream` rather than `KStream, ...>`.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#join-org.apache.kafka.streams.kstream.KStream-org.apache.kafka.streams.kstream.ValueJoiner-org.apache.kafka.streams.kstream.JoinWindows-)

**Data must be co-partitioned**: The input data for both sides must be [co-partitioned](#streams-developer-guide-dsl-joins-co-partitioning).

**Causes data re-partitioning of a stream if and only if the stream was marked for re-partitioning (if both are marked, both are re-partitioned).**

Several variants of `join` exists, see the Javadocs for details.

```java
import java.time.Duration;

KStream left = ...;
KStream right = ...;

// Java example, using lambda expressions
KStream joined = left.join(right,
(leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, /* ValueJoiner */
JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
Joined.with(
Serdes.String(), /* key */
Serdes.Long(), /* left value */
Serdes.Double()) /* right value */
);
```

Detailed behavior:

- The join is *key-based*, that is, with the join predicate `leftRecord.key == rightRecord.key`, and *window-based*, meaning two input records are joined if and only if their
timestamps are “close” to each other as defined by the user-supplied `JoinWindows`, meaning the window defines an additional join predicate over the record timestamps.
- The join will be triggered under the conditions listed below whenever new input is received. When it is triggered, the user-supplied `ValueJoiner` will be called to produce
join output records.
- Input records with a `null` key or a `null` value are ignored and do not trigger the join.

See the semantics overview at the bottom of this section for a detailed description. | | **Left Join (windowed)**

- (KStream, KStream)
→ KStream | Performs a LEFT JOIN of this stream with another stream.
Even though this operation is windowed, the joined stream will be of type `KStream` rather than `KStream, ...>`.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#leftJoin-org.apache.kafka.streams.kstream.KStream-org.apache.kafka.streams.kstream.ValueJoiner-org.apache.kafka.streams.kstream.JoinWindows-)

**Data must be co-partitioned**: The input data for both sides must be [co-partitioned](#streams-developer-guide-dsl-joins-co-partitioning).

**Causes data re-partitioning of a stream if and only if the stream was marked for re-partitioning (if both are marked, both are re-partitioned).**

Several variants of `leftJoin` exists, see the Javadocs for details.

```java
import java.time.Duration;

KStream left = ...;
KStream right = ...;

// Java example, using lambda expressions
KStream joined = left.leftJoin(right,
(leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, /* ValueJoiner */
JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
Joined.with(
Serdes.String(), /* key */
Serdes.Long(), /* left value */
Serdes.Double()) /* right value */
);
```

Detailed behavior:

- The join is *key-based*, that is, with the join predicate `leftRecord.key == rightRecord.key`, and *window-based*, meaning two input records are joined if and only if their
timestamps are “close” to each other as defined by the user-supplied `JoinWindows`, meaning the window defines an additional join predicate over the record timestamps.
- The join will be triggered under the conditions listed below whenever new input is received. When it is triggered, the user-supplied `ValueJoiner` will be called to produce
join output records.
- Input records with a `null` key or a `null` value are ignored and do not trigger the join.
- For each input record on the left side that does not have any match on the right side, the `ValueJoiner` will be called with `ValueJoiner#apply(leftRecord.value, null)`;
this explains the row with timestamp=60 and timestamp=80 in the table below, which lists `[E, null]` and `[F, null]` in the LEFT JOIN column.
Note that these left results are emitted after the specified grace period passed. **Caution:** Using the deprecated `JoinWindows.of(...).grace(...)` API might result in
eagerly emitted spurious left results.

See the semantics overview at the bottom of this section for a detailed description. | | **Outer Join (windowed)**

- (KStream, KStream)
→ KStream | Performs an OUTER JOIN of this stream with another stream.
Even though this operation is windowed, the joined stream will be of type `KStream` rather than `KStream, ...>`.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#outerJoin-org.apache.kafka.streams.kstream.KStream-org.apache.kafka.streams.kstream.ValueJoiner-org.apache.kafka.streams.kstream.JoinWindows-)

**Data must be co-partitioned**: The input data for both sides must be [co-partitioned](#streams-developer-guide-dsl-joins-co-partitioning).

**Causes data re-partitioning of a stream if and only if the stream was marked for re-partitioning (if both are marked, both are re-partitioned).**

Several variants of `outerJoin` exists, see the Javadocs for details.

```java
import java.time.Duration;

KStream left = ...;
KStream right = ...;

// Java example, using lambda expressions
KStream joined = left.outerJoin(right,
(leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, /* ValueJoiner */
JoinWindows.ofTimeDifferenceWithNoGrace(Duration.ofMinutes(5)),
Joined.with(
Serdes.String(), /* key */
Serdes.Long(), /* left value */
Serdes.Double()) /* right value */
);
```

Detailed behavior:

- The join is *key-based*, that is, with the join predicate `leftRecord.key == rightRecord.key`, and *window-based*, meaning two input records are joined if and only if their
timestamps are “close” to each other as defined by the user-supplied `JoinWindows`, meaning the window defines an additional join predicate over the record timestamps.
- The join will be triggered under the conditions listed below whenever new input is received. When it is triggered, the user-supplied `ValueJoiner` will be called to produce
join output records.
- Input records with a `null` key or a `null` value are ignored and do not trigger the join.
- For each input record on one side that does not have any match on the other side, the `ValueJoiner` will be called with `ValueJoiner#apply(leftRecord.value, null)` or
`ValueJoiner#apply(null, rightRecord.value)`, respectively; this explains the row with timestamp=60, timestamp=80, and timestamp=100 in the table below, which lists
`[E, null]`, `[F, null]`, and `[null, f]` in the OUTER JOIN column.
Note that these left and right results are emitted after the specified grace period passed. **Caution:** Using the deprecated `JoinWindows.of(...).grace(...)` API might
result in eagerly emitted spurious left or right results.

See the semantics overview at the bottom of this section for a detailed description. | Semantics of stream-stream joins : The semantics of the various stream-stream join variants are explained below. To improve the readability of the table, assume that (1) all records have the same key (and thus the key in the table is omitted), (2) all records are processed in timestamp order. A join window size of 15 seconds with a grace period of 5 seconds are assumed.
#### NOTE If you use the old and now-deprecated API to specify the grace period, that is, `JoinWindows.of(...).grace(...)`, left/outer join results are emitted eagerly, and the observed result might differ from the result shown below. The columns INNER JOIN, LEFT JOIN, and OUTER JOIN denote what is passed as arguments to the user-supplied [ValueJoiner](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/ValueJoiner.html) for the `join`, `leftJoin`, and `outerJoin` methods, respectively, whenever a new input record is received on either side of the join. An empty table cell denotes that the `ValueJoiner` is not called at all. The following table shows the output, for each processed input record, for the three join variants. Some input records do not produce output records. | Timestamp | Left (KStream) | Right (KStream) | (INNER) JOIN | LEFT JOIN | OUTER JOIN | |-------------|------------------|-------------------|--------------------------------|--------------------------------|--------------------------------| | 1 | null | | | | | | 2 | | null | | | | | 3 | A | | | | | | 4 | | a | [A, a] | [A, a] | [A, a] | | 5 | B | | [B, a] | [B, a] | [B, a] | | 6 | | b | [A, b], [B, b] | [A, b], [B, b] | [A, b], [B, b] | | 7 | null | | | | | | 8 | | null | | | | | 9 | C | | [C, a], [C, b] | [C, a], [C, b] | [C, a], [C, b] | | 10 | | c | [A, c], [B, c], [C, c] | [A, c], [B, c], [C, c] | [A, c], [B, c], [C, c] | | 11 | | null | | | | | 12 | null | | | | | | 13 | | null | | | | | 14 | | d | [A, d], [B, d], [C, d] | [A, d], [B, d], [C, d] | [A, d], [B, d], [C, d] | | 15 | D | | [D, a], [D, b], [D, c], [D, d] | [D, a], [D, b], [D, c], [D, d] | [D, a], [D, b], [D, c], [D, d] | | … | | | | | | | 40 | E | | | | | | … | | | | | | | 60 | F | | | [E,null] | [E,null] | | … | | | | | | | 80 | | f | | [F,null] | [F,null] | | … | | | | | | | 100 | G | | | | [null,f] | ##### KTable-KTable Join This is a symmetric non-window join. The semantics are a KTable lookup in the “other” stream for each KTable update. The result is a continuously updating KTable, which is a changelog stream that can contain tombstone messages with the format ``. The KTable lookup is done on the current KTable state, so out-of-order records can produce non-deterministic results. KTable-KTable joins are always *non-windowed* joins. They are designed to be consistent with their counterparts in relational databases. The changelog streams of both KTables are materialized into local state stores to represent the latest snapshot of their [table duals](../concepts.md#streams-concepts-ktable). The join result is a new KTable that represents the changelog stream of the join operation. Join output records are effectively created as follows, leveraging the user-supplied `ValueJoiner`: ```java KeyValue leftRecord = ...; KeyValue rightRecord = ...; ValueJoiner joiner = ...; KeyValue joinOutputRecord = KeyValue.pair( leftRecord.key, /* by definition, leftRecord.key == rightRecord.key */ joiner.apply(leftRecord.value, rightRecord.value) ); ``` | Transformation | Description | |-----------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | **Inner Join**

- (KTable, KTable)
→ KTable | Performs an INNER JOIN of this table with another table.
The result is an ever-updating KTable that represents the “current” result of the join.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KTable.html#join-org.apache.kafka.streams.kstream.KTable-org.apache.kafka.streams.kstream.ValueJoiner-)

**Data must be co-partitioned**: The input data for both sides must be [co-partitioned](#streams-developer-guide-dsl-joins-co-partitioning).

```java
KTable left = ...;
KTable right = ...;

// Java example, using lambda expressions
KTable joined = left.join(right,
(leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue /* ValueJoiner */
);
```

Detailed behavior:

- The join is *key-based*, that is, with the join predicate `leftRecord.key == rightRecord.key`.
- The join will be triggered under the conditions listed below whenever new input is received. When it is triggered, the user-supplied `ValueJoiner` will be called to produce
join output records.
- Input records with a `null` key are ignored and do not trigger the join.
- Input records with a `null` value are interpreted as *tombstones* for the corresponding key, which indicate the deletion of the key from the table. Tombstones do not
trigger the join. When an input tombstone is received, then an output tombstone is forwarded directly to the join result KTable if required (that is, only if the corresponding
key actually exists already in the join result KTable).

See the semantics overview at the bottom of this section for a detailed description. | | **Left Join**

- (KTable, KTable)
→ KTable | Performs a LEFT JOIN of this table with another table.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KTable.html#leftJoin-org.apache.kafka.streams.kstream.KTable-org.apache.kafka.streams.kstream.ValueJoiner-)

**Data must be co-partitioned**: The input data for both sides must be [co-partitioned](#streams-developer-guide-dsl-joins-co-partitioning).

```java
KTable left = ...;
KTable right = ...;

// Java example, using lambda expressions
KTable joined = left.leftJoin(right,
(leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue /* ValueJoiner */
);
```

Detailed behavior:

- The join is *key-based*, that is, with the join predicate `leftRecord.key == rightRecord.key`.
- The join will be triggered under the conditions listed below whenever new input is received. When it is triggered, the user-supplied `ValueJoiner` will be called to produce
join output records.
- Input records with a `null` key are ignored and do not trigger the join.
- Input records with a `null` value are interpreted as *tombstones* for the corresponding key, which indicate the deletion of the key from the table. Right-tombstones
trigger the join, but left-tombstones don’t: when an input tombstone is received, an output tombstone is forwarded directly to the join result KTable if required (that is,
only if the corresponding key actually exists already in the join result KTable).
- For each input record on the left side that does not have any match on the right side, the `ValueJoiner` will be called with `ValueJoiner#apply(leftRecord.value, null)`;
this explains the row with timestamp=3 in the table below, which lists `[A, null]` in the LEFT JOIN column.

See the semantics overview at the bottom of this section for a detailed description. | | **Outer Join**

- (KTable, KTable)
→ KTable | Performs an OUTER JOIN of this table with another table.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KTable.html#outerJoin-org.apache.kafka.streams.kstream.KTable-org.apache.kafka.streams.kstream.ValueJoiner-)

**Data must be co-partitioned**: The input data for both sides must be [co-partitioned](#streams-developer-guide-dsl-joins-co-partitioning).

```java
KTable left = ...;
KTable right = ...;

// Java example, using lambda expressions
KTable joined = left.outerJoin(right,
(leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue /* ValueJoiner */
);
```

Detailed behavior:

- The join is *key-based*, that is, with the join predicate `leftRecord.key == rightRecord.key`.
- The join will be triggered under the conditions listed below whenever new input is received. When it is triggered, the user-supplied `ValueJoiner` will be called to produce
join output records.
- Input records with a `null` key are ignored and do not trigger the join.
- Input records with a `null` value are interpreted as *tombstones* for the corresponding key, which indicate the deletion of the key from the table. Tombstones may trigger
joins, depending on the content in the left and right tables. When an input tombstone is received, then an output tombstone is forwarded directly to the join result KTable if
required (that is, only if the corresponding key actually exists already in the join result KTable).
- For each input record on one side that does not have any match on the other side, the `ValueJoiner` will be called with `ValueJoiner#apply(leftRecord.value, null)` or
`ValueJoiner#apply(null, rightRecord.value)`, respectively; this explains the rows with timestamp=3 and timestamp=7 in the table below, which list `[A, null]` and
`[null, b]`, respectively, in the OUTER JOIN column.

See the semantics overview at the bottom of this section for a detailed description. | Semantics of table-table joins : The semantics of the various table-table join variants are explained below. To improve the readability of the table, you can assume that (1) all records have the same key (and thus the key in the table is omitted) and that (2) all records are processed in timestamp order.
The columns INNER JOIN, LEFT JOIN, and OUTER JOIN denote what is passed as arguments to the user-supplied [ValueJoiner](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/ValueJoiner.html) for the `join`, `leftJoin`, and `outerJoin` methods, respectively, whenever a new input record is received on either side of the join. An empty table cell denotes that the `ValueJoiner` is not called at all. In the following table, tombstones are shown as `null (tombstone)` in the result, to contrast with results like `X,null`, which indicate a valid join result with only one join partner. | Timestamp | Left (KTable) | Right (KTable) | (INNER) JOIN | LEFT JOIN | OUTER JOIN | |-------------|------------------|------------------|------------------|------------------|------------------| | 1 | null (tombstone) | | | | | | 2 | | null (tombstone) | | | | | 3 | A | | | [A, null] | [A, null] | | 4 | | a | [A, a] | [A, a] | [A, a] | | 5 | B | | [B, a] | [B, a] | [B, a] | | 6 | | b | [B, b] | [B, b] | [B, b] | | 7 | null (tombstone) | | null (tombstone) | null (tombstone) | [null, b] | | 8 | | null (tombstone) | | | null (tombstone) | | 9 | C | | | [C, null] | [C, null] | | 10 | | c | [C, c] | [C, c] | [C, c] | | 11 | | null (tombstone) | null (tombstone) | [C, null] | [C, null] | | 12 | null (tombstone) | | | null (tombstone) | null (tombstone) | | 13 | | null (tombstone) | | | | | 14 | | d | | | [null, d] | | 15 | D | | [D, d] | [D, d] | [D, d] | | 16 | | | | | | | 17 | | d | [D, d] | [D, d] | [D, d] | ##### KTable-KTable Foreign-Key Join This is a symmetric non-window join. There are two tables involved in this join, the left table and the right table, each of which is usually keyed on different key types. The left table is keyed on the primary key, and the right table is keyed on the foreign key. Each element in the left table has a foreign-key extractor function applied to it, which extracts the foreign key. The resulting left-event is then joined with the right-event keyed on the corresponding foreign-key. Updates made to the right-event also trigger joins with the left-events containing that foreign-key. It can be helpful to think of the left-hand materialized table as events containing a foreign key, and the right-hand materialized table as entities keyed on the foreign key. KTable lookups are done on the current KTable state, so out-of-order records can produce non-deterministic results. KTable-KTable foreign-key joins are always *non-windowed* joins. Foreign-key joins are analogous to joins in SQL. As a rough example: ```sql SELECT ... FROM {this KTable} JOIN {other KTable} ON {other.key} = {result of foreignKeyExtractor(this.value)} ... ``` The output of the operation is a new KTable containing the join result. The changelog streams of both KTables are materialized into local state stores to represent the latest snapshot of their table duals. A foreign-key extractor function is applied to the left record, with a new intermediate record created and is used to lookup and join with the corresponding primary key on the right-hand side table. The result is a new KTable that represents the changelog stream of the join operation. The left KTable can have multiple records which map to the same key on the right KTable. An update to a single left KTable entry may result in a single output event, provided the corresponding key exists in the right KTable. Consequently, a single update to a right KTable entry will result in an update for each record in the left KTable that has the same foreign key. | Transformation | Description | |-----------------------------------------------------------|------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | **Inner Join**

- (KTable, KTable)
→ KTable | Performs a foreign-key INNER JOIN of this table with another table.
The result is an ever-updating KTable that represents the “current” result of the join.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KTable.html#join-org.apache.kafka.streams.kstream.KTable-java.util.function.Function-org.apache.kafka.streams.kstream.ValueJoiner-)

```java
KTable left = ...;
KTable right = ...;
//This foreignKeyExtractor simply uses the left-value to map to the right-key.
Function foreignKeyExtractor = (v) -> v;

//Alternative: with access to left table key
BiFunction foreignKeyExtractor = (k, v) -> v;

// Java example, using lambda expressions
KTable joined = left.join(right,
foreignKeyExtractor,
(leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue /* ValueJoiner */
);
```

Detailed behavior:

- The join is *key-based*, that is, with the join predicate `foreignKeyExtractor.apply(leftRecord.value) == rightRecord.key`.
- The join will be triggered under the conditions listed below whenever new input is received. When it is triggered, the user-supplied `ValueJoiner` will be called to produce
join output records.
- Input records with a `null` value are interpreted as *tombstones* for the corresponding key, which indicate the deletion of the key from the table. Tombstones do not
trigger the join. When an input tombstone is received, then an output tombstone is forwarded directly to the join result KTable if required (that is, only if the corresponding
key actually exists already in the join result KTable).

See the semantics overview at the bottom of this section for a detailed description. | | **Left Join**

- (KTable, KTable)
→ KTable | Performs a foreign-key LEFT JOIN of this table with another table.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KTable.html#leftJoin-org.apache.kafka.streams.kstream.KTable-java.util.function.Function-org.apache.kafka.streams.kstream.ValueJoiner-)

```java
KTable left = ...;
KTable right = ...;
//This foreignKeyExtractor simply uses the left-value to map to the right-key.
Function foreignKeyExtractor = (v) -> v;

//Alternative: with access to left table key
BiFunction foreignKeyExtractor = (k, v) -> v;

// Java example, using lambda expressions
KTable joined = left.leftJoin(right,
foreignKeyExtractor,
(leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue /* ValueJoiner */
);
```

Detailed behavior:

- The join is *key-based*, that is, with the join predicate `foreignKeyExtractor.apply(leftRecord.value) == rightRecord.key`.
- The join will be triggered under the conditions listed below whenever new input is received. When it is triggered, the user-supplied `ValueJoiner` will be called to produce
join output records.
- Records for which the `foreignKeyExtractor` produces `null` are ignored and do not trigger a join. If you want to join with `null` foreign keys, use a suitable sentinel
value to do so (that is, `"NULL"` for a String field, or `-1` for an auto-incrementing integer field).
- Input records with a `null` value are interpreted as *tombstones* for the corresponding key, which indicate the deletion of the key from the table. Right-tombstones
trigger the join, but left-tombstones don’t: when an input tombstone is received, an output tombstone is forwarded directly to the join result KTable if required (that is,
only if the corresponding key actually exists already in the join result KTable).
- For each input record on the left side that does not have any match on the right side, the `ValueJoiner` will be called with `ValueJoiner#apply(leftRecord.value, null)`;
this explains the row with timestamp=7 & 8 in the table below, which lists `(q,10,null)` and `(r,10,null)` in the LEFT JOIN column.

See the semantics overview at the bottom of this section for a detailed description. | Semantics of table-table foreign-key joins : The semantics of the table-table foreign-key INNER and LEFT JOIN variants are demonstrated below. The key is shown alongside the value for each record.
Records are processed in incrementing offset order. The columns INNER JOIN and LEFT JOIN denote what is passed as arguments to the user-supplied `ValueJoiner` for the `join` and `leftJoin` methods, respectively, whenever a new input record is received on either side of the join.
An empty table cell denotes that the `ValueJoiner` is not called at all.
For the purpose of this example, Function `foreignKeyExtractor` simply uses the left-value as the output. | Record Offset | Action | Left KTable (K, extracted-FK) | Right KTable (FK, VR) | (INNER) JOIN | LEFT JOIN | |-----------------|------------------------------|---------------------------------|----------------------------|----------------|---------------| | 1 | Publish event to LHS | (k,1) | (1,foo) | (k,1,foo) | (k,1,foo) | | 2 | Change LHS fk | (k,2) | (1,foo) | (k,null) | (k,2,null) | | 3 | Change LHS fk | (k,3) | (1,foo) | (k,null) | (k,3,null) | | 4 | Publish RHS entity | | (1,foo), (3,bar) | (k,3,bar) | (k,3,bar) | | 5 | Delete k | (k,null) | (1,foo), (3,bar) | (k,null) | (k,null,null) | | 6 | Publish original event again | (k,1) | (1,foo), (3,bar) | (k,1,foo) | (k,1,foo) | | 7 | Publish event to LHS | (q,10) | (1,foo), (3,bar) | | (q,10,null) | | 8 | Publish RHS entity | | (1,foo), (3,bar), (10,baz) | (q,10,baz) | (q,10,baz) | ##### KStream-KTable Join This is an asymmetric non-window join. The semantics are a KTable lookup for each KStream record, while each KTable input record updates the current KTable view but never produces any result record. The result is a KStream. The KTable lookup is done on the current KTable state, so out-of-order records can yield non-deterministic results. KStream-KTable joins are always *non-windowed* joins. They allow you to perform *table lookups* against a KTable (changelog stream) upon receiving a new record from the KStream (record stream). An example use case would be to enrich a stream of user activities (KStream) with the latest user profile information (KTable). Join output records are effectively created as follows, leveraging the user-supplied `ValueJoiner`: ```java KeyValue leftRecord = ...; KeyValue rightRecord = ...; ValueJoiner joiner = ...; KeyValue joinOutputRecord = KeyValue.pair( leftRecord.key, /* by definition, leftRecord.key == rightRecord.key */ joiner.apply(leftRecord.value, rightRecord.value) ); ``` | Transformation | Description | |-------------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | **Inner Join**

- (KStream, KTable)
→ KStream | Performs an INNER JOIN of this stream with the table, effectively doing a table lookup.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#join-org.apache.kafka.streams.kstream.KTable-org.apache.kafka.streams.kstream.ValueJoiner-)

**Data must be co-partitioned**: The input data for both sides must be [co-partitioned](#streams-developer-guide-dsl-joins-co-partitioning).

**Causes data re-partitioning of the stream if and only if the stream was marked for re-partitioning.**

Several variants of `join` exists, see the Javadocs for details.

```java
KStream left = ...;
KTable right = ...;

// Java example, using lambda expressions
KStream joined = left.join(right,
(leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, /* ValueJoiner */
Joined.keySerde(Serdes.String()) /* key */
.withValueSerde(Serdes.Long()) /* left value */
.withGracePeriod(Duration.ZERO) /* grace period */
);
```

Detailed behavior:

- The join is *key-based*, that is, with the join predicate `leftRecord.key == rightRecord.key`.
- The join will be triggered under the conditions listed below whenever new input is received. When it is triggered, the user-supplied `ValueJoiner` will be called to produce
join output records.
- Only input records for the left side (stream) trigger the join. Input records for the right side (table) update only the internal right-side join state.
- Input records for the stream with a `null` value are ignored and do not trigger the join.
- Input records for the table with a `null` value are interpreted as *tombstones* for the corresponding key, which indicate the deletion of the key from the table.
Tombstones do not trigger the join.

When the table is [versioned](#streams-developer-guide-dsl-timestamp-based-semantics), the table record to join with is determined by performing a timestamped lookup, that is, the table
record which is joined will be the latest-by-timestamp record with timestamp less than or equal to the stream record timestamp. If the stream record timestamp is older than the table’s
history retention, then the record is dropped.

To use the grace period, the table needs to be [versioned](#streams-developer-guide-dsl-timestamp-based-semantics). This causes the stream to buffer for the specified grace period
before trying to find a matching record with the right timestamp in the table. The case where the grace period would be used is if a record in the table has a timestamp less than or equal to
the stream record timestamp but arrives after the stream record. If the table record arrives within the grace period the join still occurs. If the table record does not arrive before the
grace period the join continues as normal.

See the semantics overview at the bottom of this section for a detailed description. | | **Left Join**

- (KStream, KTable)
→ KStream | Performs a LEFT JOIN of this stream with the table, effectively doing a table lookup.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#leftJoin-org.apache.kafka.streams.kstream.KTable-org.apache.kafka.streams.kstream.ValueJoiner-)

**Data must be co-partitioned**: The input data for both sides must be [co-partitioned](#streams-developer-guide-dsl-joins-co-partitioning).

**Causes data re-partitioning of the stream if and only if the stream was marked for re-partitioning.**

Several variants of `leftJoin` exists, see the Javadocs for details.

```java
KStream left = ...;
KTable right = ...;

// Java example, using lambda expressions
KStream joined = left.leftJoin(right,
(leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue, /* ValueJoiner */
Joined.keySerde(Serdes.String()) /* key */
.withValueSerde(Serdes.Long()) /* left value */
.withGracePeriod(Duration.ZERO) /* grace period */
);
```

Detailed behavior:

- The join is *key-based*, that is, with the join predicate `leftRecord.key == rightRecord.key`.
- The join will be triggered under the conditions listed below whenever new input is received. When it is triggered, the user-supplied `ValueJoiner` will be called to produce
join output records.
- Only input records for the left side (stream) trigger the join. Input records for the right side (table) update only the internal right-side join state.
- Input records for the stream with a `null` value are ignored and do not trigger the join.
- Input records for the table with a `null` value are interpreted as *tombstones* for the corresponding key, which indicate the deletion of the key from the table.
Tombstones do not trigger the join.
- For each input record on the left side that does not have any match on the right side, the `ValueJoiner` will be called with `ValueJoiner#apply(leftRecord.value, null)`;
this explains the row with timestamp=3 in the table below, which lists `[A, null]` in the LEFT JOIN column.

When the table is [versioned](#streams-developer-guide-dsl-timestamp-based-semantics), the table record to join with is determined by performing a timestamped lookup, that is, the table
record which is joined will be the latest-by-timestamp record with timestamp less than or equal to the stream record timestamp. If the stream record timestamp is older than the table’s
history retention, then the record that is joined will be `null`.

To use the grace period, the table needs to be [versioned](#streams-developer-guide-dsl-timestamp-based-semantics). This causes the stream to buffer for the specified grace period
before trying to find a matching record with the right timestamp in the table. The case where the grace period would be used is if a record in the table has a timestamp less than or equal to
the stream record timestamp but arrives after the stream record. If the table record arrives within the grace period the join still occurs. If the table record does not arrive before the
grace period the join continues as normal.

See the semantics overview at the bottom of this section for a detailed description. | Semantics of stream-table joins : The semantics of the various stream-table join variants are explained below. To improve the readability of the table, assume that:
- All records have the same key, so the key is omitted in the table; - All records are processed in timestamp order.
The columns INNER JOIN and LEFT JOIN denote what is passed as arguments to the user-supplied [ValueJoiner](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/ValueJoiner.html) for the `join` and `leftJoin` methods, respectively, whenever a new input record is received on either side of the join.
An empty table cell denotes that the `ValueJoiner` is not called at all. The following table shows the output, for each processed input record, for both join variants. Some input records do not produce output records. | Timestamp | Left (KStream) | Right (KTable) | (INNER) JOIN | LEFT JOIN | |-------------|------------------|------------------|----------------|-------------| | 1 | null | | | | | 2 | | null (tombstone) | | | | 3 | A | | | [A, null] | | 4 | | a | | | | 5 | B | | [B, a] | [B, a] | | 6 | | b | | | | 7 | null | | | | | 8 | | null (tombstone) | | | | 9 | C | | | [C, null] | | 10 | | c | | | | 11 | | null | | | | 12 | null | | | | | 13 | | null | | | | 14 | | d | | | | 15 | D | | [D, d] | [D, d] | ##### KStream-GlobalKTable Join KStream-GlobalKTable joins are always *non-windowed* joins. They allow you to perform *table lookups* against a [GlobalKTable](../concepts.md#streams-concepts-globalktable) (entire changelog stream) upon receiving a new record from the KStream (record stream). An example use case would be “star queries” or “star joins”, where you would enrich a stream of user activities (KStream) with the latest user profile information (GlobalKTable) and further context information (further GlobalKTables). At a high-level, KStream-GlobalKTable joins are very similar to [KStream-KTable joins](#streams-developer-guide-dsl-joins-kstream-ktable). However, global tables provide you with much more flexibility at the [some expense](../concepts.md#streams-concepts-globalktable) when compared to partitioned tables: * They do not require [data co-partitioning](#streams-developer-guide-dsl-joins-co-partitioning). * They allow for efficient “star joins”; that is, joining a large-scale “facts” stream against “dimension” tables. * They allow for joining against foreign keys; that is, you can look up data in the table not just by the keys of records in the stream, but also by data in the record values. * They make many use cases feasible where you must work on heavily skewed data and thus suffer from hot partitions. * They are often more efficient than their partitioned KTable counterpart when you need to perform multiple joins in succession. Join output records are effectively created as follows, leveraging the user-supplied `ValueJoiner`: ```java KeyValue leftRecord = ...; KeyValue rightRecord = ...; ValueJoiner joiner = ...; KeyValue joinOutputRecord = KeyValue.pair( leftRecord.key, /* by definition, leftRecord.key == rightRecord.key */ joiner.apply(leftRecord.value, rightRecord.value) ); ``` | Transformation | Description | |-------------------------------------------------------------------|----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | **Inner Join**

- (KStream, GlobalKTable)
→ KStream | Performs an INNER JOIN of this stream with the global table, effectively doing a table lookup.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#join-org.apache.kafka.streams.kstream.GlobalKTable-org.apache.kafka.streams.kstream.KeyValueMapper-org.apache.kafka.streams.kstream.ValueJoiner-)

The `GlobalKTable` is fully bootstrapped upon (re)start of a `KafkaStreams` instance, which means the table is fully populated with all the data in the underlying topic that is
available at the time of the startup. The actual data processing begins only after bootstrapping completes.

**Causes data re-partitioning of the stream if and only if the stream was marked for re-partitioning.**

```java
KStream left = ...;
GlobalKTable right = ...;

// Java example, using lambda expressions
KStream joined = left.join(right,
(leftKey, leftValue) -> leftKey.length(), /* derive a (potentially) new key by which to lookup against the table */
(leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue /* ValueJoiner */
);
```

Detailed behavior:

- The join is indirectly *key-based*, that is, with the join predicate `KeyValueMapper#apply(leftRecord.key, leftRecord.value) == rightRecord.key`.
- The join will be triggered under the conditions listed below whenever new input is received. When it is triggered, the user-supplied `ValueJoiner` will be called to produce
join output records.
- Only input records for the left side (stream) trigger the join. Input records for the right side (table) update only the internal right-side join state.
- Input records for the stream with a `null` value are ignored and do not trigger the join.
- Input records for the table with a `null` value are interpreted as *tombstones*, which indicate the deletion of a record key from the table. Tombstones do not trigger the join. | | **Left Join**

- (KStream, GlobalKTable)
→ KStream | Performs a LEFT JOIN of this stream with the global table, effectively doing a table lookup.
[(details)](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#leftJoin-org.apache.kafka.streams.kstream.GlobalKTable-org.apache.kafka.streams.kstream.KeyValueMapper-org.apache.kafka.streams.kstream.ValueJoiner-)

The `GlobalKTable` is fully bootstrapped upon (re)start of a `KafkaStreams` instance, which means the table is fully populated with all the data in the underlying topic that is
available at the time of the startup. The actual data processing begins only after bootstrapping completes.

**Causes data re-partitioning of the stream if and only if the stream was marked for re-partitioning.**

```java
KStream left = ...;
GlobalKTable right = ...;

// Java example, using lambda expressions
KStream joined = left.leftJoin(right,
(leftKey, leftValue) -> leftKey.length(), /* derive a (potentially) new key by which to lookup against the table */
(leftValue, rightValue) -> "left=" + leftValue + ", right=" + rightValue /* ValueJoiner */
);
```

Detailed behavior:

- The join is indirectly *key-based*, that is, with the join predicate `KeyValueMapper#apply(leftRecord.key, leftRecord.value) == rightRecord.key`.
- The join will be triggered under the conditions listed below whenever new input is received. When it is triggered, the user-supplied `ValueJoiner` will be called to produce
join output records.
- Only input records for the left side (stream) trigger the join. Input records for the right side (table) update only the internal right-side join state.
- Input records for the stream with a `null` value are ignored and do not trigger the join.
- Input records for the table with a `null` value are interpreted as *tombstones*, which indicate the deletion of a record key from the table. Tombstones do not trigger the join.
- For each input record on the left side that does not have any match on the right side, the `ValueJoiner` will be called with `ValueJoiner#apply(leftRecord.value, null)`. | Semantics of stream-global-table joins : KStream-GlobalKTable joins have different semantics than KStream-KTable joins.
- Unlike a normal KTable, a global table is fully populated on application startup, effectively ignoring the event-timestamps of its underlying data. - Global table joins don’t synchronize time, unlike joins against a normal KTable. - The left input record is first “mapped” with a user-supplied `KeyValueMapper` into the table’s keyspace prior to the table lookup. #### Windowing Windowing lets you control how to group records that have the same key for stateful operations such as [aggregations](#streams-developer-guide-dsl-aggregating) or [joins](#streams-developer-guide-dsl-joins) into so-called windows. Windows are tracked per record key. For example, in join operations, a windowing state store is used to store all the records received so far within the defined window boundary. In aggregating operations, a windowing state store is used to store the latest aggregation results per window. Old records in the state store are purged after the specified [window retention period](../concepts.md#streams-concepts-windowing). Kafka Streams guarantees to keep a window for at least this specified time; the default value is one day and can be changed via `Materialized#withRetention()`. A related operation is [grouping](#streams-developer-guide-dsl-transformations-stateless), which groups all records that have the same key to ensure that data is properly partitioned (“keyed”) for subsequent operations. Once grouped, windowing allows you to further sub-group the records of a key. The DSL supports the following types of windows: | Window name | Behavior | Short description | |---------------------------------------------|---------------|------------------------------------------------------------------------------------| | [Tumbling time window](#windowing-tumbling) | Time-based | Fixed-size, non-overlapping, gap-less windows | | [Hopping time window](#windowing-hopping) | Time-based | Fixed-size, overlapping windows | | [Sliding time window](#windowing-sliding) | Time-based | Fixed-size, overlapping windows that work on differences between record timestamps | | [Session window](#windowing-session) | Session-based | Dynamically-sized, non-overlapping, data-driven windows | An example of implementing a [custom time window](#streams-custom-window-start-end-times) is provided at the end of this section. ##### Tumbling time windows Tumbling time windows are a special case of hopping time windows and, like the latter, are windows based on time intervals. They model fixed-size, non-overlapping, gap-less windows. A tumbling window is defined by a single property: the window’s *size*. A tumbling window is a hopping window whose window size is equal to its advance interval. Since tumbling windows never overlap, a data record will belong to one and only one window.  Tumbling time windows are *aligned to the epoch*, with the lower interval bound being inclusive and the upper bound being exclusive. “Aligned to the epoch” means that the first window starts at timestamp zero. For example, tumbling windows with a size of 5,000 ms have predictable window boundaries `[0;5000),[5000;10000),...` — and **not** `[1000;6000),[6000;11000),...` or even something “random” like `[1452;6452),[6452;11452),...`. The following code defines a tumbling window with a size of 5 minutes: ```java import java.time.Duration; import org.apache.kafka.streams.kstream.TimeWindows; // A tumbling time window with a size of 5 minutes (and, by definition, an implicit // advance interval of 5 minutes). Duration windowSizeMs = Duration.ofMinutes(5); TimeWindows.ofSizeWithNoGrace(windowSizeMs); // The above is equivalent to the following code: TimeWindows.ofSizeWithNoGrace(windowSizeMs).advanceBy(windowSizeMs); ``` Counting example using tumbling windows: ```java // Key (String) is user ID, value (Avro record) is the page view event for that user. // Such a data stream is often called a "clickstream". KStream pageViews = ...; // Count page views per window, per user, with tumbling windows of size 5 minutes KTable, Long> windowedPageViewCounts = pageViews .groupByKey(Grouped.with(Serdes.String(), genericAvroSerde)) .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5))) .count(); ``` ##### Hopping time windows Hopping time windows are windows based on time intervals. They model fixed-sized, (possibly) overlapping windows. A hopping window is defined by two properties: the window’s *size* and its *advance interval* (aka “hop”). The advance interval specifies by how much a window moves forward relative to the previous one. For example, you can configure a hopping window with a size 5 minutes and an advance interval of 1 minute. Since hopping windows can overlap – and in general they do – a data record may belong to more than one such window. Hopping windows vs. sliding windows : Hopping windows are sometimes called “sliding windows” in other stream processing tools. Kafka Streams follows the terminology in academic literature, where the semantics of sliding windows are different to those of hopping windows. The following code defines a hopping window with a size of 5 minutes and an advance interval of 1 minute: ```java import java.time.Duration; import org.apache.kafka.streams.kstream.TimeWindows; // A hopping time window with a size of 5 minutes and an advance interval of 1 minute. // The window's name -- the string parameter -- is used to e.g. name the backing state store. Duration windowSizeMs = Duration.ofMinutes(5); Duration advanceMs = Duration.ofMinutes(1); TimeWindows.ofSizeWithNoGrace(windowSizeMs).advanceBy(advanceMs); ```  Hopping time windows are *aligned to the epoch*, with the lower interval bound being inclusive and the upper bound being exclusive. “Aligned to the epoch” means that the first window starts at timestamp zero. For example, hopping windows with a size of 5,000 ms and an advance interval (“hop”) of 3,000 ms have predictable window boundaries `[0;5000),[3000;8000),...` — and **not** `[1000;6000),[4000;9000),...` or even something “random” like `[1452;6452),[4452;9452),...`. Counting example using hopping windows: ```java // Key (String) is user ID, value (Avro record) is the page view event for that user. // Such a data stream is often called a "clickstream". KStream pageViews = ...; // Count page views per window, per user, with hopping windows of size 5 minutes that advance every 1 minute KTable, Long> windowedPageViewCounts = pageViews .groupByKey(Grouped.with(Serdes.String(), genericAvroSerde)) .windowedBy(TimeWindows.ofSizeWithNoGrace(Duration.ofMinutes(5)).advanceBy(Duration.ofMinutes(1))) .count(); ``` Unlike the non-windowed aggregates described previously, windowed aggregates return a *windowed KTable* whose key type is `Windowed`. This is to differentiate aggregate values with the same key from different windows. The corresponding window instance and the embedded key can be retrieved as `Windowed#window()` and `Windowed#key()`, respectively. ##### Sliding time windows Sliding windows differ significantly from hopping and tumbling windows. In Kafka Streams, sliding windows are used only for [join operations](#streams-developer-guide-dsl-joins), and are specified by using the `JoinWindows` class, and windowed aggregations, specified by using the `SlidingWindows` class.  A sliding window models a fixed-size window that slides continuously over the time axis. In this model, two data records are said to be included in the same window if (in the case of symmetric windows) the difference of their timestamps is within the window size. As a sliding window moves along the time axis, records can fall into multiple snapshots of the sliding window, but each unique combination of records appears only in one sliding window snapshot. Sliding windows *require* that you set a grace period, as shown below. For time windows and session windows, setting the grace period is optional and defaults to 24 hours. Sliding windows are aligned to the data record timestamps, not to the epoch. In contrast to hopping and tumbling windows, the lower and upper window time interval bounds of sliding windows are *both inclusive*. The following code example defines a sliding window with a time difference of 10 minutes and a grace period of 30 minutes. ```java import org.apache.kafka.streams.kstream.SlidingWindows; // A sliding time window with a time difference of 10 minutes Duration windowTimeDifference = Duration.ofMinutes(10); Duration grace = Duration.ofMinutes(30); SlidingWindows.withTimeDifferenceAndGrace(windowTimeDifference, grace); ``` ##### Session Windows Session windows aggregate key-based events into so-called *sessions*, a process called *sessionization*. Sessions represent a **period of activity** separated by a defined **gap of inactivity** (or “idleness”). Any events processed that fall within the inactivity gap of any existing sessions are merged into the existing sessions. If an event falls outside of the session gap, then a new session is created. Session windows are different from the other window types in that: - all windows are tracked independently across keys – for example, windows of different keys typically have different start and end times - their window sizes vary – even windows for the same key typically have different sizes The prime area of application for session windows is **user behavior analysis**. Session-based analyses can range from simple metrics, for example, count of user visits on a news website or social platform, to more complex metrics, for example, customer conversion funnel and event flows. The following code defines a session window with an inactivity gap of 5 minutes: ```java import java.time.Duration; import org.apache.kafka.streams.kstream.SessionWindows; // A session window with an inactivity gap of 5 minutes. SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(5)); ``` Given the previous session window example, here is what would happen on an input stream of six records. When the first three records arrive (upper part in the diagram below), you have three sessions (see lower part) after having processed those records: two for the green record key, with one session starting and ending at the 0-minute mark (only due to the illustration it looks as if the session goes from 0 to 1), and another starting and ending at the 6-minute mark; and one session for the blue record key, starting and ending at the 2-minute mark.  If you then receive three additional records (including two out-of-order records), what would happen is that the two existing sessions for the green record key are merged into a single session starting at time 0 and ending at time 6, consisting of a total of three records. The existing session for the blue record key is extended to end at time 5, consisting of a total of two records. And, finally, there is a new session for the blue key starting and ending at time 11.  Counting example using session windows : Assume you want to analyze reader behavior on a news website, like *The New York Times*, given a session definition of “As long as a person views (clicks on) another page at least once every 5 minutes (= inactivity gap), consider this to be a single visit and so a single, contiguous reading session for that person.” What you want to compute from this stream of input data is the number of page views per session. ```java // Key (String) is user ID, value (Avro record) is the page view event for that user. // Such a data stream is often called a "clickstream". KStream pageViews = ...; // Count page views per session, per user, with session windows that have an inactivity gap of 5 minutes KTable, Long> sessionizedPageViewCounts = pageViews .groupByKey(Grouped.with(Serdes.String(), genericAvroSerde)) .windowedBy(SessionWindows.ofInactivityGapWithNoGrace(Duration.ofMinutes(5))) .count(); ``` ##### Window Final Results In Kafka Streams, windowed computations update their results continuously. As new data arrives for a window, freshly computed results are emitted downstream. For many applications, this is ideal, since fresh results are always available, and Kafka Streams is designed to make programming continuous computations seamless. However, some applications need to take action **only** on the final result of a windowed computation. Common examples of this are sending alerts or delivering results to a system that doesn’t support updates. Suppose that you have an hourly windowed count of events per user. If you want to send an alert when a user has *less than* three events in an hour, you have a real challenge. All users would match this condition at first, until they accrue enough events, so you can’t simply send an alert when someone matches the condition; you have to wait until you know you won’t see any more events for a particular window, and *then* send the alert. Kafka Streams offers a clean way to define this logic: after defining your windowed computation, you can `suppress` the intermediate results, emitting the final count for each user when the window is **closed**. For example: ```java KGroupedStream grouped = ...; grouped .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofHours(1), Duration.ofMinutes(10))) .count() .suppress(Suppressed.untilWindowCloses(unbounded())) .filter((windowedUserId, count) -> count < 3) .toStream() .foreach((windowedUserId, count) -> sendAlert(windowedUserId.window(), windowedUserId.key(), count)); ``` The key parts of this program are: `grace(Duration.ofMinutes(10))` : This allows you to bound how out-of-order (delayed) events can be before the window stops accepting them. For example, the 09:00 to 10:00 window accepts out-of-order records until 10:10, at which point the window is **closed**. `.suppress(Suppressed.untilWindowCloses(...))` : This configures the suppression operator to emit nothing for a window until it closes, and then emit the final result. For example, if user `U` gets 10 events between 09:00 and 10:10, the `filter` downstream of the suppression will get no events for the windowed key `@09:00-10:00` until 10:10, and then it will get exactly one event with the value `10`. This is the final result of the windowed count. `unbounded()` : This configures the buffer used for storing events until their windows close. Production code is able to put a cap on the amount of memory to use for the buffer, but this simple example creates a buffer with no upper bound. One thing to note is that suppression is like any other Kafka Streams operator, so you can build a topology with two branches emerging from the `count`, one suppressed, and one not, or even multiple differently configured suppressions. This enables you to apply suppressions where they are needed and otherwise rely on the default continuous update behavior. For more detailed information, see the Javadoc on the `Suppressed` config object and [KIP-328](https://cwiki.apache.org/confluence/x/sQU0BQ). ##### Emit strategies for windowed aggregations The `suppress` operator described in the preceding section solves the “final results only” problem, but it buffers records in memory and doesn’t support RocksDB. As an alternative, you can use `TimeWindowedKStream#emitStrategy()` and `SessionWindowedKStream#emitStrategy()` to control when a windowed aggregation emits results. These methods use the aggregation’s own state store instead of a separate suppression buffer. Pass one of these `EmitStrategy` instances to `emitStrategy()`: - `EmitStrategy.onWindowUpdate()` (default): Emit a result every time the window is updated. This is the continuous-update behavior described earlier in this section. - `EmitStrategy.onWindowClose()`: Emit a result only after the window closes, that is, after stream time passes the window’s end time plus its grace period. This achieves the same “final results only” outcome as `suppress(Suppressed.untilWindowCloses(...))`, but works with RocksDB state stores and doesn’t require an additional in-memory buffer. ```java KGroupedStream groupedByWord = ...; groupedByWord .windowedBy(TimeWindows.ofSizeAndGrace(Duration.ofHours(1), Duration.ofMinutes(10))) .emitStrategy(EmitStrategy.onWindowClose()) .count() .toStream() .foreach((windowedUserId, count) -> sendAlert(windowedUserId.window(), windowedUserId.key(), count)); ``` `EmitStrategy.onWindowClose()` can be used only with windows that close, that is, `TimeWindows`, `SlidingWindows`, and `SessionWindows`. It can’t be used with `UnlimitedWindows`. For more information about controlling when aggregated results are produced, see [KIP-825](https://cwiki.apache.org/confluence/display/KAFKA/KIP-825%3A+introduce+a+new+API+to+control+when+aggregated+results+are+produced). ##### Window duration and joins After you change the window duration in a join operation, the state store segments are no longer valid, because the segment ranges and their corresponding IDs change. Although this situation would resolve eventually when existing segments are dropped after a full grace-period range, your application misses join results until this happens. If you don’t want to reset your internal topics after you change window duration, you can erase RocksDB and let it rebuild from the changelog. This approach won’t give you data older than the original grace period, but it does cause existing data joins to emit correctly. ##### Example: Custom time window In addition to using the windows implementations provided with the Kafka Streams client library, you can extend the [Java Windows abstract class](https://github.com/apache/kafka/blob/trunk/streams/src/main/java/org/apache/kafka/streams/kstream/Windows.java) to create custom time windows to suit your use cases. To view a custom implementation of a daily window starting every day at 6pm, see [streams/window example](https://github.com/confluentinc/kafka-streams-examples/tree/latest/src/test/java/io/confluent/examples/streams/window/). The example also shows a potential problem in dealing with time zones that have [daylight saving time](https://en.wikipedia.org/wiki/Daylight_saving_time). ### Applying processors (Processor API integration) Beyond the aforementioned [stateless](#streams-developer-guide-dsl-transformations-stateless) and [stateful](#streams-developer-guide-dsl-transformations-stateful) transformations, you can also leverage the [Processor API](processor-api.md#streams-developer-guide-processor-api) from the DSL. There are a number of scenarios where this might be helpful: - **Customization:** You need to implement special, customized logic that is not or not yet available in the DSL. - **Combining ease-of-use with full flexibility where it’s needed:** Even though you generally prefer to use the expressiveness of the DSL, there are certain steps in your processing that require more flexibility and tinkering than the DSL provides. For example, only the Processor API provides access to a [record’s metadata](../faq.md#streams-faq-processing-record-metadata) such as its topic, partition, and offset information. However, you don’t want to switch completely to the Processor API because of that. - **Migrating from other tools:** You are migrating from other stream processing technologies that provide an imperative API, and migrating some of your legacy code to the Processor API was faster and easier than to migrate completely to the DSL right away. #### Operations and concepts - `KStream#process`: Process all records in a stream, one record at a time, by applying a `Processor` (provided by a given `ProcessorSupplier`); - `KStream#processValues`: Process all records in a stream, one record at a time, by applying a `FixedKeyProcessor` (provided by a given `FixedKeyProcessorSupplier`); - `Processor`: A processor of key-value pair records; - `ContextualProcessor`: An abstract implementation of `Processor` that manages the `ProcessorContext` instance; - `FixedKeyProcessor`: A processor of key-value pair records where keys are immutable; - `ContextualFixedKeyProcessor`: An abstract implementation of `FixedKeyProcessor` that manages the `FixedKeyProcessorContext` instance; - `ProcessorSupplier`: A processor supplier that can create one or more `Processor` instances; and - `FixedKeyProcessorSupplier`: A processor supplier that can create one or more `FixedKeyProcessor` instances. #### Examples The following examples show how to apply `process` and `processValues` to your Kafka Streams application. | Example Operation State Type | | | |---------------------------------------------------------------------------------------------------------|---------------|-----------| | [Categorize logs by severity](#streams-developer-guide-dsl-process-categorize-logs) | process | Stateless | | [Cumulative discounts for a loyalty program](#streams-developer-guide-dsl-process-cumulative-discounts) | process | Stateful | | [Replace slang in text messages](#streams-developer-guide-dsl-process-replace-slang) | processValues | Stateless | | [Traffic radar monitoring car count](#streams-developer-guide-dsl-process-car-count) | processValues | Stateful | ##### Categorize logs by severity - Idea: You have a stream of log messages. Each message contains a severity level, for example, INFO, WARN, ERROR, in the value. The processor filters messages, routing ERROR messages to a dedicated topic and discarding INFO messages. The rest (WARN) are forwarded to a dedicated topic too. - Real-world context: In a production monitoring system, categorizing logs by severity ensures ERROR logs are sent to a critical incident management system, WARN logs are analyzed for potential risks, and INFO logs are stored for basic reporting purposes. ```java public class CategorizingLogsBySeverityExample { private static final String ERROR_LOGS_TOPIC = "error-logs-topic"; private static final String INPUT_LOGS_TOPIC = "input-logs-topic"; private static final String UNKNOWN_LOGS_TOPIC = "unknown-logs-topic"; private static final String WARN_LOGS_TOPIC = "warn-logs-topic"; public static void categorizeWithProcess(final StreamsBuilder builder) { final KStream logStream = builder.stream(INPUT_LOGS_TOPIC); logStream.process(LogSeverityProcessor::new) .to((key, value, recordContext) -> { // Determine the target topic dynamically if ("ERROR".equals(key)) return ERROR_LOGS_TOPIC; if ("WARN".equals(key)) return WARN_LOGS_TOPIC; return UNKNOWN_LOGS_TOPIC; }); } private static class LogSeverityProcessor extends ContextualProcessor { @Override public void process(final Record record) { if (record.value() == null) { return; // Skip null values } // Assume the severity is the first word in the log message // For example: "ERROR: Disk not found" -> "ERROR" final int colonIndex = record.value().indexOf(':'); final String severity = colonIndex > 0 ? record.value().substring(0, colonIndex).trim() : "UNKNOWN"; // Route logs based on severity switch (severity) { case "ERROR": context().forward(record.withKey(ERROR_LOGS_TOPIC)); break; case "WARN": context().forward(record.withKey(WARN_LOGS_TOPIC)); break; case "INFO": // INFO logs are ignored break; default: // Forward to an "unknown" topic for logs with unrecognized severities context().forward(record.withKey(UNKNOWN_LOGS_TOPIC)); } } } } ``` ##### Cumulative discounts for a loyalty program - Idea: A stream of purchase events contains user IDs and transaction amounts. Use a state store to accumulate the total spending of each user. When their total crosses a threshold, apply a discount on their next transaction and update their accumulated total. - Real-world context: In a retail loyalty program, tracking cumulative customer spending enables dynamic rewards, such as issuing a discount when a customer’s total purchases exceed a predefined limit. ```java public class CumulativeDiscountsForALoyaltyProgramExample { private static final double DISCOUNT_THRESHOLD = 100.0; private static final String CUSTOMER_SPENDING_STORE = "customer-spending-store"; private static final String DISCOUNT_NOTIFICATION_MESSAGE = "Discount applied! You have received a reward for your purchases."; private static final String DISCOUNT_NOTIFICATIONS_TOPIC = "discount-notifications-topic"; private static final String PURCHASE_EVENTS_TOPIC = "purchase-events-topic"; public static void applyDiscountWithProcess(final StreamsBuilder builder) { // Define the state store for tracking cumulative spending builder.addStateStore( Stores.keyValueStoreBuilder( Stores.inMemoryKeyValueStore(CUSTOMER_SPENDING_STORE), Serdes.String(), Serdes.Double() ) ); final KStream purchaseStream = builder.stream(PURCHASE_EVENTS_TOPIC); // Apply the Processor with the state store final KStream notificationStream = purchaseStream.process(CumulativeDiscountProcessor::new, CUSTOMER_SPENDING_STORE); // Send the notifications to the output topic notificationStream.to(DISCOUNT_NOTIFICATIONS_TOPIC); } private static class CumulativeDiscountProcessor implements Processor { private KeyValueStore spendingStore; private ProcessorContext context; @Override public void init(final ProcessorContext context) { this.context = context; // Retrieve the state store for cumulative spending spendingStore = context.getStateStore(CUSTOMER_SPENDING_STORE); } @Override public void process(final Record record) { if (record.value() == null) { return; // Skip null purchase amounts } // Get the current spending total for the customer Double currentSpending = spendingStore.get(record.key()); if (currentSpending == null) { currentSpending = 0.0; } // Update the cumulative spending currentSpending += record.value(); spendingStore.put(record.key(), currentSpending); // Check if the customer qualifies for a discount if (currentSpending >= DISCOUNT_THRESHOLD) { // Reset the spending after applying the discount spendingStore.put(record.key(), currentSpending - DISCOUNT_THRESHOLD); // Send a discount notification context.forward(record.withValue(DISCOUNT_NOTIFICATION_MESSAGE)); } } } } ``` ##### Replace slang in text messages - Idea: A messaging stream contains user-generated content, and you want to replace slang words with their formal equivalents, for example, “u” becomes “you”, “brb” becomes “be right back”. The operation modifies only the message value and keeps the key intact. - Real-world context: In customer support chat systems, normalizing text by replacing slang with formal equivalents ensures that automated sentiment analysis tools work accurately and provide reliable insights. ```java public class ReplacingSlangTextInMessagesExample { private static final Map SLANG_DICTIONARY = Map.of( "u", "you", "brb", "be right back", "omg", "oh my god", "btw", "by the way" ); private static final String INPUT_MESSAGES_TOPIC = "input-messages-topic"; private static final String OUTPUT_MESSAGES_TOPIC = "output-messages-topic"; public static void replaceWithProcessValues(final StreamsBuilder builder) { KStream messageStream = builder.stream(INPUT_MESSAGES_TOPIC); messageStream.processValues(SlangReplacementProcessor::new).to(OUTPUT_MESSAGES_TOPIC); } private static class SlangReplacementProcessor extends ContextualFixedKeyProcessor { @Override public void process(final FixedKeyRecord record) { if (record.value() == null) { return; // Skip null values } // Replace slang words in the message final String[] words = record.value().split("\\s+"); for (final String word : words) { String replacedWord = SLANG_DICTIONARY.getOrDefault(word, word); context().forward(record.withValue(replacedWord)); } } } } ``` ##### Traffic radar monitoring car count - Idea: A radar monitors cars passing along a road stretch. A system counts the cars for each day, maintaining a cumulative total for the current day in a state store. At the end of the day, the count is emitted and the state is cleared for the next day. - Real-world context: A car counting system can be useful for determining measures for widening or controlling traffic depending on the number of cars passing through the monitored stretch. ```java public class TrafficRadarMonitoringCarCountExample { private static final String DAILY_COUNT_STORE = "price-state-store"; private static final String DAILY_COUNT_TOPIC = "price-state-topic"; private static final String RADAR_COUNT_TOPIC = "car-radar-topic"; public static void countWithProcessValues(final StreamsBuilder builder) { // Define a state store for tracking daily car counts builder.addStateStore( Stores.keyValueStoreBuilder( Stores.inMemoryKeyValueStore(DAILY_COUNT_STORE), Serdes.String(), Serdes.Long() ) ); final KStream radarStream = builder.stream(RADAR_COUNT_TOPIC); // Apply the FixedKeyProcessor with the state store radarStream.processValues(DailyCarCountProcessor::new, DAILY_COUNT_STORE) .to(DAILY_COUNT_TOPIC); } private static class DailyCarCountProcessor implements FixedKeyProcessor { private FixedKeyProcessorContext context; private KeyValueStore stateStore; private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd").withZone(ZoneId.systemDefault()); @Override public void init(final FixedKeyProcessorContext context) { this.context = context; stateStore = context.getStateStore(DAILY_COUNT_STORE); } @Override public void process(final FixedKeyRecord record) { if (record.value() == null) { return; // Skip null events } // Derive the current day from the event timestamp final long timestamp = System.currentTimeMillis(); // Use system time for simplicity final String currentDay = DATE_FORMATTER.format(Instant.ofEpochMilli(timestamp)); // Retrieve the current count for the day Long dailyCount = stateStore.get(currentDay); if (dailyCount == null) { dailyCount = 0L; } // Increment the count dailyCount++; stateStore.put(currentDay, dailyCount); // Emit the current day's count context.forward(record.withValue(String.format("Day: %s, Car Count: %s", currentDay, dailyCount))); } } } ``` ##### Key points - Type safety and flexibility: The `process` and `processValues` APIs utilize `ProcessorContext` and `Record` or `FixedKeyRecord` objects for better type safety and flexibility of custom processing logic. - Clear state and logic management: Implementations for `Processor` or `FixedKeyProcessor` should manage state and logic clearly. Use `context().forward()` for emitting records downstream. - Unified API: Consolidates multiple methods into a single, versatile API. - Future-proof: Ensures compatibility with the latest Kafka Streams releases. ### Transformers removal and migration to processors As of Confluent Platform 8.0 (Kafka 4.0), several deprecated methods in the Kafka Streams API, such as `transform`, `flatTransform`, `transformValues`, `flatTransformValues`, and `process` have been removed. These methods have been replaced with the more versatile Processor API. This guide provides detailed steps for migrating existing code to use the new Processor API and explains the benefits of the changes. The following deprecated methods are no longer available in Kafka Streams: - `KStream#transform` - `KStream#flatTransform` - `KStream#transformValues` - `KStream#flatTransformValues` - `KStream#process` The Processor API now serves as a unified replacement for all these methods. It simplifies the API surface while maintaining support for both stateless and stateful operations. #### Migration examples To migrate from the deprecated `transform`, `transformValues`, `flatTransform`, and `flatTransformValues` methods to the Processor API (PAPI) in Kafka Streams, this section resumes the previous examples. The new `process` and `processValues` methods enable a more flexible and reusable approach by requiring implementations of the `Processor` or `FixedKeyProcessor` interfaces. #### IMPORTANT If you are using `KStream.transformValues()` or `KStream.flatTransformValues()`, and you have the “merge repartition topics” optimization enabled, rewriting your program to `KStream.processValues()` might not be safe, due to [KAFKA-19668](https://issues.apache.org/jira/browse/KAFKA-19668). For this case, do not upgrade to Confluent Platform 8.0.0. Instead, upgrade directly to Confluent Platform 8.0.1, which contains a fix. If you’re upgrading to Confluent Platform 8.1.x, no extra caution is needed for this issue: Confluent Platform 8.1.0 already includes the fix, unlike the equivalent Kafka Streams 4.1.0 release, which requires 4.1.1. For backward compatibility reasons, the fix is not enabled by default. To enable it, set `TopologyConfig.InternalConfig.ENABLE_PROCESS_PROCESSVALUE_FIX` to `true` and pass the configuration to the `StreamsBuilder` constructor through a `TopologyConfig`, as shown in the following example. ```java final Properties properties = new Properties(); properties.put(StreamsConfig.APPLICATION_ID_CONFIG, ...); properties.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, ...); properties.put(TopologyConfig.InternalConfig.ENABLE_PROCESS_PROCESSVALUE_FIX, true); final StreamsBuilder builder = new StreamsBuilder(new TopologyConfig(new StreamsConfig(properties))); ``` You should compare the output of `Topology.describe()` for the old and new topology, to verify if the rewrite to `processValues()` is correct and that it does not introduce any incompatibilities. Also, you should test the upgrade in a non-production environment. | Example Migrating from Migrating | to State Type | | | |---------------------------------------------------------------------------------------------------------|-----------------------|-----------------|-----------| | [Categorize logs by severity](#streams-developer-guide-dsl-migrate-categorize-logs) | `flatTransform` | `process` | Stateless | | [Cumulative discounts for a loyalty program](#streams-developer-guide-dsl-migrate-cumulative-discounts) | `transform` | `process` | Stateful | | [Replace slang in text messages](#streams-developer-guide-dsl-migrate-replace-slang) | `flatTransformValues` | `processValues` | Stateless | | [Traffic radar monitoring car count](#streams-developer-guide-dsl-migrate-car-count) | `transformValues` | `processValues` | Stateful | ##### Categorize logs by severity In the following code example, the `categorizeWithFlatTransform` and `categorizeWithProcess` methods show how you can migrate from `flatTransform` to `process`. ```java public class CategorizingLogsBySeverityExample { private static final String ERROR_LOGS_TOPIC = "error-logs-topic"; private static final String INPUT_LOGS_TOPIC = "input-logs-topic"; private static final String UNKNOWN_LOGS_TOPIC = "unknown-logs-topic"; private static final String WARN_LOGS_TOPIC = "warn-logs-topic"; public static void categorizeWithFlatTransform(final StreamsBuilder builder) { final KStream logStream = builder.stream(INPUT_LOGS_TOPIC); logStream.flatTransform(LogSeverityTransformer::new) .to((key, value, recordContext) -> { // Determine the target topic dynamically if ("ERROR".equals(key)) return ERROR_LOGS_TOPIC; if ("WARN".equals(key)) return WARN_LOGS_TOPIC; return UNKNOWN_LOGS_TOPIC; }); } public static void categorizeWithProcess(final StreamsBuilder builder) { final KStream logStream = builder.stream(INPUT_LOGS_TOPIC); logStream.process(LogSeverityProcessor::new) .to((key, value, recordContext) -> { // Determine the target topic dynamically if ("ERROR".equals(key)) return ERROR_LOGS_TOPIC; if ("WARN".equals(key)) return WARN_LOGS_TOPIC; return UNKNOWN_LOGS_TOPIC; }); } private static class LogSeverityTransformer implements Transformer>> { @Override public void init(org.apache.kafka.streams.processor.ProcessorContext context) { } @Override public Iterable> transform(String key, String value) { if (value == null) { return Collections.emptyList(); // Skip null values } // Assume the severity is the first word in the log message // For example: "ERROR: Disk not found" -> "ERROR" int colonIndex = value.indexOf(':'); String severity = colonIndex > 0 ? value.substring(0, colonIndex).trim() : "UNKNOWN"; // Create appropriate KeyValue pair based on severity return switch (severity) { case "ERROR" -> List.of(new KeyValue<>("ERROR", value)); case "WARN" -> List.of(new KeyValue<>("WARN", value)); case "INFO" -> Collections.emptyList(); // INFO logs are ignored default -> List.of(new KeyValue<>("UNKNOWN", value)); }; } @Override public void close() { } } private static class LogSeverityProcessor extends ContextualProcessor { @Override public void process(final Record record) { if (record.value() == null) { return; // Skip null values } // Assume the severity is the first word in the log message // For example: "ERROR: Disk not found" -> "ERROR" final int colonIndex = record.value().indexOf(':'); final String severity = colonIndex > 0 ? record.value().substring(0, colonIndex).trim() : "UNKNOWN"; // Route logs based on severity switch (severity) { case "ERROR": context().forward(record.withKey(ERROR_LOGS_TOPIC)); break; case "WARN": context().forward(record.withKey(WARN_LOGS_TOPIC)); break; case "INFO": // INFO logs are ignored break; default: // Forward to an "unknown" topic for logs with unrecognized severities context().forward(record.withKey(UNKNOWN_LOGS_TOPIC)); } } } } ``` ##### Cumulative discounts for a loyalty program - Idea: A stream of purchase events contains user IDs and transaction amounts. Use a state store to accumulate the total spending of each user. When their total crosses a threshold, apply a discount on their next transaction and update their accumulated total. - Real-world context: In a retail loyalty program, tracking cumulative customer spending enables dynamic rewards, such as issuing a discount when a customer’s total purchases exceed a predefined limit. In the following code example, the `applyDiscountWithTransform` and `applyDiscountWithProcess` methods show how you can migrate from `transform` to `process`. ```java public class CumulativeDiscountsForALoyaltyProgramExample { private static final double DISCOUNT_THRESHOLD = 100.0; private static final String CUSTOMER_SPENDING_STORE = "customer-spending-store"; private static final String DISCOUNT_NOTIFICATION_MESSAGE = "Discount applied! You have received a reward for your purchases."; private static final String DISCOUNT_NOTIFICATIONS_TOPIC = "discount-notifications-topic"; private static final String PURCHASE_EVENTS_TOPIC = "purchase-events-topic"; public static void applyDiscountWithTransform(final StreamsBuilder builder) { // Define the state store for tracking cumulative spending builder.addStateStore( Stores.keyValueStoreBuilder( Stores.inMemoryKeyValueStore(CUSTOMER_SPENDING_STORE), Serdes.String(), Serdes.Double() ) ); final KStream purchaseStream = builder.stream(PURCHASE_EVENTS_TOPIC); // Apply the Transformer with the state store final KStream notificationStream = purchaseStream.transform(CumulativeDiscountTransformer::new, CUSTOMER_SPENDING_STORE); // Send the notifications to the output topic notificationStream.to(DISCOUNT_NOTIFICATIONS_TOPIC); } public static void applyDiscountWithProcess(final StreamsBuilder builder) { // Define the state store for tracking cumulative spending builder.addStateStore( Stores.keyValueStoreBuilder( Stores.inMemoryKeyValueStore(CUSTOMER_SPENDING_STORE), org.apache.kafka.common.serialization.Serdes.String(), org.apache.kafka.common.serialization.Serdes.Double() ) ); final KStream purchaseStream = builder.stream(PURCHASE_EVENTS_TOPIC); // Apply the Processor with the state store final KStream notificationStream = purchaseStream.process(CumulativeDiscountProcessor::new, CUSTOMER_SPENDING_STORE); // Send the notifications to the output topic notificationStream.to(DISCOUNT_NOTIFICATIONS_TOPIC); } private static class CumulativeDiscountTransformer implements Transformer> { private KeyValueStore spendingStore; @Override public void init(final org.apache.kafka.streams.processor.ProcessorContext context) { // Retrieve the state store for cumulative spending spendingStore = context.getStateStore(CUSTOMER_SPENDING_STORE); } @Override public KeyValue transform(final String key, final Double value) { if (value == null) { return null; // Skip null purchase amounts } // Get the current spending total for the customer Double currentSpending = spendingStore.get(key); if (currentSpending == null) { currentSpending = 0.0; } // Update the cumulative spending currentSpending += value; spendingStore.put(key, currentSpending); // Check if the customer qualifies for a discount if (currentSpending >= DISCOUNT_THRESHOLD) { // Reset the spending after applying the discount spendingStore.put(key, currentSpending - DISCOUNT_THRESHOLD); // Return a notification message return new KeyValue<>(key, DISCOUNT_NOTIFICATION_MESSAGE); } return null; // No discount, so no output for this record } @Override public void close() { } } private static class CumulativeDiscountProcessor implements Processor { private KeyValueStore spendingStore; private ProcessorContext context; @Override public void init(final ProcessorContext context) { this.context = context; // Retrieve the state store for cumulative spending spendingStore = context.getStateStore(CUSTOMER_SPENDING_STORE); } @Override public void process(final Record record) { if (record.value() == null) { return; // Skip null purchase amounts } // Get the current spending total for the customer Double currentSpending = spendingStore.get(record.key()); if (currentSpending == null) { currentSpending = 0.0; } // Update the cumulative spending currentSpending += record.value(); spendingStore.put(record.key(), currentSpending); // Check if the customer qualifies for a discount if (currentSpending >= DISCOUNT_THRESHOLD) { // Reset the spending after applying the discount spendingStore.put(record.key(), currentSpending - DISCOUNT_THRESHOLD); // Send a discount notification context.forward(record.withValue(DISCOUNT_NOTIFICATION_MESSAGE)); } } } } ``` ##### Replace slang in text messages In the following code example, the `replaceWithFlatTransformValues` and `replaceWithProcessValues` methods show how you can migrate from `flatTransformValues` to `processValues`. ```java public class ReplacingSlangTextInMessagesExample { private static final Map SLANG_DICTIONARY = Map.of( "u", "you", "brb", "be right back", "omg", "oh my god", "btw", "by the way" ); private static final String INPUT_MESSAGES_TOPIC = "input-messages-topic"; private static final String OUTPUT_MESSAGES_TOPIC = "output-messages-topic"; public static void replaceWithFlatTransformValues(final StreamsBuilder builder) { KStream messageStream = builder.stream(INPUT_MESSAGES_TOPIC); messageStream.flatTransformValues(SlangReplacementTransformer::new).to(OUTPUT_MESSAGES_TOPIC); } public static void replaceWithProcessValues(final StreamsBuilder builder) { KStream messageStream = builder.stream(INPUT_MESSAGES_TOPIC); messageStream.processValues(SlangReplacementProcessor::new).to(OUTPUT_MESSAGES_TOPIC); } private static class SlangReplacementTransformer implements ValueTransformer> { @Override public void init(final org.apache.kafka.streams.processor.ProcessorContext context) { } @Override public Iterable transform(final String value) { if (value == null) { return Collections.emptyList(); // Skip null values } // Replace slang words in the message final String[] words = value.split("\\s+"); return Arrays.asList( Arrays.stream(words) .map(word -> SLANG_DICTIONARY.getOrDefault(word, word)) .toArray(String[]::new) ); } @Override public void close() { } } private static class SlangReplacementProcessor extends ContextualFixedKeyProcessor { @Override public void process(final FixedKeyRecord record) { if (record.value() == null) { return; // Skip null values } // Replace slang words in the message final String[] words = record.value().split("\\s+"); for (final String word : words) { String replacedWord = SLANG_DICTIONARY.getOrDefault(word, word); context().forward(record.withValue(replacedWord)); } } } } ``` ##### Traffic radar monitoring car count In the following code example, the `countWithTransformValues` and `countWithProcessValues` methods show how you can migrate from `transformValues` to `processValues`. ```java public class TrafficRadarMonitoringCarCountExample { private static final String DAILY_COUNT_STORE = "price-state-store"; private static final String DAILY_COUNT_TOPIC = "price-state-topic"; private static final String RADAR_COUNT_TOPIC = "car-radar-topic"; public static void countWithTransformValues(final StreamsBuilder builder) { // Define a state store for tracking daily car counts builder.addStateStore( Stores.keyValueStoreBuilder( Stores.inMemoryKeyValueStore(DAILY_COUNT_STORE), org.apache.kafka.common.serialization.Serdes.String(), org.apache.kafka.common.serialization.Serdes.Long() ) ); final KStream radarStream = builder.stream(RADAR_COUNT_TOPIC); // Apply the ValueTransformer with the state store radarStream.transformValues(DailyCarCountTransformer::new, DAILY_COUNT_STORE) .to(DAILY_COUNT_TOPIC); } public static void countWithProcessValues(final StreamsBuilder builder) { // Define a state store for tracking daily car counts builder.addStateStore( Stores.keyValueStoreBuilder( Stores.inMemoryKeyValueStore(DAILY_COUNT_STORE), org.apache.kafka.common.serialization.Serdes.String(), org.apache.kafka.common.serialization.Serdes.Long() ) ); final KStream radarStream = builder.stream(RADAR_COUNT_TOPIC); // Apply the FixedKeyProcessor with the state store radarStream.processValues(DailyCarCountProcessor::new, DAILY_COUNT_STORE) .to(DAILY_COUNT_TOPIC); } private static class DailyCarCountTransformer implements ValueTransformerWithKey { private KeyValueStore stateStore; private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd").withZone(ZoneId.systemDefault()); @Override public void init(final org.apache.kafka.streams.processor.ProcessorContext context) { // Access the state store stateStore = context.getStateStore(DAILY_COUNT_STORE); } @Override public String transform(Void readOnlyKey, String value) { if (value == null) { return null; // Skip null events } // Derive the current day from the event timestamp final long timestamp = System.currentTimeMillis(); // Use system time for simplicity final String currentDay = DATE_FORMATTER.format(Instant.ofEpochMilli(timestamp)); // Retrieve the current count for the day Long dailyCount = stateStore.get(currentDay); if (dailyCount == null) { dailyCount = 0L; } // Increment the count dailyCount++; stateStore.put(currentDay, dailyCount); // Return the current day's count return String.format("Day: %s, Car Count: %s", currentDay, dailyCount); } @Override public void close() { } } private static class DailyCarCountProcessor implements FixedKeyProcessor { private FixedKeyProcessorContext context; private KeyValueStore stateStore; private static final DateTimeFormatter DATE_FORMATTER = DateTimeFormatter.ofPattern("yyyy-MM-dd").withZone(ZoneId.systemDefault()); @Override public void init(final FixedKeyProcessorContext context) { this.context = context; stateStore = context.getStateStore(DAILY_COUNT_STORE); } @Override public void process(final FixedKeyRecord record) { if (record.value() == null) { return; // Skip null events } // Derive the current day from the event timestamp final long timestamp = System.currentTimeMillis(); // Use system time for simplicity final String currentDay = DATE_FORMATTER.format(Instant.ofEpochMilli(timestamp)); // Retrieve the current count for the day Long dailyCount = stateStore.get(currentDay); if (dailyCount == null) { dailyCount = 0L; } // Increment the count dailyCount++; stateStore.put(currentDay, dailyCount); // Emit the current day's count context.forward(record.withValue(String.format("Day: %s, Car Count: %s", currentDay, dailyCount))); } } } ``` ### Removal of old `process` method In addition to the methods mentioned previously, the `process` method, which integrated the ‘old’ Processor API, that is, `Processor` as opposed to the new `api.Processor`, into the DSL, has also been removed. The following example shows how to migrate to the new `process`. #### Example - Idea: The system monitors page views for a website in real-time. When a page reaches a predefined popularity threshold, for example, 1000 views, the system sends an email alert automatically to the site administrator or marketing team to notify them of the page’s success. This helps teams quickly identify high-performing content and act on it, such as promoting the page further or analyzing the traffic source. - Real-world context: In a content management system (CMS) for a news or blogging platform, it’s crucial to track the popularity of articles or posts. For example: - Marketing teams: Use the notification to highlight trending content on social media or email newsletters. - Operations teams: Use the alert to ensure the site can handle increased traffic for popular pages. - Ad managers: Identify pages where additional ad placements might maximize revenue. By automating the detection of popular pages, the system eliminates the need for manual monitoring and ensures timely actions to capitalize on the content’s performance. ```java public class PopularPageEmailAlertExample { private static final String ALERTS_EMAIL = "alerts@yourcompany.com"; private static final String PAGE_VIEWS_TOPIC = "page-views-topic"; public static void alertWithOldProcess(StreamsBuilder builder) { KStream pageViews = builder.stream(PAGE_VIEWS_TOPIC); // Filter pages with exactly 1000 views and process them using the old API pageViews.filter((pageId, viewCount) -> viewCount == 1000) .process(PopularPageEmailAlertOld::new); } public static void alertWithNewProcess(StreamsBuilder builder) { KStream pageViews = builder.stream(PAGE_VIEWS_TOPIC); // Filter pages with exactly 1000 views and process them using the new API pageViews.filter((pageId, viewCount) -> viewCount == 1000) .process(PopularPageEmailAlertNew::new); } private static class PopularPageEmailAlertOld extends AbstractProcessor { @Override public void init(org.apache.kafka.streams.processor.ProcessorContext context) { super.init(context); System.out.println("Initialized email client for: " + ALERTS_EMAIL); } @Override public void process(String key, Long value) { if (value == null) return; if (value == 1000) { // Send an email alert System.out.printf("ALERT (Old API): Page %s has reached 1000 views. Sending email to %s%n", key, ALERTS_EMAIL); } } @Override public void close() { System.out.println("Tearing down email client for: " + ALERTS_EMAIL); } } private static class PopularPageEmailAlertNew implements Processor { @Override public void init(ProcessorContext context) { System.out.println("Initialized email client for: " + ALERTS_EMAIL); } @Override public void process(Record record) { if (record.value() == null) return; if (record.value() == 1000) { // Send an email alert System.out.printf("ALERT (New API): Page %s has reached 1000 views. Sending email to %s%n", record.key(), ALERTS_EMAIL); } } @Override public void close() { System.out.println("Tearing down email client for: " + ALERTS_EMAIL); } } } ``` ## Name operators in a Kafka Streams DSL application Kafka Streams enables you to [name processors](dsl-topology-naming.md#streams-developer-dsl-topology-naming) created by using the Streams DSL. ## Control KTable emit rate A KTable is logically a continuously updated table. These updates make their way to downstream operators whenever new data is available, ensuring that the whole computation is as fresh as possible. Most programs describe a series of logical transformations, and the update rate is not a factor in the program behavior. In these cases, the rate of update is a performance concern, which is best addressed directly through the relevant configurations. However, for some applications, the rate of update itself is an important semantic property. Rather than achieving this as a side-effect of the [record caches](memory-mgmt.md#streams-developer-guide-memory-management-record-cache), you can directly impose a rate limit through the `KTable#suppress` operator. For example: ```java KGroupedTable groupedTable = ...; groupedTable .count() .suppress(untilTimeLimit(Duration.ofMinutes(5), maxBytes(1_000_000L).emitEarlyWhenFull())) .toStream(); ``` This configuration ensures that, downstream of `suppress`, each key is updated no more than once every 5 minutes (in stream time, not wall-clock time). Note that the latest state for each key has to be buffered in memory for that 5-minute period. You have the option to control the maximum amount of memory to use for this buffer (in this case, 1 MB). There is also an option to impose a limit in terms of number of records or to leave both limits unspecified. Additionally, it is possible to choose what happens if the buffer fills up. This example takes a relaxed approach and emits the oldest records before their 5-minute time limit to bring the buffer back down to size. Alternatively, you can choose to stop processing and shut the application down. This might seem extreme, but it gives you a guarantee that the 5-minute time limit is absolutely enforced. After the application shuts down, you could allocate more memory for the buffer and resume processing. Emitting early is preferable for most applications. For more detailed information, see the Javadoc on the `Suppressed` config object and [KIP-328](https://cwiki.apache.org/confluence/x/sQU0BQ). ## Timestamp-based semantics for table processors By default, tables in Kafka Streams use offset-based semantics. When multiple records arrive for the same key, the one with the largest record offset is considered the latest record for the key and is the record that appears in aggregation and join results computed on the table. This is true even in the event of [out-of-order data](../concepts.md#streams-concepts-out-out-order-handling). The record with the largest offset is considered to be the latest record for the key, even if this record does not have the largest timestamp. An alternative to offset-based semantics is timestamp-based semantics. With timestamp-based semantics, the record with the largest timestamp is considered the latest record, even if there is another record with a larger offset (and smaller timestamp). If there is no out-of-order data (per key), then offset-based semantics and timestamp-based semantics are equivalent; the difference appears only when there is out-of-order data. Starting with Confluent Platform 7.5 (Kafka Streams 3.5), Kafka Streams supports timestamp-based semantics through the use of [versioned state stores](processor-api.md#streams-developer-guide-versioned-state-stores). When a table is materialized with a versioned state store, it is a versioned table and results in different processor semantics in the presence of out-of-order data. - When performing a stream-table join, stream-side records join with the latest-by-timestamp table record which has a timestamp less than or equal to the stream record’s timestamp. This is in contrast to joining a stream to an unversioned table, in which case the latest-by-offset table record is joined, even if the stream-side record is out-of-order and has a lower timestamp. - Aggregations computed on the table include the latest-by-timestamp record for each key, instead of the latest-by-offset record. Out-of-order updates (per key) don’t trigger a new aggregation result. This is true for `count` and `reduce` operations as well, in addition to `aggregate` operations. - Table joins use the latest-by-timestamp record for each key, instead of the latest-by-offset record. Out-of-order updates (per key) don’t trigger a new join result. This is true for both primary-key table-table joins and also foreign-key table-table joins. If a versioned table is joined with an unversioned table, the result is the join of the latest-by-timestamp record from the versioned table with the latest-by-offset record from the unversioned table. - Table filter operations no longer suppress consecutive tombstones, so you may observe more `null` records downstream of the filter than when you filter an unversioned table. This is done to preserve a complete version history downstream, in the event of out-of-order data. Once a table is materialized with a versioned store, downstream tables are also considered versioned until any of the following occurs: - A downstream table is materialized explicitly, either with an unversioned store supplier or with no store supplier. All stores are unversioned by default, including the default store supplier. - Any stateful transformation occurs, including aggregations and joins. - A table is converted to a stream and back. The results of certain processors should not be materialized with versioned stores, as these processors don’t produce a complete older version history and therefore materialization as a versioned table leads to unpredictable results: - Aggregate processors, for both table and stream aggregations. This includes `aggregate`, `count`, and `reduce` operations. - Table-table join processors, including both primary-key and foreign-key joins. For more information, see [Versioned key-value state stores](processor-api.md#streams-developer-guide-versioned-state-stores). ## Writing streams back to Kafka Any streams and tables may be (continuously) written back to a Kafka topic. As described in more detail below, the output data might be re-partitioned on its way to Kafka, depending on the situation. | | Writing to Kafka | Description | |-----------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | **To**

- KStream -> void | **Terminal operation.** Write the records to Kafka topic(s).
([KStream details](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/kstream/KStream.html#to(java.lang.String)))

When to provide serdes explicitly:

- If you do not specify Serdes explicitly, the default Serdes from the
[configuration](config-streams.md#streams-developer-guide-configuration) are used.
- You **must specify Serdes explicitly** via the `Produced` class if the key and/or value types of the
`KStream` do not match the configured default Serdes.
- See [Kafka Streams Data Types and Serialization for Confluent Platform](datatypes.md#streams-developer-guide-serdes) for information about configuring default Serdes, available Serdes,
and implementing your own custom Serdes.

A variant of `to` exists that enables you to specify how the data is produced by using a `Produced`
instance to specify, for example, a `StreamPartitioner` that gives you control over
how output records are distributed across the partitions of the output topic.

Another variant of `to` enables you to dynamically choose which topic to send to for each record
via a `TopicNameExtractor` instance.

```java
KStream stream = ...;
KTable table = ...;

// Write the stream to the output topic, using the configured default key
// and value serdes of your `StreamsConfig`.
stream.to("my-stream-output-topic");

// Write the stream to the output topic, using explicit key and value serdes,
// (thus overriding the defaults of your `StreamsConfig`).
stream.to("my-stream-output-topic", Produced.with(Serdes.String(), Serdes.Long()));

// Write the stream to the output topics. The topic name is determined dynamically for
// each record; also use explicit stream partitioner to determine which partition
// of the topic to send to.
stream.to(
(key, value, recordContext) -> { // topicNameExtractor
if (myPattern.matcher(key).matches()) {
return "special-stream-output-topic";
} else {
return "normal-stream-output-topic";
}
},
Produced.streamPartitioner(
(topic, key, value, numPartitions) -> {
if (topic.equals("special-stream-output-topic")) {
return specialHash(key, value, numPartitions);
} else {
return md5Hash(key, value, numPartitions);
}
}
)
);
```

**Causes data re-partitioning if any of the following conditions is true:**

1. If the output topic has a different number of partitions than the stream/table.
2. If the `KStream` was marked for re-partitioning.
3. If you provide a custom `StreamPartitioner` to explicitly control how to distribute the output records
across the partitions of the output topic.
4. If the key of an output record is `null`. | When you want to write to systems other than Kafka : Beside writing the data back to Kafka, you can also apply a [custom processor](#streams-developer-guide-dsl-process) as a stream sink at the end of the processing, for example, to write to external databases.
This is not a preferred pattern, and Confluent suggests using the [Kafka Connect API](../../connect/index.md#kafka-connect) instead. But if you do use such a sink processor, be aware that it’s your responsibility to guarantee message delivery semantics when communicating with such external systems, for example, to retry on delivery failure or to prevent message duplication. ## Test a Streams application Kafka Streams comes with a `test-utils` module to help you test your application. For more information, see [Test a Streams Application](test-streams.md#streams-developer-testing). ## Kafka Streams DSL for Scala #### Deprecated Deprecated since version 8.3: The `kafka-streams-scala` module is deprecated in Confluent Platform 8.3 (Kafka Streams 4.3) and might be removed in a future release. New Kafka Streams applications should use the Java DSL directly. For migration guidance, see [KIP-1244 in the upgrade guide](../upgrade-guide.md#streams-upgrade-guide-kip-1244). Kafka Streams provides a Scala wrapper for the Java API to provide: 1. Better type inference in Scala. 2. Less boilerplate in application code. 3. The usual builder-style composition that developers get with the original Java API. 4. Implicit serializers and de-serializers leading to better abstraction and less verbosity. 5. Better type safety during compile time. All functionality provided by Kafka Streams DSL for Scala is under the root package name of `org.apache.kafka.streams.scala`. Kafka Streams wraps many of the public-facing types from the Java API. The following Scala abstractions are available to you: - `org.apache.kafka.streams.scala.StreamsBuilder` - `org.apache.kafka.streams.scala.kstream.KStream` - `org.apache.kafka.streams.scala.kstream.KTable` - `org.apache.kafka.streams.scala.kstream.KGroupedStream` - `org.apache.kafka.streams.scala.kstream.KGroupedTable` - `org.apache.kafka.streams.scala.kstream.SessionWindowedKStream` - `org.apache.kafka.streams.scala.kstream.TimeWindowedKStream` The library also has several utility abstractions and modules that the user needs to use for proper semantics. - `org.apache.kafka.streams.scala.ImplicitConversions`: Class that brings into scope the implicit conversions between the Scala and Java classes. - `org.apache.kafka.streams.scala.Serdes`: Class that contains core Serdes that can be imported as implicits and a helper to create custom Serdes. (see [Implicit Serdes](#streams-developer-guide-dsl-scala-implicit-serdes)) The library is cross-built with two Scala versions. To reference the library compiled against Scala 2.13, add the following in your maven `pom.xml`: ```xml org.apache.kafka kafka-streams-scala_2.13 8.3.1-ccs ``` When using SBT then you can reference the correct library using the following: ```none libraryDependencies += "org.apache.kafka" %% "kafka-streams-scala" % "4.3.0" ``` ### Example usage The library works by wrapping the original Java abstractions of Kafka Streams within a Scala wrapper object. All the Scala abstractions are named identically as the corresponding Java abstraction, but they reside in a different package of the library. For example, the Scala class `org.apache.kafka.streams.scala.StreamsBuilder` is a wrapper around `org.apache.kafka.streams.StreamsBuilder`, `org.apache.kafka.streams.scala.kstream.KStream` is a wrapper around `org.apache.kafka.streams.kstream.KStream`, and so on. The net result is that the following code is structured just like using the Java API, but with fewer type annotations compared to using the Java API directly from Scala. The difference in type annotation usage is more obvious when given an example. Here is an example of the classic WordCount program that uses the Scala `StreamsBuilder` that builds an instance of `KStream` which is a wrapper around Java `KStream`. Then the stream is converted to a table, producing a `KTable`, which, again is a wrapper around Java `KTable`. ```scala import java.time.Duration import java.util.Properties import org.apache.kafka.streams.kstream.Materialized import org.apache.kafka.streams.scala.ImplicitConversions._ import org.apache.kafka.streams.scala._ import org.apache.kafka.streams.scala.kstream._ import org.apache.kafka.streams.{KafkaStreams, StreamsConfig} object WordCountApplication extends App { import Serdes._ val props: Properties = { val p = new Properties() p.put(StreamsConfig.APPLICATION_ID_CONFIG, "wordcount-application") p.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "kafka-broker1:9092") p } val builder: StreamsBuilder = new StreamsBuilder val textLines: KStream[String, String] = builder.stream[String, String]("TextLinesTopic") val wordCounts: KTable[String, Long] = textLines .flatMapValues(textLine => textLine.toLowerCase.split("\\W+")) .groupBy((_, word) => word) .count(Materialized.as("counts-store")) wordCounts.toStream.to("WordsWithCountsTopic") val streams: KafkaStreams = new KafkaStreams(builder.build(), props) streams.start() sys.ShutdownHookThread { streams.close(Duration.ofSeconds(10)) } } ``` In the preceding code snippet, you don’t have to provide any Serdes, `Grouped`, `Produced`, `Consumed` or `Joined` explicitly. They will also not be dependent on any Serdes specified in the config. **In fact all Serdes specified in the config will be ignored by the Scala APIs**. All Serdes and `Grouped`, `Produced`, `Consumed` or `Joined` will be handled through implicit Serdes as discussed later in the [Implicit Serdes](#streams-developer-guide-dsl-scala-implicit-serdes) section. The complete independence from configuration based Serdes is what makes this library completely typesafe. Any missing instances of Serdes, `Grouped`, `Produced`, `Consumed` or `Joined` will be flagged as a compile time error. ### Implicit Serdes The library uses the power of [Scala implicit parameters](https://docs.scala-lang.org/tour/implicit-parameters.html) to avoid repetitively having to specify Serdes throughout the topology. As a user you can provide implicit Serdes or implicit values of `Grouped`, `Produced`, `Repartitioned`, `Consumed`, or `Joined` once and make your code less verbose. The library also bundles all implicit Serdes of the commonly used types in `org.apache.kafka.streams.scala.Serdes`. Importing this class’s members removes the need to specify serdes for any standard data type. Here’s an example: ```scala // Serdes brings into scope pre-defined implicit Serdes // that will set up all Grouped, Produced, Consumed and Joined instances. // So all APIs below that accept Grouped, Produced, Consumed or Joined will // get these instances automatically import Serdes._ import org.apache.kafka.streams.scala.Serdes._ import org.apache.kafka.streams.scala.ImplicitConversions._ val builder = new StreamsBuilder() val userClicksStream: KStream[String, Long] = builder.stream(userClicksTopic) val userRegionsTable: KTable[String, String] = builder.table(userRegionsTopic) // The following code fragment does not have a single instance of Grouped, // Produced, Consumed or Joined supplied explicitly. // All of them are taken care of by the implicit Serdes imported by Serdes val clicksPerRegion: KTable[String, Long] = userClicksStream .leftJoin(userRegionsTable)((clicks, region) => (if (region == null) "UNKNOWN" else region, clicks)) .map((_, regionWithClicks) => regionWithClicks) .groupByKey .reduce(_ + _) clicksPerRegion.toStream.to(outputTopic) ``` The preceding code snippet warrants some elaboration: - The code snippet does not depend on any config defined Serdes. In fact, any Serdes defined as part of the config are ignored. - All Serdes are picked up from the implicits in scope. And `import Serdes._` brings all necessary Serdes in scope. - Any needed Serde not provided by the imported implicits would be a compile-time error. - The code is tidy and focused on the actual transformation. ### User-Defined Serdes When the core Serdes are not enough and you need to define custom Serdes, the usage is exactly the same as in the preceding example. Define the implicit Serdes and start building the stream transformation. Here is an example with `AvroSerde`: ```scala // domain object as a case class case class UserClicks(clicks: Long) // An implicit Serde implementation for the values we want to // serialize as avro implicit val userClicksSerde: Serde[UserClicks] = new AvroSerde // Primitive Serdes import Serdes._ // And then business as usual .. val userClicksStream: KStream[String, UserClicks] = builder.stream(userClicksTopic) val userRegionsTable: KTable[String, String] = builder.table(userRegionsTopic) // Compute the total per region by summing the individual click counts per region. val clicksPerRegion: KTable[String, Long] = userClicksStream // Join the stream against the table. .leftJoin(userRegionsTable)((clicks, region) => (if (region == null) "UNKNOWN" else region, clicks.clicks)) // Change the stream from -> to -> .map((_, regionWithClicks) => regionWithClicks) // Compute the total per region by summing the individual click counts per region. .groupByKey .reduce(_ + _) // Write the (continuously updating) results to the output topic. clicksPerRegion.toStream.to(outputTopic) ``` A complete example of user-defined Serdes can be found in a test class within the library. #### 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).