private KafkaStreams buildStreamWithDirtyStateDir(final String stateDirPath,
                                                      final CountDownLatch recordProcessLatch,
                                                      final String eosConfig) throws Exception {

        final StreamsBuilder builder = new StreamsBuilder();
        final TaskId taskId = new TaskId(0, 0);

        final Properties props = props(stateDirPath, eosConfig);

        final StateDirectory stateDirectory = new StateDirectory(
            new StreamsConfig(props), new MockTime(), true, false);

        new OffsetCheckpoint(new File(stateDirectory.getOrCreateDirectoryForTask(taskId), ".checkpoint"))
            .write(Collections.singletonMap(new TopicPartition("unknown-topic", 0), 5L));

        assertTrue(new File(stateDirectory.getOrCreateDirectoryForTask(taskId),
                            "rocksdb/KSTREAM-AGGREGATE-STATE-STORE-0000000001").mkdirs());

        builder.stream(inputTopic,
                       Consumed.with(Serdes.Integer(), Serdes.Integer()))
               .groupByKey()
               .count()
               .toStream()
               .peek((key, value) -> recordProcessLatch.countDown());

        return new KafkaStreams(builder.build(), props);
    }
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