private int maybeResetInputAndSeekToEndIntermediateTopicOffsets(final Map<Object, Object> consumerConfig,
                                                                    final StreamsResetterOptions options)
        throws IOException, ParseException {
        final List<String> inputTopics = options.inputTopicsOption();
        final List<String> intermediateTopics = options.intermediateTopicsOption();
        int topicNotFound = EXIT_CODE_SUCCESS;

        final List<String> notFoundInputTopics = new ArrayList<>();
        final List<String> notFoundIntermediateTopics = new ArrayList<>();

        if (inputTopics.size() == 0 && intermediateTopics.size() == 0) {
            System.out.println("No input or intermediate topics specified. Skipping seek.");
            return EXIT_CODE_SUCCESS;
        }

        if (inputTopics.size() != 0) {
            System.out.println("Reset-offsets for input topics " + inputTopics);
        }
        if (intermediateTopics.size() != 0) {
            System.out.println("Seek-to-end for intermediate topics " + intermediateTopics);
        }

        final Set<String> topicsToSubscribe = new HashSet<>(inputTopics.size() + intermediateTopics.size());

        for (final String topic : inputTopics) {
            if (!allTopics.contains(topic)) {
                notFoundInputTopics.add(topic);
            } else {
                topicsToSubscribe.add(topic);
            }
        }
        for (final String topic : intermediateTopics) {
            if (!allTopics.contains(topic)) {
                notFoundIntermediateTopics.add(topic);
            } else {
                topicsToSubscribe.add(topic);
            }
        }

        if (!notFoundInputTopics.isEmpty()) {
            System.out.println("Following input topics are not found, skipping them");
            for (final String topic : notFoundInputTopics) {
                System.out.println("Topic: " + topic);
            }
            topicNotFound = EXIT_CODE_ERROR;
        }

        if (!notFoundIntermediateTopics.isEmpty()) {
            System.out.println("Following intermediate topics are not found, skipping them");
            for (final String topic : notFoundIntermediateTopics) {
                System.out.println("Topic:" + topic);
            }
            topicNotFound = EXIT_CODE_ERROR;
        }

        // Return early if there are no topics to reset (the consumer will raise an error if we
        // try to poll with an empty subscription)
        if (topicsToSubscribe.isEmpty()) {
            return topicNotFound;
        }

        final Properties config = new Properties();
        config.putAll(consumerConfig);
        config.setProperty(ConsumerConfig.GROUP_ID_CONFIG, options.applicationId());
        config.setProperty(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "false");

        try (final KafkaConsumer<byte[], byte[]> client =
                 new KafkaConsumer<>(config, new ByteArrayDeserializer(), new ByteArrayDeserializer())) {

            final Collection<TopicPartition> partitions = topicsToSubscribe.stream().map(client::partitionsFor)
                    .flatMap(Collection::stream)
                    .map(info -> new TopicPartition(info.topic(), info.partition()))
                    .collect(Collectors.toList());
            client.assign(partitions);

            final Set<TopicPartition> inputTopicPartitions = new HashSet<>();
            final Set<TopicPartition> intermediateTopicPartitions = new HashSet<>();

            for (final TopicPartition p : partitions) {
                final String topic = p.topic();
                if (options.isInputTopic(topic)) {
                    inputTopicPartitions.add(p);
                } else if (options.isIntermediateTopic(topic)) {
                    intermediateTopicPartitions.add(p);
                } else {
                    System.err.println("Skipping invalid partition: " + p);
                }
            }

            maybeReset(client, inputTopicPartitions, options);

            maybeSeekToEnd(options.applicationId(), client, intermediateTopicPartitions);

            if (!options.hasDryRun()) {
                for (final TopicPartition p : partitions) {
                    client.position(p);
                }
                client.commitSync();
            }
        } catch (final IOException | ParseException e) {
            System.err.println("ERROR: Resetting offsets failed.");
            throw e;
        }
        System.out.println("Done.");
        return topicNotFound;
    }
{@link StreamsResetter} resets the processing state of a Kafka Streams application so that, for example,
you can reprocess its input from scratch.
<p>
<strong>This class is not part of public API. For backward compatibility,
use the provided script in "bin/" instead of calling this class directly from your code.</strong>
<p>
Resetting the processing state of an application includes the following actions:
<ol>
<li>setting the application's consumer offsets for input and internal topics to zero</li>
<li>skip over all intermediate user topics (i.e., "seekToEnd" for consumers of intermediate topics)</li>
<li>deleting any topics created internally by Kafka Streams for this application</li>
</ol>
<p>
Do only use this tool if <strong>no</strong> application instance is running.
Otherwise, the application will get into an invalid state and crash or produce wrong results.
<p>
If you run multiple application instances, running this tool once is sufficient.
However, you need to call {@code KafkaStreams#cleanUp()} before re-starting any instance
(to clean local state store directory).
Otherwise, your application is in an invalid state.
<p>
User output topics will not be deleted or modified by this tool.
If downstream applications consume intermediate or output topics,
it is the user's responsibility to adjust those applications manually if required.
tools/src/main/java/org/apache/kafka/tools/StreamsResetter.java