@ParameterizedTest
    @ValueSource(strings = {StreamsConfig.EXACTLY_ONCE, StreamsConfig.EXACTLY_ONCE_V2})
    public void shouldSurviveWithOneTaskAsStandby(final String eosConfig) throws Exception {
        IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
            inputTopic,
            Collections.singletonList(
                new KeyValue<>(0, 0)
            ),
            TestUtils.producerConfig(
                CLUSTER.bootstrapServers(),
                IntegerSerializer.class,
                IntegerSerializer.class,
                new Properties()
            ),
            10L
        );

        final String stateDirPath = TestUtils.tempDirectory(appId).getPath();

        final CountDownLatch instanceLatch = new CountDownLatch(1);

        streamInstanceOne = buildStreamWithDirtyStateDir(stateDirPath + "/" + appId + "-1/", instanceLatch, eosConfig);
        streamInstanceTwo = buildStreamWithDirtyStateDir(stateDirPath + "/" + appId + "-2/", instanceLatch, eosConfig);

        startApplicationAndWaitUntilRunning(asList(streamInstanceOne, streamInstanceTwo), Duration.ofSeconds(60));

        // Wait for the record to be processed
        assertTrue(instanceLatch.await(15, TimeUnit.SECONDS));

        streamInstanceOne.close();
        streamInstanceTwo.close();

        streamInstanceOne.cleanUp();
        streamInstanceTwo.cleanUp();
    }
An integration test to verify the conversion of a dirty-closed EOS
task towards a standby task is safe across restarts of the application.
streams/src/test/java/org/apache/kafka/streams/integration/StandbyTaskEOSIntegrationTest.java