@Test
    public void testResetWhenLongSessionTimeoutConfiguredWithForceOption(final TestInfo testInfo) throws Exception {
        final String appID = IntegrationTestUtils.safeUniqueTestName(testInfo);
        streamsConfig.put(StreamsConfig.APPLICATION_ID_CONFIG, appID);
        streamsConfig.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, Integer.toString(STREAMS_CONSUMER_TIMEOUT * 100));

        // Run
        streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
        streams.start();
        final List<KeyValue<Long, Long>> result = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);

        streams.close();

        // RESET
        streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfig);
        streams.cleanUp();

        // Reset would fail since long session timeout has been configured
        final boolean cleanResult = tryCleanGlobal(false, null, null, appID);
        assertFalse(cleanResult);

        // Reset will success with --force, it will force delete active members on broker side
        cleanGlobal(false, "--force", null, appID);
        assertThat("Group is not empty after cleanGlobal", isEmptyConsumerGroup(adminClient, appID));

        assertInternalTopicsGotDeleted(null);

        // RE-RUN
        streams.start();
        final List<KeyValue<Long, Long>> resultRerun = IntegrationTestUtils.waitUntilMinKeyValueRecordsReceived(resultConsumerConfig, OUTPUT_TOPIC, 10);
        streams.close();

        assertThat(resultRerun, equalTo(result));
        cleanGlobal(false, "--force", null, appID);
    }
Tests local state store and global application cleanup.
streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationTest.java