KTableFilter implements KTableProcessorSupplier<K, V, V> { @Override public Processor<K, Change<V>> get() { return new KTableFilterProcessor(); } KTableFilter(KTableImpl<K, ?, V> parent, Predicate<K, V> predicate, boolean filterNot); @Override Processor<K, Change<V>> get(); @Override KTableValueGetterSupplier<K, V> view(); @Override void enableSendingOldValues();  }
@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")); }