@Test public void testSendingOldValue() 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; } }); table2.enableSendingOldValues(); MockProcessorSupplier<String, Integer> proc1 = new MockProcessorSupplier<>(); MockProcessorSupplier<String, Integer> proc2 = new MockProcessorSupplier<>(); builder.addProcessor("proc1", proc1, table1.name); builder.addProcessor("proc2", proc2, table2.name); driver = new KStreamTestDriver(builder, stateDir, null, null); driver.process(topic1, "A", 1); driver.process(topic1, "B", 1); driver.process(topic1, "C", 1); proc1.checkAndClearProcessResult("A:(1<-null)", "B:(1<-null)", "C:(1<-null)"); proc2.checkEmptyAndClearProcessResult(); driver.process(topic1, "A", 2); driver.process(topic1, "B", 2); proc1.checkAndClearProcessResult("A:(2<-1)", "B:(2<-1)"); proc2.checkAndClearProcessResult("A:(2<-null)", "B:(2<-null)"); driver.process(topic1, "A", 3); proc1.checkAndClearProcessResult("A:(3<-2)"); proc2.checkAndClearProcessResult("A:(null<-2)"); driver.process(topic1, "A", null); driver.process(topic1, "B", null); proc1.checkAndClearProcessResult("A:(null<-3)", "B:(null<-2)"); proc2.checkAndClearProcessResult("B:(null<-2)"); }
@Override public void enableSendingOldValues() { parent.enableSendingOldValues(); sendOldValues = true; }
KTableFilter implements KTableProcessorSupplier<K, V, V> { @Override public void enableSendingOldValues() { parent.enableSendingOldValues(); sendOldValues = true; } }
KTableFilter implements KTableProcessorSupplier<K, V, V> { @Override public void enableSendingOldValues() { parent.enableSendingOldValues(); sendOldValues = true; } KTableFilter(KTableImpl<K, ?, V> parent, Predicate<K, V> predicate, boolean filterNot); }
KTableFilter implements KTableProcessorSupplier<K, V, V> { @Override public void enableSendingOldValues() { parent.enableSendingOldValues(); sendOldValues = true; } 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(); }
KTableFilter implements KTableProcessorSupplier<K, V, V> { @Override public void enableSendingOldValues() { parent.enableSendingOldValues(); sendOldValues = true; } 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();  }