@SuppressWarnings("deprecation")
    private KafkaStreams buildWithDeduplicationTopology(final String stateDirPath, final String eosConfig) {
        final StreamsBuilder builder = new StreamsBuilder();

        builder.addStateStore(Stores.keyValueStoreBuilder(
            Stores.persistentKeyValueStore(storeName),
            Serdes.Integer(),
            Serdes.Integer())
        );
        builder.<Integer, Integer>stream(inputTopic)
            .transform(
                () -> new Transformer<Integer, Integer, KeyValue<Integer, Integer>>() {
                    private KeyValueStore<Integer, Integer> store;

                    @SuppressWarnings("unchecked")
                    @Override
                    public void init(final ProcessorContext context) {
                        store = (KeyValueStore<Integer, Integer>) context.getStateStore(storeName);
                    }

                    @Override
                    public KeyValue<Integer, Integer> transform(final Integer key, final Integer value) {
                        if (skipRecord.get()) {
                            // we only forward so we can verify the skipping by reading the output topic
                            // the goal is skipping is to not modify the state store
                            return KeyValue.pair(key, value);
                        }

                        if (store.get(key) != null) {
                            return null;
                        }

                        store.put(key, value);
                        store.flush();

                        if (key == KEY_1) {
                            // after error injection, we need to avoid a consecutive error after rebalancing
                            skipRecord.set(true);
                            throw new RuntimeException("Injected test error");
                        }

                        return KeyValue.pair(key, value);
                    }

                    @Override
                    public void close() { }
                },
                storeName
            )
            .to(outputTopic);

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