@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);
                    }
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