@Test public void testSendingOldValue() throws IOException { KStreamBuilder builder = new KStreamBuilder(); String topic1 = "topic1"; KTableImpl<String, String, String> table1 = (KTableImpl<String, String, String>) builder.table(stringSerde, stringSerde, topic1, "anyStoreName"); 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); } }); table2.enableSendingOldValues(); MockProcessorSupplier<String, Integer> proc = new MockProcessorSupplier<>(); builder.addProcessor("proc", proc, table2.name); driver = new KStreamTestDriver(builder, stateDir, null, null); assertTrue(table1.sendingOldValueEnabled()); assertTrue(table2.sendingOldValueEnabled()); driver.process(topic1, "A", "01"); driver.process(topic1, "B", "01"); driver.process(topic1, "C", "01"); driver.flushState(); proc.checkAndClearProcessResult("A:(1<-null)", "B:(1<-null)", "C:(1<-null)"); driver.process(topic1, "A", "02"); driver.process(topic1, "B", "02"); driver.flushState(); proc.checkAndClearProcessResult("A:(2<-1)", "B:(2<-1)"); driver.process(topic1, "A", "03"); driver.flushState(); proc.checkAndClearProcessResult("A:(3<-2)"); driver.process(topic1, "A", null); driver.flushState(); proc.checkAndClearProcessResult("A:(null<-3)"); }
@Override public void enableSendingOldValues() { parent.enableSendingOldValues(); sendOldValues = true; }
KTableMapValues implements KTableProcessorSupplier<K, V, V1> { @Override public void enableSendingOldValues() { parent.enableSendingOldValues(); sendOldValues = true; } }
KTableMapValues implements KTableProcessorSupplier<K, V, V1> { @Override public void enableSendingOldValues() { parent.enableSendingOldValues(); sendOldValues = true; } KTableMapValues(final KTableImpl<K, ?, V> parent, final ValueMapper<? super V, ? extends V1> mapper,
                           final String queryableName); }
KTableMapValues implements KTableProcessorSupplier<K, V, V1> { @Override public void enableSendingOldValues() { parent.enableSendingOldValues(); sendOldValues = true; } KTableMapValues(final KTableImpl<K, ?, V> parent, final ValueMapper<? super V, ? extends V1> mapper,
                           final String queryableName); @Override Processor<K, Change<V>> get(); @Override KTableValueGetterSupplier<K, V1> view(); @Override void enableSendingOldValues(); }
KTableMapValues implements KTableProcessorSupplier<K, V, V1> { @Override public void enableSendingOldValues() { parent.enableSendingOldValues(); sendOldValues = true; } KTableMapValues(final KTableImpl<K, ?, V> parent, final ValueMapper<? super V, ? extends V1> mapper,
                           final String queryableName); @Override Processor<K, Change<V>> get(); @Override KTableValueGetterSupplier<K, V1> view(); @Override void enableSendingOldValues();  }