@Test
public void testSendingOldValue() throws IOException {

    KStreamBuilder builder = new KStreamBuilder();
    String topic1 = "topic1";
    KTableImpl<String, String, String> table1 =
            (KTableImpl<String, String, String>) builder.table(stringSerde, stringSerde, topic1);
    KTableImpl<String, String, Integer> table2 = (KTableImpl<String, String, Integer>) table1.mapValues(
            new ValueMapper<String, Integer>() {
                @Override
                public Integer apply(String value) {
                    return new Integer(value);
                }
            });
    table2.enableSendingOldValues();
    MockProcessorSupplier<String, Integer> proc = new MockProcessorSupplier<>();
    builder.addProcessor("proc", proc, table2.name);
    driver = new KStreamTestDriver(builder, stateDir, null, null);
    assertTrue(table1.sendingOldValueEnabled());
    assertTrue(table2.sendingOldValueEnabled());
    driver.process(topic1, "A", "01");
    driver.process(topic1, "B", "01");
    driver.process(topic1, "C", "01");
    proc.checkAndClearProcessResult("A:(1<-null)", "B:(1<-null)", "C:(1<-null)");
    driver.process(topic1, "A", "02");
    driver.process(topic1, "B", "02");
    proc.checkAndClearProcessResult("A:(2<-1)", "B:(2<-1)");
    driver.process(topic1, "A", "03");
    proc.checkAndClearProcessResult("A:(3<-2)");
    driver.process(topic1, "A", null);
    proc.checkAndClearProcessResult("A:(null<-3)");
}