@Log4j2
@Component
@EnableBinding(MySink.class)
public class JoinConsumer {
@StreamListener
public void joinConsumers(@Input(MySink.INPUT_1) KStream<String, String> input1,
@Input(MySink.INPUT_2) KStream<String, String> input2) {
input1.filter((k,v) -> v.equals("done")).join(input2, (v1, v2) -> v2, JoinWindows.of(
Duration.ofHours(24))).foreach((k, v) ->log.info("k: {}, v: {}", k, v));
}
}
代码按预期运行,但我不知道如何为此编写单元测试。
暂无答案!
目前还没有任何答案,快来回答吧!