<a id="streams-developer-guide-processor-api"></a>

# Kafka Streams Processor API for Confluent Platform

The Processor API allows developers to define and connect custom processors and to interact with state stores. With the
Processor API, you can define arbitrary stream processors that process one received record at a time, and connect these
processors with their associated state stores to compose the processor topology that represents a customized processing
logic.

## Overview

The Processor API can be used to implement both **stateless** and **stateful** operations, where the latter is
achieved through the use of [state stores](#streams-developer-guide-state-store).

For a complete list of available API functionality, see the [Kafka Streams API docs](../javadocs.md#streams-javadocs).

<a id="streams-developer-guide-stream-processor"></a>

## Defining a Stream Processor

A [stream processor](../concepts.md#streams-concepts) is a node in the processor topology that represents a single processing step.
With the Processor API, you can define arbitrary stream processors that process one received record at a time, and connect
these processors with their associated state stores to compose the processor topology.

You can define a customized stream processor by implementing the `Processor` interface, which provides the `process()` API method.
The `process()` method is called on each of the received records.

The `Processor` interface also has an `init()` method, which the Kafka Streams library calls during the task construction
phase. Processor instances should perform any required initialization in this method. The `init()` method passes in a `ProcessorContext`
instance, which provides access to the metadata of the currently processed record, including its source Apache Kafka® topic and partition,
its corresponding message offset, and further such information. You can also use this context instance to schedule a punctuation
function (by using `ProcessorContext#schedule()`), to forward a new record as a key-value pair to the downstream processors (by using `ProcessorContext#forward()`),
and to request a commit of the current processing progress (by using `ProcessorContext#commit()`).
Any resources you set up in `init()` can be cleaned up in the `close()`
method. Kafka Streams might reuse a single Processor object by calling `init()`
on it again after `close()`.

The Processor interface takes four generic parameters: `KIn`, `VIn`,
`KOut`, and `VOut`. These define the input and output types that the
processor implementation can handle. `KIn` and `VIn` define the key and
value types of the `Record` that is passed to `process()`. Likewise,
`KOut` and `VOut` define the forwarded key and value types for the result
`Record` that `ProcessorContext#forward()` accepts. If your processor does
not forward any records at all, or if it forwards only null keys or values, a
best practice is to set the output generic type argument to `Void`. If it
needs to forward multiple types that don’t share a common superclass, you must
set the output generic type argument to `Object`.

Both the `Processor#process()` and the `ProcessorContext#forward()` methods
handle records in the form of the `Record<K, V>` data class. This class gives
you access to the main components of a Kafka record: the key, value, timestamp
and headers. When forwarding records, you can use the constructor to create a
new `Record` from scratch, or you can use the convenience builder methods to
replace one of the `Record` properties and copy over the rest. For example,
`inputRecord.withValue(newValue)` copies the key, timestamp, and headers from
`inputRecord` while setting the output record’s value to `newValue`. This
call doesn’t mutate `inputRecord` but instead creates a shallow copy. Because this is
only a shallow copy, if you plan to mutate the key, value, or headers elsewhere
in the program, you must create a deep copy of those fields manually.

In addition to handling incoming records by using `Processor#process()`, you
can schedule periodic invocation, referred to as “punctuation”, in your processor’s
`init()` method by calling `ProcessorContext#schedule()` and passing it a
`Punctuator`. The `PunctuationType` determines what notion of time is used
for the punctuation scheduling: either [stream-time](../concepts.md#streams-concepts-time)
or wall-clock-time. By default, stream-time is configured to represent event-time
via `TimestampExtractor`. When stream-time is used, `punctuate()` is triggered
purely by data, because stream-time is determined (and advanced forward) by the
timestamps derived from the input data. When there is no new input data arriving,
stream-time is not advanced and `punctuate()` is not called.

For example, if you schedule a `Punctuator` function every 10 seconds based
on `PunctuationType.STREAM_TIME` and if you process a stream of 60 records
with consecutive timestamps from 1 (first record) to 60 seconds (last record),
then `punctuate()` is called 6 times. This happens regardless of the actual
time required to process these records, whether processing these 60 records takes a second, a minute, or
an hour.

#### IMPORTANT
When tasks are moved to a different client or a different thread,
Kafka Streams calls `close()` on the old owner and `init` on the new one.
You should also `cancel()` the punctuator inside the `close()` method.

When wall-clock-time (`PunctuationType.WALL_CLOCK_TIME`) is used,
`punctuate()` is triggered purely by the wall-clock time.
Reusing the example above, if the `Punctuator` function is scheduled based
on `PunctuationType.WALL_CLOCK_TIME`, and if these 60 records were processed
within 20 seconds, `punctuate()` is called 2 times, one time every 10 seconds.
If these 60 records were processed within 5 seconds, then no `punctuate()` is
called at all. Note that you can schedule multiple `Punctuator` callbacks with
different `PunctuationType` types within the same processor by calling
`ProcessorContext#schedule()` multiple times inside the `init()` method.

Stream-time is advanced only when Kafka Streams processes records. If there are no
records to process, or if Kafka Streams is waiting for new records due to the
[Task Idling](config-streams.md#streams-developer-guide-max-idle) configuration, stream-
time doesn’t advance, and `punctuate()` isn’t triggered if
`PunctuationType.STREAM_TIME` was specified. This behavior is independent
of the configured timestamp extractor, which means that using
`WallclockTimestampExtractor` doesn’t enable wall-clock triggering of
`punctuate()`.

#### NOTE
A single Kafka Streams task, including all of its processors, is always
executed by exactly one `StreamThread`. Because of this, `process()`
and `punctuate()` calls for a given task are never executed concurrently.
They run one at a time, interleaved on that task’s thread. You don’t need
additional synchronization to protect state (for example, a local state
store or an in-memory field on your `Processor`) that’s accessed from
both `process()` and `punctuate()`.

The following example `Processor` defines a simple word-count algorithm, and this example performs the following actions:

- In the `init()` method, schedule the punctuation every second (the minimum time unit supported is one millisecond) and retrieve the local state store by its name “Counts”.
- In the `process()` method, upon each received record, split the value string into words, and update their counts into the state store (described later in this section).
- In the `punctuate()` method, iterate the local state store and send the aggregated counts to the downstream processor (downstream processors are described later in this section), and commit the current stream state.

#### ATTENTION
The code shows a simplified example that only works for single partition input topics.
A generic Processor API word-count would require two processors: the first is stateless, splits each line into words, and sets the words as record keys.
The result of the first processor is written back to an additional topic that is consumed by the second (stateful) processor that does the actual counting.
Writing the words back to a topic is required to group the same words together so they go to the same instance of the second processor.
This is similar in function to the shuffle phase of a Map-Reduce computation.

```java
public class WordCountProcessor implements Processor<String, String, String, String> {
    private KeyValueStore<String, Integer> kvStore;

    @Override
    public void init(final ProcessorContext<String, String> context) {
        context.schedule(Duration.ofSeconds(1), PunctuationType.STREAM_TIME, timestamp -> {
            try (final KeyValueIterator<String, Integer> iter = kvStore.all()) {
                while (iter.hasNext()) {
                    final KeyValue<String, Integer> entry = iter.next();
                    context.forward(new Record<>(entry.key, entry.value.toString(), timestamp));
                }
            }
        });
        kvStore = context.getStateStore("Counts");
    }

    @Override
    public void process(final Record<String, String> record) {
        final String[] words = record.value().toLowerCase(Locale.getDefault()).split("\\W+");

        for (final String word : words) {
            final Integer oldValue = kvStore.get(word);

            if (oldValue == null) {
                kvStore.put(word, 1);
            } else {
                kvStore.put(word, oldValue + 1);
            }
        }
    }

    @Override
    public void close() {
        // close any resources managed by this processor
        // Note: Do not close any StateStores as these are managed by the library
    }
}
```

Stateful processing with state stores
: The `WordCountProcessor` defined above can access the currently received
  record in its `process()` method, and it can leverage
  [state stores](#streams-developer-guide-state-store) to maintain
  processing states, for example, to remember recently arrived records for
  stateful processing needs like aggregations and joins. For more information,
  see [state stores](#streams-developer-guide-state-store).

<a id="streams-developer-guide-processor-context"></a>

## Accessing Processor Context

As mentioned [previously](#streams-developer-guide-stream-processor),
a `ProcessorContext` controls the processing workflow, such as scheduling
a punctuation function, and committing the current processed state.

You can use this object to access the metadata related to the application,
like `applicationId`, `taskId`, and `stateDir`. Also, you can access
`RecordMetadata` like `topic`, `partition`, and `offset`.

The following example `process()` function enriches the record differently based on the record context:

```java
public class EnrichProcessor implements Processor<String, String, String, String> {

  private ProcessorContext<String, String> context;

  @Override
  public void init(ProcessorContext<String, String> context) {
      // keep the processor context locally because we need it in process()
      this.context = context;
  }

  @Override
  public void process(Record<String, String> record) {
      String topic = context.recordMetadata().map(RecordMetadata::topic).orElse("");
      switch (topic) {
          case "alerts":
              context.forward(record.withValue(decorateWithHighPriority(record.value())));
              break;
          case "notifications":
              context.forward(record.withValue(decorateWithMediumPriority(record.value())));
              break;
          default:
              context.forward(record.withValue(decorateWithLowPriority(record.value())));
      }
  }
}
```

Record context metadata
: The metadata of the currently processing record may not always be available.
  For example, if the current processing record is not piped from any source
  topic, but is generated from a punctuation function, then its metadata field
  `topic` is empty, and `partition` and `offset` are a sentinel
  value (`-1` in this case), while its `timestamp` field is the
  triggering time of the punctuation function that generated this record.

## Accessing Header Metadata

You can append metadata to records by using the
[headers()](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/processor/ProcessorContext.html#headers())
method.

Headers are useful for scenarios like propagating tracing context between
different components and adding operational information that you can use for
filtering records.

You can access message metadata in the `process()` method.

The following code example shows how to add a header to records:

```java
public void process(String key, String value) {

    // Add a header to the elements.
    context().headers().add(key, value.getBytes());
}
```

<a id="streams-developer-guide-state-store"></a>

## State stores

To implement a **stateful** `Processor`, you must provide one or more state
stores to the processor or transformer (*stateless* processors do not need
state stores). You can use state stores to remember recently received input
records, to track rolling aggregates, to de-duplicate input records, and more.

Another feature of state stores is that they can be
[interactively queried](interactive-queries.md#streams-developer-guide-interactive-queries)
from other applications, such as a NodeJS-based dashboard or a microservice
implemented in Scala or Go.

The
[available state store types](#streams-developer-guide-state-store-defining) in Kafka Streams have
[fault tolerance](#streams-developer-guide-state-store-fault-tolerance) enabled by default.

<a id="streams-developer-guide-state-store-defining"></a>

### Define and create a state store

You can either use one of the available store types or
[implement your own custom store type](#streams-developer-guide-state-store-custom).
It’s common practice to use an existing store type through the `Stores` factory.

When using Kafka Streams, you usually don’t create or instantiate state stores
directly in your code. Instead, you define state stores indirectly by creating
a `StoreBuilder`.  This builder is used by Kafka Streams as a factory to
instantiate the actual state stores locally in application instances when and
where needed.

The following store types are available.

| Store Type                           | Storage Engine   | Fault-tolerant?          | Description                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                                         |
|--------------------------------------|------------------|--------------------------|-------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------|
| Persistent<br/>`KeyValueStore<K, V>` | RocksDB          | Yes (enabled by default) | - **The recommended store type for most use cases.**<br/>- Stores its data on local disk.<br/>- Storage capacity:<br/>  managed local state can be larger than the memory (heap space) of an<br/>  application instance, but must fit into the available local disk<br/>  space.<br/>- RocksDB settings can be fine-tuned, see<br/>  [RocksDB configuration](config-streams.md#streams-developer-guide-rocksdb-config).<br/>- Available store variants:<br/>  : [time window key-value store](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/Stores.html#persistentWindowStore-java.lang.String-java.time.Duration-java.time.Duration-boolean-),<br/>    [session window key-value store](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/Stores.html#persistentSessionStore-java.lang.String-java.time.Duration-),<br/>    [timestamped key-value store](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/Stores.html#persistentTimestampedKeyValueStore-java.lang.String-),<br/>    [timestamped window key-value store](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/Stores.html#persistentTimestampedWindowStore-java.lang.String-java.time.Duration-java.time.Duration-boolean-)<br/><br/>```java<br/>// Creating a persistent key-value store:<br/>// here, we create a `KeyValueStore<String, Long>` named "persistent-counts".<br/>import org.apache.kafka.streams.state.StoreBuilder;<br/>import org.apache.kafka.streams.state.Stores;<br/><br/>StoreBuilder countStoreBuilder =<br/>  Stores.keyValueStoreBuilder(<br/>    Stores.persistentKeyValueStore("persistent-counts"),<br/>    Serdes.String(),<br/>    Serdes.Long()<br/>  );<br/>```<br/><br/>See<br/>[PersistentKeyValueStore](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/Stores.html#persistentKeyValueStore-java.lang.String-)<br/>for detailed factory options. |
| In-memory<br/>`KeyValueStore<K, V>`  | -                | Yes (enabled by default) | - Stores its data in memory.<br/>- Storage capacity:<br/>  managed local state must fit into memory (heap space) of an<br/>  application instance.<br/>- Useful when application instances run in an environment where local<br/>  disk space is either not available or local disk space is wiped<br/>  in-between app instance restarts.<br/>- Available store variants:<br/>  : [time window key-value store](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/Stores.html#inMemoryWindowStore-java.lang.String-java.time.Duration-java.time.Duration-boolean-),<br/>    [session window key-value store](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/Stores.html#inMemorySessionStore-java.lang.String-long-),<br/>    [timestamped key-value store](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/Stores.html),<br/>    [timestamped window key-value store](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/Stores.html)<br/><br/>```java<br/>// Creating an in-memory key-value store:<br/>// here, we create a `KeyValueStore<String, Long>` named "inmemory-counts".<br/>import org.apache.kafka.streams.state.StoreBuilder;<br/>import org.apache.kafka.streams.state.Stores;<br/><br/>StoreBuilder countStoreBuilder =<br/>  Stores.keyValueStoreBuilder(<br/>    Stores.inMemoryKeyValueStore("inmemory-counts"),<br/>    Serdes.String(),<br/>    Serdes.Long()<br/>  );<br/>```<br/><br/>See<br/>[InMemoryKeyValueStore](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/Stores.html#inMemoryKeyValueStore-java.lang.String-) for detailed factory options.                                                                                                                                                                                                                                                                    |

<a id="streams-developer-guide-state-store-fault-tolerance"></a>

### Fault-tolerant state stores

To make state stores fault-tolerant and to allow for state store migration without data loss, a state store can be
continuously backed up to a Kafka topic behind the scenes. For example, to migrate a stateful stream task from one
machine to another when [elastically adding or removing capacity from your application](running-app.md#streams-developer-guide-execution-scaling).
This topic is sometimes referred to as the state store’s associated *changelog topic*, or its *changelog*.  For example, if
you experience machine failure, the state store and the application’s state can be fully restored from its changelog. You can
[enable or disable this backup feature](#streams-developer-guide-state-store-enable-disable-fault-tolerance) for a
state store.

Fault-tolerant state stores are backed by a
[compacted](/kafka/design/log_compaction.html) changelog topic.  The purpose of compacting this
topic is to prevent the topic from growing indefinitely, to reduce the storage consumed in the associated Kafka cluster,
and to minimize recovery time if a state store needs to be restored from its changelog topic.

Fault-tolerant windowed state stores are backed by a topic that uses both compaction and
deletion. Because of the structure of the message keys sent to the changelog topics, this combination of
deletion and compaction is required for the changelog topics of window stores. For window stores, the message keys are
composite keys that include the “normal” key and window timestamps.  For these types of composite keys it would not
be sufficient to only enable compaction to prevent a changelog topic from growing out of bounds.  With deletion
enabled, Kafka’s log cleaner cleans up old windows that have expired as the log segments expire.  The
default retention setting is `Materialized#withRetention()` + 1 day.  You can override this setting by specifying
`StreamsConfig.WINDOW_STORE_CHANGE_LOG_ADDITIONAL_RETENTION_MS_CONFIG` in the `StreamsConfig`.

When you open an `Iterator` from a state store you must call `close()` on the iterator when you are done working with
it to reclaim resources; or you can use the iterator from within a try-with-resources statement. If you do not close an iterator,
you may encounter an OOM error.

<a id="streams-developer-guide-state-store-enable-disable-fault-tolerance"></a>

### Enable or disable fault tolerance of state stores (store changelogs)

You can enable or disable fault tolerance for a state store by enabling or disabling the change logging
of the store through `withLoggingEnabled()` and `withLoggingDisabled()`.
You can also fine-tune the associated topic’s configuration if needed.

Example for disabling fault-tolerance:

```java
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;

StoreBuilder<KeyValueStore<String, Long>> countStoreSupplier = Stores.keyValueStoreBuilder(
  Stores.persistentKeyValueStore("Counts"),
    Serdes.String(),
    Serdes.Long())
  .withLoggingDisabled(); // disable backing up the store to a changelog topic
```

#### IMPORTANT
If the changelog is disabled, the attached state store is no longer fault
tolerant, and it can’t have any
[standby replicas](config-streams.md#streams-developer-guide-standby-replicas).

Here is an example for enabling fault tolerance, with additional changelog-topic configuration:
You can add any log config from [kafka.log.LogConfig](https://github.com/apache/kafka/blob/trunk/core/src/main/scala/kafka/log/LogConfig.scala#L61).
Kafka Streams ignores unrecognized configurations.

```java
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;

Map<String, String> changelogConfig = new HashMap<>();
// override min.insync.replicas
changelogConfig.put("min.insync.replicas", "1");

StoreBuilder<KeyValueStore<String, Long>> countStoreSupplier = Stores.keyValueStoreBuilder(
  Stores.persistentKeyValueStore("Counts"),
    Serdes.String(),
    Serdes.Long())
  .withLoggingEnabled(changelogConfig); // enable changelogging, with custom changelog settings
```

<a id="streams-developer-guide-timestamped-state-store"></a>

### Timestamped state stores

KTables always store timestamps by default. A timestamped state store improves
stream processing semantics and enables management of out-of-order data in source
KTables, detects out-of-order joins and aggregations, and gets the
timestamp of the latest update in an Interactive Query.

You can query timestamped state stores with or without a timestamp.

### Upgrade state stores

You can upgrade with a single rolling bounce per instance.

- For Processor API users, nothing changes in existing applications,
  and you have the option of using the timestamped stores.
- For DSL operators, store data is upgraded lazily in the background.
- No upgrade happens if you provide a custom `XxxBytesStoreSupplier`,
  but you can opt in by implementing the
  [TimestampedBytesStore](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/TimestampedBytesStore.html)
  interface. In this case, the old format is retained, and Kafka Streams
  uses a proxy store that removes/adds timestamps on read/write.

<a id="streams-developer-guide-versioned-state-stores"></a>

### Versioned key-value state stores

Versioned key-value state stores are available as of Confluent Platform 7.5 (Kafka Streams 3.5).
Rather than storing a single record version (value and timestamp) per key,
versioned state stores may store multiple record versions per key. This enables
versioned state stores to support timestamped retrieval operations to return
the latest record (per key) at a specified timestamp.

You can create a persistent, versioned state store by passing a
[VersionedBytesStoreSupplier](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/VersionedBytesStoreSupplier.html)
to the
[StoreBuilder<VersionedKeyValueStore<K,V>>](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/Stores.html#versionedKeyValueStoreBuilder-org.apache.kafka.streams.state.VersionedBytesStoreSupplier-org.apache.kafka.common.serialization.Serde-org.apache.kafka.common.serialization.Serde-),
or by implementing your own
[VersionedKeyValueStore](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/VersionedKeyValueStore.html).

Each versioned store has an associated, fixed-duration *history retention*
parameter that specifies how long old record versions should be kept for. In
particular, a versioned store guarantees returning accurate results for
timestamped retrieval operations where the timestamp being queried is within
the history retention period of the current observed stream time.

History retention also doubles as its *grace period*, which determines how far
back in time out-of-order writes to the store are accepted. A versioned store
doesn’t accept writes (inserts, updates, or deletions) if the timestamp
associated with the write is older than the current observed stream time by
more than the grace period. Stream time in this context is tracked per-partition,
rather than per-key, which means it’s important that grace period (history
retention) be set high enough to accommodate a record with one key arriving
out-of-order relative to a record for another key.

Because the memory footprint of versioned key-value stores is higher than that
of non-versioned key-value stores, you may need to adjust your
[RocksDB memory settings](memory-mgmt.md#rocksdb-mem-mgmt) accordingly.
Benchmarking your application with versioned stores is also advised, because
performance is typically lower than with non-versioned stores.

Versioned stores don’t support caching or interactive queries. Also, you can’t
version window stores or global tables.

#### Upgrade to versioned state stores

Versioned state stores are opt-in only, so no automatic upgrades from
non-versioned to versioned stores occur.

Upgrades are supported from persistent, non-versioned key-value stores to
persistent, versioned key-value stores as long as the original store has the
same changelog topic format as the versioned store being upgraded to. Both
[persistent key-value stores](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/Stores.html#persistentKeyValueStore(java.lang.String).html)
and
[timestamped key-value stores](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/Stores.html#persistentTimestampedKeyValueStore(java.lang.String).html)
share the same changelog topic format as
[persistent versioned key-value stores](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/state/Stores.html#persistentVersionedKeyValueStore(java.lang.String,java.time.Duration).html),
so both are eligible for upgrades.

Follow this procedure to upgrade an application that has persistent,
non-versioned key-value stores to use persistent, versioned key-value
stores.

1. Stop all application instances, and
   [clear any local state directories](app-reset-tool.md#streams-developer-guide-reset-local-environment)
   for the store(s) being upgraded.
2. Update your application code to use versioned stores where desired.
3. Update your changelog topic configs for the relevant state stores, to set
   the value of `min.compaction.lag.ms` to be at least your desired history
   retention. Use history retention plus one day as a buffer for the
   use of broker wall-clock time during compaction.
4. Restart your application instances and allow time for the versioned stores
   to rebuild state from changelog.

### Read-only state stores

A read-only state store materializes the data from its input topic. Also, it
uses the input topic for fault-tolerance, and so does not have an additional
changelog topic (the input topic is re-used as changelog). The input topic
should be configured with log compaction. Note that no other processor should
modify the content of the state store, and the only writer should be the
associated “state update processor”; other processors may read the content of
the read-only store.

#### NOTE
Beware of the partitioning requirements when using read-only state stores
for lookups during processing. You might want to make sure the original
changelog topic is co-partitioned with the processors reading the
read-only state store.

<a id="streams-developer-guide-state-store-custom"></a>

### Implement custom state stores

You can use the [built-in state store types](#streams-developer-guide-state-store-defining)
or implement your own. The primary interface to implement for the store is
`org.apache.kafka.streams.processor.StateStore`.  Kafka Streams also has a few
extended interfaces, such as `KeyValueStore` and `VersionedKeyValueStore`.

Your customized `org.apache.kafka.streams.processor.StateStore` implementation
also must provide the logic on how to restore the state with the
`org.apache.kafka.streams.processor.StateRestoreCallback` or
`org.apache.kafka.streams.processor.BatchingStateRestoreCallback` interface.
For more information on how to instantiate these interfaces, see
[StateStore](/platform/current/streams/javadocs/javadoc/org/apache/kafka/streams/processor/StateStore.html).

You also need to provide a “factory” for the store by implementing the
`org.apache.kafka.streams.state.StoreBuilder` interface, which Kafka Streams uses to create instances of
your store.

<a id="streams-developer-guide-state-store-custom-managed-offsets"></a>

## Self-manage changelog offsets (KIP-1035)

Confluent Platform 8.3 (Kafka Streams 4.3) adds three default methods on
`org.apache.kafka.streams.processor.StateStore` that let a store manage its
own changelog offsets atomically with its data, instead of relying on the
separate `.checkpoint` file:

- `boolean managesOffsets()`: Returns `true` if the store persists its
  consumer offset alongside its data. The default implementation returns
  `false`, preserving the legacy `.checkpoint` file behavior. This method
  is deprecated and might be removed in a future release. New implementations
  should return `true` and manage their own offsets.
- `void commit(Map<TopicPartition, Long> changelogOffsets)`: Atomically
  persists pending writes and the given changelog offsets. Replaces `flush()`
  for stores that manage their own offsets.
- `Long committedOffset(TopicPartition partition)`: Returns the last
  committed changelog offset for the given partition, or `null` if no offset
  has been committed.

A store that overrides `managesOffsets()` to return `true` must also
override `commit()` and `committedOffset()` to maintain offset-data
atomicity.

RocksDB-backed stores in Kafka Streams 4.3 use this mechanism through a dedicated
column family for offsets, which eliminates the corruption window that existed
between writing the store and writing the legacy `.checkpoint` file.

The `flush()` method on `StateStore` is deprecated. Existing custom stores
that do not override `managesOffsets()` continue to work unchanged.

For more information, see
[KIP-1035](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1035%3A+StateStore+Managed+Changelog+Offsets).

<a id="streams-developer-guide-state-store-header-aware"></a>

## Header-aware state stores (KIP-1271)

Confluent Platform 8.3 (Kafka Streams 4.3) introduces header-aware variants of the standard
state store interfaces. These stores persist Kafka record headers alongside
keys, values, and timestamps. New interfaces:

- `KeyValueStoreWithHeaders<K, V>`
- `TimestampedKeyValueStoreWithHeaders<K, V>`
- `WindowStoreWithHeaders<K, V>`
- `TimestampedWindowStoreWithHeaders<K, V>`
- `SessionStoreWithHeaders<K, V>`

Each interface extends its non-header-aware counterpart and adds `put` /
`get` overloads that accept or return a `Headers` parameter. Records
written through the non-header-aware methods are stored with empty headers.

To materialize a header-aware store from the DSL, set
[dsl.store.format=HEADERS](config-streams.md#streams-developer-guide-dsl-store-format) in
your application config. To build a header-aware store directly with the
Processor API, use the new factory methods on
`org.apache.kafka.streams.state.Stores`:

- `Stores.headersAwareKeyValueStoreBuilder(...)`
- `Stores.headersAwareTimestampedKeyValueStoreBuilder(...)`
- `Stores.headersAwareWindowStoreBuilder(...)`
- `Stores.headersAwareTimestampedWindowStoreBuilder(...)`
- `Stores.headersAwareSessionStoreBuilder(...)`

For test access, `TopologyTestDriver` adds matching accessor methods, such
as `getKeyValueStoreWithHeaders(String)` and
`getTimestampedKeyValueStoreWithHeaders(String)`, that return the
header-aware view of a store materialized with `HEADERS` format.

Header-aware stores are required to use the Schema Registry
[schema GUID in record header](../../schema-registry/fundamentals/serdes-develop/index.md#messages-wire-format-schema-id-in-header)
format with Kafka Streams. For migration, performance, and current limitations,
see the [dsl.store.format](config-streams.md#streams-developer-guide-dsl-store-format)
config reference.

For more information, see
[KIP-1271](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1271%3A+Headers-Aware+State+Stores)
and
[KIP-1285](https://cwiki.apache.org/confluence/display/KAFKA/KIP-1285%3A+DSL+Opt-in+Support+for+Headers-Aware+State+Stores).

## Connect processors and state stores

Now that a [processor](#streams-developer-guide-stream-processor) (WordCountProcessor) and the
state stores have been defined, you can construct the processor topology by connecting these processors and state stores together by
using the `Topology` instance.  In addition, you can add source processors with the specified Kafka topics
to generate input data streams into the topology, and sink processors with the specified Kafka topics to generate
output data streams out of the topology.

Here is an example implementation:

```java
Topology builder = new Topology();
// add the source processor node that takes Kafka topic "source-topic" as input
builder.addSource("Source", "source-topic")
    // add the WordCountProcessor node which takes the source processor as its upstream processor
    .addProcessor("Process", () -> new WordCountProcessor(), "Source")
    // add the count store associated with the WordCountProcessor processor
    .addStateStore(countStoreBuilder, "Process")
    // add the sink processor node that takes Kafka topic "sink-topic" as output
    // and the WordCountProcessor node as its upstream processor
    .addSink("Sink", "sink-topic", "Process");
```

Here is a quick explanation of this example:

- You add a source processor node named `"Source"` to the topology using the `addSource` method, with one Kafka topic
  `"source-topic"` fed to it. You can specify optional key and value deserializers to read the source, such as
  [Confluent GenericAvroSerde and SpecificAvroSerde](datatypes.md#streams-developer-guide-serdes).
- You then add a processor node named `"Process"` with the pre-defined `WordCountProcessor` logic as the downstream
  processor of the `"Source"` node using the `addProcessor` method.
- A predefined persistent key-value state store is created and associated with the `"Process"` node, using
  `countStoreSupplier`.
- You then add a sink processor node to complete the topology using the `addSink` method, taking the `"Process"` node
  as its upstream processor and writing to a separate `"sink-topic"` Kafka topic. You can also use another
  overloaded variant of `addSink` to dynamically determine the Kafka topic to write to for each received record from the upstream processor.

In some cases, it may be more convenient to add and connect a state store when
you add the processor to the topology. This can be done by implementing
`ConnectedStoreProvider#stores()` on the `ProcessorSupplier` instead of
calling `Topology#addStateStore()`, like this:

```java
Topology builder = new Topology();
// add the source processor node that takes Kafka "source-topic" as input
builder.addSource("Source", "source-topic")
    // add the WordCountProcessor node which takes the source processor as its upstream processor.
    // the ProcessorSupplier provides the count store associated with the WordCountProcessor
    .addProcessor("Process", new ProcessorSupplier<String, String, String, String>() {
        public Processor<String, String, String, String> get() {
            return new WordCountProcessor();
        }

        public Set<StoreBuilder<?>> stores() {
            final StoreBuilder<KeyValueStore<String, Long>> countsStoreBuilder =
                Stores
                    .keyValueStoreBuilder(
                        Stores.persistentKeyValueStore("Counts"),
                        Serdes.String(),
                        Serdes.Long()
                    );
            return Collections.singleton(countsStoreBuilder);
        }
    }, "Source")
    // add the sink processor node that takes Kafka topic "sink-topic" as output
    // and the WordCountProcessor node as its upstream processor
    .addSink("Sink", "sink-topic", "Process");
```

This enables a processor to “own” state stores, effectively encapsulating their
usage from the user wiring the topology. Multiple processors that share a state
store may provide the same store with this technique, as long as the
`StoreBuilder` is the same instance.

In these topologies, the `"Process"` stream processor node is considered a
downstream processor of the `"Source"` node, and an upstream processor of the
`"Sink"` node. As a result, whenever the `"Source"` node forwards a newly
fetched record from Kafka to its downstream `"Process"` node, the
`WordCountProcessor#process()` method is triggered to process the record and
update the associated state store. Whenever `context#forward()` is called in
the `WordCountProcessor#punctuate()` method, the aggregate records are
sent via the `"Sink"` processor node to the Kafka topic `"sink-topic"`.
Note that in the `WordCountProcessor` implementation, you must refer to the
same store name `"Counts"` when accessing the key-value store, otherwise an
exception is thrown at runtime, indicating that the state store cannot be
found. If the state store is not associated with the processor in the
`Topology` code, accessing it in the processor’s `init()` method also
throws an exception at runtime, indicating the state store is not accessible
from this processor.

The `Topology#addProcessor` function takes a `ProcessorSupplier` argument,
and the supplier pattern requires that a new `Processor` instance is returned
each time `ProcessorSupplier#get()` is called. Creating a single `Processor`
object and returning the same object reference in `ProcessorSupplier#get()`
is a violation of the supplier pattern and leads to runtime exceptions, so don’t
provide a singleton `Processor` instance to `Topology`. The `ProcessorSupplier`
should always generate a new instance each time `ProcessorSupplier#get()` is called.

Now that you have fully defined your processor topology in your application, you can proceed to
[running the Kafka Streams application](running-app.md#streams-developer-guide-execution).

<a id="streams-developer-guide-describing-a-topology"></a>

## Describe a topology

After a `Topology` is specified, you can retrieve a description of the corresponding DAG by using `#describe()`, which returns a `TopologyDescription`.
A `TopologyDescription` contains all added source, processor, and sink nodes as well as all attached stores.
You can access the specified input and output topic names and patterns for source and sink nodes.
For processor nodes, the attached stores are added to the description.
Additionally, all nodes have a list to all their connected successor and predecessor nodes.
`TopologyDescription` lets you retrieve the `DAG` structure of the specified topology.
Note that global stores are listed explicitly because they are accessible by all nodes without the need to explicitly connect them.
Furthermore, nodes are grouped by `SubTopology`, where each `SubTopology` is a group of processor nodes that are directly connected to each other (that is, either by a direct connection–but not a topic–or by sharing a store).
During execution, each `SubTopology` is processed by one or multiple tasks.
Each `SubTopology` describes an independent unit of work that can be executed by different threads in parallel.
Describing a `Topology` before starting your streams application with the specified topology is helpful to reason about tasks and thus maximum parallelism (described later in this section).
It is also helpful to get insight into a `Topology` if it is not specified directly as described above but by using the Kafka Streams DSL.

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