@Test
public void testValueGetter() throws IOException {

    final KStreamBuilder builder = new KStreamBuilder();
    String topic1 = "topic1";
    KTableImpl<String, String, String> table1 = (KTableImpl<String, String, String>) builder.table(stringSerde, stringSerde, topic1);
    KTableValueGetterSupplier<String, String> getterSupplier1 = table1.valueGetterSupplier();
    driver = new KStreamTestDriver(builder, stateDir, null, null);
    KTableValueGetter<String, String> getter1 = getterSupplier1.get();
    getter1.init(driver.context());
    driver.process(topic1, "A", "01");
    driver.process(topic1, "B", "01");
    driver.process(topic1, "C", "01");
    assertEquals("01", getter1.get("A"));
    assertEquals("01", getter1.get("B"));
    assertEquals("01", getter1.get("C"));
    driver.process(topic1, "A", "02");
    driver.process(topic1, "B", "02");
    assertEquals("02", getter1.get("A"));
    assertEquals("02", getter1.get("B"));
    assertEquals("01", getter1.get("C"));
    driver.process(topic1, "A", "03");
    assertEquals("03", getter1.get("A"));
    assertEquals("02", getter1.get("B"));
    assertEquals("01", getter1.get("C"));
    driver.process(topic1, "A", null);
    driver.process(topic1, "B", null);
    assertNull(getter1.get("A"));
    assertNull(getter1.get("B"));
    assertEquals("01", getter1.get("C"));
}