# Reset Kafka Streams Applications in Confluent Platform You can reset an application and force it to reprocess its data from scratch by using the application reset tool. This can be useful for development and testing, or when fixing bugs. The application reset tool handles the Kafka Streams [user topics](manage-topics.md#streams-developer-guide-topics-user) (input, output, and intermediate topics) and [internal topics](manage-topics.md#streams-developer-guide-topics-internal) differently when resetting the application. Here’s what the application reset tool does for each topic type: * Input topics: Reset offsets to specified position. By default they are reset to the beginning of the topic. * Intermediate topics: Skip to the end of the topic, that is, set the application’s committed consumer offsets for all partitions to each partition’s `logSize` (for consumer group `application.id`). * Internal topics: Delete the internal topic (this automatically deletes any committed offsets). The application reset tool does not: * Reset output topics of an application. If any output topics are consumed by downstream applications, it is your responsibility to adjust those downstream applications as appropriate when you reset the upstream application. * Reset the local environment of your application instances. It is your responsibility to delete the local state on any machine on which an application instance was run. See the instructions in section [Step 2: Reset the local environments of your application instances](#streams-developer-guide-reset-local-environment) on how to do this. * Delete schemas in Schema Registry for internal topics. You must delete schemas for internal topics manually if you reset an app that uses Schema Registry. The reset tool has a “dry run” option you can use to see the internal topics that the tool will delete. Prerequisites : - All instances of your application must be stopped. Otherwise, the application may enter an invalid state, crash, or produce incorrect results. You can verify whether the consumer group with ID `application.id` is still active by using `bin/kafka-consumer-groups`. - Use this tool with care and double-check its parameters: If you provide wrong parameter values (for example, typos in `application.id`) or specify parameters inconsistently (for example, specify the wrong input topics for the application), this tool might invalidate the application’s state or even impact other applications, consumer groups, or your Apache Kafka® topics. - Delete and re-create any intermediate topics manually before running the application reset tool. This frees up disk space in Kafka brokers. You can skip this step if either of the following applies: * You have external downstream consumers for the application’s intermediate topics. * You are in a development environment where manually deleting and re-creating intermediate topics is unnecessary. ## Step 1: Run the application reset tool If you are using **Streams Rebalance Protocol**, which is available beginning with Kafka 4.2 (Confluent Platform 8.2), use the [Streams groups CLI](kafka-streams-group-tool.md#streams-groups-tool). If you are using **classic rebalance protocol**, run the classic application reset tool as described in the following steps. Invoke the application reset tool from the command line: ```bash $CONFLUENT_HOME/bin/kafka-streams-application-reset ``` The tool accepts the following parameters: | Option | Description | |------------------------------------------------------|-----------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------| | –application-id | (REQUIRED) The Kafka Streams application ID
: (`application.id`). | | –bootstrap-server connect to> | REQUIRED unless `--bootstrap-servers`
(deprecated) is specified. The server(s)
to connect to. The broker list string has
the format `HOST1:PORT1,HOST2:PORT2`. | | –by-duration | Reset offsets to offset by duration from the current
timestamp. Format: `PnDTnHnMnS`. | | –config-file | Property file containing configs to be passed to
admin clients and embedded consumer. | | –dry-run | Display the actions that would be performed, without
executing the reset commands. | | –force | Force removing members of the consumer group.
Intended to remove left-over members if long session
timeout is configured. | | –from-file | Reset offsets to values defined in CSV file. | | –input-topics | Comma-separated list of user input topics. For these
topics, the tool resets the offset to the earliest
available offset. | | –internal-topics | Comma-separated list of internal topics to delete.
Must be a subset of the internal topics marked for
deletion by the default behavior. Tip: Do a dry-run
without this option to view these topics. | | –shift-by | Reset offsets, shifting the current offset by *n*.
The value of *n* can be positive or negative. | | –to-datetime | Reset offsets to offset from a datetime. Format:
`YYYY-MM-DDThh:mm:ss.sss`. | | –to-earliest | Reset offsets to the earliest offset. | | –to-latest | Reset offsets to the latest offset. | | –to-offset | Reset offsets to the specified offset. | Consider the following options as reset-offset scenarios for `input-topics`: - by-duration - from-file - shift-by - to-datetime - to-earliest - to-latest - to-offset Only one of these scenarios can be defined. If not defined, `to-earliest` is executed by default. You can combine all of the other parameters as needed. For example, if you want to restart an application from an empty internal state but not reprocess previous data, omit the parameter `--input-topics`. ## Step 2: Reset the local environments of your application instances For a complete application reset, you must delete the application’s local state directory on any machines where the application instance was run. You must do this before restarting an application instance on the same machine. You can use either of these methods: - The API method `KafkaStreams#cleanUp()` in your application code. - Manually delete the corresponding local state directory (default location: `/${java.io.tmpdir}/kafka-streams/`). For more information, see [state.dir](../javadocs.md#streams-javadocs) StreamsConfig class. ### Code example The following code example shows how to use the application reset tool to reprocess data. ```java package io.confluent.developer; import org.apache.kafka.common.serialization.Serde; import org.apache.kafka.common.serialization.Serdes; import org.apache.kafka.streams.KafkaStreams; import org.apache.kafka.streams.StreamsBuilder; import org.apache.kafka.streams.StreamsConfig; import org.apache.kafka.streams.Topology; import org.apache.kafka.streams.kstream.Consumed; import org.apache.kafka.streams.kstream.Produced; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import java.time.Duration; import java.util.Properties; import java.util.concurrent.CountDownLatch; public class KafkaStreamsApplication { private static final Logger LOG = LoggerFactory.getLogger(KafkaStreamsApplication.class); private final Serde stringSerde = Serdes.String(); public static final String INPUT_TOPIC = "input"; public static final String OUTPUT_TOPIC = "output"; public Topology buildTopology(Properties allProps) { StreamsBuilder builder = new StreamsBuilder(); builder.stream(INPUT_TOPIC, Consumed.with(stringSerde, stringSerde)) .peek((k, v) -> LOG.info("Observed event: {}", v)) .mapValues(s -> s.toUpperCase()) .peek((k, v) -> LOG.info("Transformed event: {}", v)) .to(OUTPUT_TOPIC, Produced.with(stringSerde, stringSerde)); return builder.build(allProps); } public static void main(String[] args) { Properties properties; if (args.length > 0) { properties = Utils.loadProperties(args[0]); } else { properties = Utils.loadProperties(); } properties.put(StreamsConfig.APPLICATION_ID_CONFIG, "kafka-streams-application"); KafkaStreamsApplication kafkaStreamsApplication = new KafkaStreamsApplication(); Topology topology = kafkaStreamsApplication.buildTopology(properties); try (KafkaStreams kafkaStreams = new KafkaStreams(topology, properties)) { CountDownLatch countDownLatch = new CountDownLatch(1); Runtime.getRuntime().addShutdownHook(new Thread(() -> { kafkaStreams.close(Duration.ofSeconds(5)); countDownLatch.countDown(); })); // For local running only don't do this in production as it wipes out all local state kafkaStreams.cleanUp(); kafkaStreams.start(); countDownLatch.await(); } catch (InterruptedException e) { Thread.currentThread().interrupt(); } } } ``` For the full code example, see [How to reset a Kafka Streams application to reprocess input topics](https://github.com/confluentinc/tutorials/tree/master/kafka-streams-application-reset/kstreams). ## Compatible topology changes In general, altering a Kafka Streams topology by adding or removing operations requires an application reset. This is because Kafka Streams must manage persistent resources like state stores and internal topics by name. You can see these names by printing the topology description with the `Topology#describe()` method. To check if a topology change is incompatible and requires a reset, perform a comparison of the `TopologyDescription` for the old and new topology. Names of everything internal that is stateful, like state store names and their changelog topic names, and repartition topics, should not change, otherwise the application can’t start. Potentially compatible changes: - Changing a filter condition - Inserting new filters, which are record-by-record operations - Inserting a new map, a record-by-record operation, if the data types are compatible, but be aware that a new map might create a downstream repartition topic that didn’t exist before, which can break compatibility - Calling `mapValues()`, if the value type doesn’t change Potentially incompatible changes: - Changing the structure of the DAG topology - Changing input or output data types of stateful operations, like aggregations or joins Some compatibility checks are done on startup. Here are some examples of changes that are detected to determine if the new topology is incompatible. Kafka Streams can’t detect all incompatible changes, so other changes can cause a topology to be incompatible. Changing the number of partitions : If you change the number of partitions of a topic used in the topology, the existing state stores can become inconsistent or incomplete. This can result in data loss or incorrect results. To avoid this, you should always use the same number of partitions for a topic throughout the life of the application.
A case of a compatible partition change is calling `builder.stream("topic1").map().repartition(/*set partition count*/).join(...)`. Changing the number of partitions of `topic1` is compatible, because the repartition topic count is fixed. Changing the key or value type : If you change the key or value types of a topic used in the topology, the existing state stores can become incompatible with the new type. This can result in data loss or incorrect results. To avoid this, you should always use the same key and value types for a topic throughout the life of the application.
Also, if schema evolution occurs, and the `schema-id` in Schema Registry changes for a key, the change can be incompatible. Changing the state store configuration : If you change the configuration of a state store used in the topology, like the name, retention policy, or changelog topic, the existing state stores can become incompatible with the new configuration. An example of an incompatible modification to changelog configuration is changing the values in the map provided in `withLoggingEnabled(Map config)`. This can result in data loss or incorrect results. To avoid this, you should always use the same state store configuration throughout the life of the application. Changing the topology structure : If you change the structure of the topology by adding or removing nodes, the existing state stores can become inconsistent or incomplete. This can result in data loss or incorrect results. To avoid this, you should always plan the topology structure carefully before deploying the application, and avoid making changes that are incompatible with the existing topology. ## Related content - Blog: [Kafka Streams’ Take on Watermarks and Triggers](https://www.confluent.io/blog/kafka-streams-take-on-watermarks-and-triggers/) - [Kafka Streams Data (Re)Processing Scenarios](https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Data+%28Re%29Processing+Scenarios) #### 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).