/usr/lib/kafka/bin/kafka-streams-application-reset.sh

This tool helps to quickly reset an application in order to reprocess its data from scratch.
* This tool resets offsets of input topics to the earliest available offset (by default), or to a specific defined position* This tool deletes the internal topics that were created by Kafka Streams (topics starting with "<application.id>-").
The tool finds these internal topics automatically. If the topics flagged automatically for deletion by the dry-run are unsuitable, you can specify a subset with the "--internal-topics" option.
* This tool will not delete output topics (if you want to delete them, you need to do it yourself with the bin/kafka-topics.sh command).
* This tool will not clean up the local state on the stream application instances (the persisted stores used to cache aggregation results).
You need to call KafkaStreams#cleanUp() in your application or manually delete them from the directory specified by "state.dir" configuration (${java.io.tmpdir}/kafka-streams/<application.id> by default).
* When long session timeout has been configured, active members could take longer to get expired on the broker thus blocking the reset job to complete. Use the "--force" option could remove those left-over members immediately. Make sure to stop all stream applications when this option is specified to avoid unexpected disruptions.

*** Important! You will get wrong output if you don't clean up the local stores after running the reset tool!

*** Warning! This tool makes irreversible changes to your application. It is strongly recommended that you run this once with "--dry-run" to preview your changes before making them.


Option (* = required)                  Description
---------------------                  -----------
* --application-id <String: id>        REQUIRED: The Kafka Streams
                                         application ID (application.id).
--bootstrap-server <String: server to  The server(s) to connect to. The
  connect to>                            broker list string in the form HOST1:
                                         PORT1,HOST2:PORT2. (default:
                                         localhost:9092)
--by-duration <String>                 Reset offsets to offset by duration
                                         from current timestamp. Format:
                                         'PnDTnHnMnS'
--command-config <String: file name>   Config properties file to be passed to
                                         admin clients and embedded consumer.
--config-file <String: file name>      (DEPRECATED) Property file containing
                                         configs to be passed to admin
                                         clients and embedded consumer. This
                                         option will be removed in a future
                                         version. Use --command-config
                                         instead.
--dry-run                              Display the actions that would be
                                         performed without executing the
                                         reset commands.
--force                                Force the removal of members of the
                                         consumer group (intended to remove
                                         stopped members if a long session
                                         timeout was used). Make sure to shut
                                         down all stream applications when
                                         this option is specified to avoid
                                         unexpected rebalances.
--from-file <String>                   Reset offsets to values defined in CSV
                                         file.
--help                                 Print usage information.
--input-topics <String: list>          Comma-separated list of user input
                                         topics. For these topics, the tool
                                         by default will reset the offset to
                                         the earliest available offset. Reset
                                         to other offset position by
                                         appending other reset offset option,
                                         ex: --input-topics foo --shift-by 5
--intermediate-topics <String: list>   [deprecated] Comma-separated list of
                                         intermediate user topics (topics
                                         that are input and output topics).
                                         For these topics, the tool will skip
                                         to the end.
--internal-topics <String: list>       Comma-separated list of internal
                                         topics to delete. Must be a subset
                                         of the internal topics marked for
                                         deletion by the default behaviour
                                         (do a dry-run without this option to
                                         view these topics).
--shift-by <Long: number-of-offsets>   Reset offsets shifting current offset
                                         by 'n', where 'n' can be positive or
                                         negative
--to-datetime <String>                 Reset offsets to offset from datetime.
                                         Format: 'YYYY-MM-DDThh:mm:ss.sss'
--to-earliest                          Reset offsets to earliest offset.
--to-latest                            Reset offsets to latest offset.
--to-offset <Long>                     Reset offsets to a specific offset.
--version                              Display Kafka version.