@Test
public void testTransform() {

    KStreamBuilder builder = new KStreamBuilder();
    TransformerSupplier<Integer, Integer, KeyValue<Integer, Integer>> transformerSupplier =
        new TransformerSupplier<Integer, Integer, KeyValue<Integer, Integer>>() {
            public Transformer<Integer, Integer, KeyValue<Integer, Integer>> get() {
                return new Transformer<Integer, Integer, KeyValue<Integer, Integer>>() {
                    private int total = 0;
                    @Override
                    public void init(ProcessorContext context) {
                    }
                    @Override
                    public KeyValue<Integer, Integer> transform(Integer key, Integer value) {
                        total += value;
                        return KeyValue.pair(key * 2, total);
                    }
                    @Override
                    public KeyValue<Integer, Integer> punctuate(long timestamp) {
                        return KeyValue.pair(-1, (int) timestamp);
                    }
                    @Override
                    public void close() {
                    }
                };
            }
        };
    final int[] expectedKeys = {1, 10, 100, 1000};
    MockProcessorSupplier<Integer, Integer> processor = new MockProcessorSupplier<>();
    KStream<Integer, Integer> stream = builder.stream(intSerde, intSerde, topicName);
    stream.transform(transformerSupplier).process(processor);
    driver = new KStreamTestDriver(builder);
    for (int i = 0; i < expectedKeys.length; i++) {
        driver.process(topicName, expectedKeys[i], expectedKeys[i] * 10);
    }
    driver.punctuate(2);
    driver.punctuate(3);
    assertEquals(6, processor.processed.size());
    String[] expected = {"2:10", "20:110", "200:1110", "2000:11110", "-1:2", "-1:3"};
    for (int i = 0; i < expected.length; i++) {
        assertEquals(expected[i], processor.processed.get(i));
    }
}