@Test public void testSendingOldValue() 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"); table1.enableSendingOldValues(); assertTrue(table1.sendingOldValueEnabled()); MockProcessorSupplier<String, Integer> proc1 = new MockProcessorSupplier<>(); builder.addProcessor("proc1", proc1, table1.name); driver = new KStreamTestDriver(builder, stateDir, null, null); driver.process(topic1, "A", "01"); driver.process(topic1, "B", "01"); driver.process(topic1, "C", "01"); driver.flushState(); proc1.checkAndClearProcessResult("A:(01<-null)", "B:(01<-null)", "C:(01<-null)"); driver.process(topic1, "A", "02"); driver.process(topic1, "B", "02"); driver.flushState(); proc1.checkAndClearProcessResult("A:(02<-01)", "B:(02<-01)"); driver.process(topic1, "A", "03"); driver.flushState(); proc1.checkAndClearProcessResult("A:(03<-02)"); driver.process(topic1, "A", null); driver.process(topic1, "B", null); driver.flushState(); proc1.checkAndClearProcessResult("A:(null<-03)", "B:(null<-02)"); }
public void enableSendingOldValues() { sendOldValues = true; }
KTableSource implements ProcessorSupplier<K, V> { public void enableSendingOldValues() { sendOldValues = true; } }
KTableSource implements ProcessorSupplier<K, V> { public void enableSendingOldValues() { sendOldValues = true; } KTableSource(String storeName); }
KTableSource implements ProcessorSupplier<K, V> { public void enableSendingOldValues() { sendOldValues = true; } KTableSource(String storeName); @Override Processor<K, V> get(); void enableSendingOldValues(); }
KTableSource implements ProcessorSupplier<K, V> { public void enableSendingOldValues() { sendOldValues = true; } KTableSource(String storeName); @Override Processor<K, V> get(); void enableSendingOldValues(); final String storeName; }