@ParameterizedTest
    @ValueSource(strings = {StreamsConfig.EXACTLY_ONCE, StreamsConfig.EXACTLY_ONCE_V2})
    public void shouldWipeOutStandbyStateDirectoryIfCheckpointIsMissing(final String eosConfig) throws Exception {
        final long time = System.currentTimeMillis();
        final String base = TestUtils.tempDirectory(appId).getPath();

        IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
            inputTopic,
            Collections.singletonList(
                new KeyValue<>(KEY_0, 0)
            ),
            TestUtils.producerConfig(
                CLUSTER.bootstrapServers(),
                IntegerSerializer.class,
                IntegerSerializer.class,
                new Properties()
            ),
            10L + time
        );

        streamInstanceOne = buildWithDeduplicationTopology(base + "-1", eosConfig);
        streamInstanceTwo = buildWithDeduplicationTopology(base + "-2", eosConfig);

        // start first instance and wait for processing
        startApplicationAndWaitUntilRunning(streamInstanceOne);
        IntegrationTestUtils.waitUntilMinRecordsReceived(
            TestUtils.consumerConfig(
                CLUSTER.bootstrapServers(),
                IntegerDeserializer.class,
                IntegerDeserializer.class
            ),
            outputTopic,
            1
        );

        // start second instance and wait for standby replication
        startApplicationAndWaitUntilRunning(streamInstanceTwo);
        waitForCondition(
            () -> streamInstanceTwo.store(
                StoreQueryParameters.fromNameAndType(
                    storeName,
                    QueryableStoreTypes.<Integer, Integer>keyValueStore()
                ).enableStaleStores()
            ).get(KEY_0) != null,
            REBALANCE_TIMEOUT,
            "Could not get key from standby store"
        );
        // sanity check that first instance is still active
        waitForCondition(
            () -> streamInstanceOne.store(
                StoreQueryParameters.fromNameAndType(
                    storeName,
                    QueryableStoreTypes.<Integer, Integer>keyValueStore()
                )
            ).get(KEY_0) != null,
            "Could not get key from main store"
        );

        // inject poison pill and wait for crash of first instance and recovery on second instance
        IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
            inputTopic,
            Collections.singletonList(
                new KeyValue<>(KEY_1, 0)
            ),
            TestUtils.producerConfig(
                CLUSTER.bootstrapServers(),
                IntegerSerializer.class,
                IntegerSerializer.class,
                new Properties()
            ),
            10L + time
        );
        waitForCondition(
            () -> streamInstanceOne.state() == KafkaStreams.State.ERROR,
            "Stream instance 1 did not go into error state"
        );
        streamInstanceOne.close();

        IntegrationTestUtils.waitUntilMinRecordsReceived(
            TestUtils.consumerConfig(
                CLUSTER.bootstrapServers(),
                IntegerDeserializer.class,
                IntegerDeserializer.class
            ),
            outputTopic,
            2
        );

        streamInstanceOneRecovery = buildWithDeduplicationTopology(base + "-1", eosConfig);

        // "restart" first client and wait for standby recovery
        // (could actually also be active, but it does not matter as long as we enable "state stores"
        startApplicationAndWaitUntilRunning(streamInstanceOneRecovery);
        waitForCondition(
            () -> streamInstanceOneRecovery.store(
                StoreQueryParameters.fromNameAndType(
                    storeName,
                    QueryableStoreTypes.<Integer, Integer>keyValueStore()
                ).enableStaleStores()
            ).get(KEY_0) != null,
            "Could not get key from recovered standby store"
        );

        streamInstanceTwo.close();
        waitForCondition(
            () -> streamInstanceOneRecovery.store(
                StoreQueryParameters.fromNameAndType(
                    storeName,
                    QueryableStoreTypes.<Integer, Integer>keyValueStore()
                )
            ).get(KEY_0) != null,
            REBALANCE_TIMEOUT,
            "Could not get key from recovered main store"
        );

        // re-inject poison pill and wait for crash of first instance
        skipRecord.set(false);
        IntegrationTestUtils.produceKeyValuesSynchronouslyWithTimestamp(
            inputTopic,
            Collections.singletonList(
                new KeyValue<>(KEY_1, 0)
            ),
            TestUtils.producerConfig(
                CLUSTER.bootstrapServers(),
                IntegerSerializer.class,
                IntegerSerializer.class,
                new Properties()
            ),
            10L + time
        );
        waitForCondition(
            () -> streamInstanceOneRecovery.state() == KafkaStreams.State.ERROR,
            "Stream instance 1 did not go into error state. Is in " + streamInstanceOneRecovery.state() + " state."
        );
    }
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