@Test
public void testValueGetter() 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;
                }
            });
    KTableImpl<String, Integer, Integer> table3 = (KTableImpl<String, Integer, Integer>) table1.filterNot(
            new Predicate<String, Integer>() {
                @Override
                public boolean test(String key, Integer value) {
                    return (value % 2) == 0;
                }
            });
    KTableValueGetterSupplier<String, Integer> getterSupplier2 = table2.valueGetterSupplier();
    KTableValueGetterSupplier<String, Integer> getterSupplier3 = table3.valueGetterSupplier();
    driver = new KStreamTestDriver(builder, stateDir, null, null);
    KTableValueGetter<String, Integer> getter2 = getterSupplier2.get();
    KTableValueGetter<String, Integer> getter3 = getterSupplier3.get();
    getter2.init(driver.context());
    getter3.init(driver.context());
    driver.process(topic1, "A", 1);
    driver.process(topic1, "B", 1);
    driver.process(topic1, "C", 1);
    assertNull(getter2.get("A"));
    assertNull(getter2.get("B"));
    assertNull(getter2.get("C"));
    assertEquals(1, (int) getter3.get("A"));
    assertEquals(1, (int) getter3.get("B"));
    assertEquals(1, (int) getter3.get("C"));
    driver.process(topic1, "A", 2);
    driver.process(topic1, "B", 2);
    assertEquals(2, (int) getter2.get("A"));
    assertEquals(2, (int) getter2.get("B"));
    assertNull(getter2.get("C"));
    assertNull(getter3.get("A"));
    assertNull(getter3.get("B"));
    assertEquals(1, (int) getter3.get("C"));
    driver.process(topic1, "A", 3);
    assertNull(getter2.get("A"));
    assertEquals(2, (int) getter2.get("B"));
    assertNull(getter2.get("C"));
    assertEquals(3, (int) getter3.get("A"));
    assertNull(getter3.get("B"));
    assertEquals(1, (int) getter3.get("C"));
    driver.process(topic1, "A", null);
    driver.process(topic1, "B", null);
    assertNull(getter2.get("A"));
    assertNull(getter2.get("B"));
    assertNull(getter2.get("C"));
    assertNull(getter3.get("A"));
    assertNull(getter3.get("B"));
    assertEquals(1, (int) getter3.get("C"));
}