@Test
    public void shouldNotAllowToResetWhenInputTopicAbsent(final TestInfo testInfo) {
        final String appID = IntegrationTestUtils.safeUniqueTestName(testInfo);
        final String[] parameters = new String[] {
            "--application-id", appID,
            "--bootstrap-server", cluster.bootstrapServers(),
            "--input-topics", NON_EXISTING_TOPIC
        };
        final Properties cleanUpConfig = new Properties();
        cleanUpConfig.put(ConsumerConfig.HEARTBEAT_INTERVAL_MS_CONFIG, 100);
        cleanUpConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, Integer.toString(CLEANUP_CONSUMER_TIMEOUT));

        final int exitCode = new StreamsResetter().execute(parameters, cleanUpConfig);
        assertEquals(1, exitCode);
    }
Tests local state store and global application cleanup.
streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationTest.java