@Test
public void testSedingOldValue() throws IOException {

    final KStreamBuilder builder = new KStreamBuilder();
    String topic1 = "topic1";
    KTableImpl<String, String, String> table1 = (KTableImpl<String, String, String>) builder.table(stringSerde, stringSerde, topic1);
    table1.enableSendingOldValues();
    assertTrue(table1.sendingOldValueEnabled());
    MockProcessorSupplier<String, Integer> proc1 = new MockProcessorSupplier<>();
    builder.addProcessor("proc1", proc1, table1.name);
    driver = new KStreamTestDriver(builder, stateDir, null, null);
    driver.process(topic1, "A", "01");
    driver.process(topic1, "B", "01");
    driver.process(topic1, "C", "01");
    proc1.checkAndClearProcessResult("A:(01<-null)", "B:(01<-null)", "C:(01<-null)");
    driver.process(topic1, "A", "02");
    driver.process(topic1, "B", "02");
    proc1.checkAndClearProcessResult("A:(02<-01)", "B:(02<-01)");
    driver.process(topic1, "A", "03");
    proc1.checkAndClearProcessResult("A:(03<-02)");
    driver.process(topic1, "A", null);
    driver.process(topic1, "B", null);
    proc1.checkAndClearProcessResult("A:(null<-03)", "B:(null<-02)");
}