KTableMapValues implements KTableProcessorSupplier<K, V, V1> { @Override public Processor<K, Change<V>> get() { return new KTableMapValuesProcessor(); } KTableMapValues(KTableImpl<K, ?, V> parent, ValueMapper<V, V1> mapper); @Override Processor<K, Change<V>> get(); @Override KTableValueGetterSupplier<K, V1> view(); @Override void enableSendingOldValues();  }
@Test public void testValueGetter() throws IOException { KStreamBuilder builder = new KStreamBuilder(); String topic1 = "topic1"; String topic2 = "topic2"; KTableImpl<String, String, String> table1 = (KTableImpl<String, String, String>) builder.table(stringSerde, stringSerde, topic1); KTableImpl<String, String, Integer> table2 = (KTableImpl<String, String, Integer>) table1.mapValues( new ValueMapper<String, Integer>() { @Override public Integer apply(String value) { return new Integer(value); } }); KTableImpl<String, Integer, Integer> table3 = (KTableImpl<String, Integer, Integer>) table2.filter( new Predicate<String, Integer>() { @Override public boolean test(String key, Integer value) { return (value % 2) == 0; } }); KTableImpl<String, String, String> table4 = (KTableImpl<String, String, String>) table1.through(stringSerde, stringSerde, topic2); KTableValueGetterSupplier<String, String> getterSupplier1 = table1.valueGetterSupplier(); KTableValueGetterSupplier<String, Integer> getterSupplier2 = table2.valueGetterSupplier(); KTableValueGetterSupplier<String, Integer> getterSupplier3 = table3.valueGetterSupplier(); KTableValueGetterSupplier<String, String> getterSupplier4 = table4.valueGetterSupplier(); driver = new KStreamTestDriver(builder, stateDir, null, null); KTableValueGetter<String, String> getter1 = getterSupplier1.get(); getter1.init(driver.context()); KTableValueGetter<String, Integer> getter2 = getterSupplier2.get(); getter2.init(driver.context()); KTableValueGetter<String, Integer> getter3 = getterSupplier3.get(); getter3.init(driver.context()); KTableValueGetter<String, String> getter4 = getterSupplier4.get(); getter4.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")); assertEquals(new Integer(1), getter2.get("A")); assertEquals(new Integer(1), getter2.get("B")); assertEquals(new Integer(1), getter2.get("C")); assertNull(getter3.get("A")); assertNull(getter3.get("B")); assertNull(getter3.get("C")); assertEquals("01", getter4.get("A")); assertEquals("01", getter4.get("B")); assertEquals("01", getter4.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")); assertEquals(new Integer(2), getter2.get("A")); assertEquals(new Integer(2), getter2.get("B")); assertEquals(new Integer(1), getter2.get("C")); assertEquals(new Integer(2), getter3.get("A")); assertEquals(new Integer(2), getter3.get("B")); assertNull(getter3.get("C")); assertEquals("02", getter4.get("A")); assertEquals("02", getter4.get("B")); assertEquals("01", getter4.get("C")); driver.process(topic1, "A", "03"); assertEquals("03", getter1.get("A")); assertEquals("02", getter1.get("B")); assertEquals("01", getter1.get("C")); assertEquals(new Integer(3), getter2.get("A")); assertEquals(new Integer(2), getter2.get("B")); assertEquals(new Integer(1), getter2.get("C")); assertNull(getter3.get("A")); assertEquals(new Integer(2), getter3.get("B")); assertNull(getter3.get("C")); assertEquals("03", getter4.get("A")); assertEquals("02", getter4.get("B")); assertEquals("01", getter4.get("C")); driver.process(topic1, "A", null); assertNull(getter1.get("A")); assertEquals("02", getter1.get("B")); assertEquals("01", getter1.get("C")); assertNull(getter2.get("A")); assertEquals(new Integer(2), getter2.get("B")); assertEquals(new Integer(1), getter2.get("C")); assertNull(getter3.get("A")); assertEquals(new Integer(2), getter3.get("B")); assertNull(getter3.get("C")); assertNull(getter4.get("A")); assertEquals("02", getter4.get("B")); assertEquals("01", getter4.get("C")); }