@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, "anyStoreName"); 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")); }
@Override public Processor<K, V> get() { return new KTableSourceProcessor(); }
KTableSource implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KTableSourceProcessor(); } }
KTableSource implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KTableSourceProcessor(); } KTableSource(String storeName); }
KTableSource implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KTableSourceProcessor(); } KTableSource(String storeName); @Override Processor<K, V> get(); void enableSendingOldValues(); }
KTableSource implements ProcessorSupplier<K, V> { @Override public Processor<K, V> get() { return new KTableSourceProcessor(); } KTableSource(String storeName); @Override Processor<K, V> get(); void enableSendingOldValues(); final String storeName; }