@Test
public void testSendingOldValue() throws IOException {

    KStreamBuilder builder = new KStreamBuilder();
    String topic1 = "topic1";
    KTableImpl<String, Integer, Integer> table1 =
            (KTableImpl<String, Integer, Integer>) builder.table(stringSerde, intSerde, topic1);
    KTableImpl<String, Integer, Integer> table2 = (KTableImpl<String, Integer, Integer>) table1.filter(
            new Predicate<String, Integer>() {
                @Override
                public boolean test(String key, Integer value) {
                    return (value % 2) == 0;
                }
            });
    table2.enableSendingOldValues();
    MockProcessorSupplier<String, Integer> proc1 = new MockProcessorSupplier<>();
    MockProcessorSupplier<String, Integer> proc2 = new MockProcessorSupplier<>();
    builder.addProcessor("proc1", proc1, table1.name);
    builder.addProcessor("proc2", proc2, table2.name);
    driver = new KStreamTestDriver(builder, stateDir, null, null);
    driver.process(topic1, "A", 1);
    driver.process(topic1, "B", 1);
    driver.process(topic1, "C", 1);
    proc1.checkAndClearProcessResult("A:(1<-null)", "B:(1<-null)", "C:(1<-null)");
    proc2.checkEmptyAndClearProcessResult();
    driver.process(topic1, "A", 2);
    driver.process(topic1, "B", 2);
    proc1.checkAndClearProcessResult("A:(2<-1)", "B:(2<-1)");
    proc2.checkAndClearProcessResult("A:(2<-null)", "B:(2<-null)");
    driver.process(topic1, "A", 3);
    proc1.checkAndClearProcessResult("A:(3<-2)");
    proc2.checkAndClearProcessResult("A:(null<-2)");
    driver.process(topic1, "A", null);
    driver.process(topic1, "B", null);
    proc1.checkAndClearProcessResult("A:(null<-3)", "B:(null<-2)");
    proc2.checkAndClearProcessResult("B:(null<-2)");
}