马老师工作流引擎-activiti
KTable<Windowed<String>, List<String>> kTable = kStream.groupBy((k, v) -> "defaultKey")
.windowedBy(TimeWindows.of(windowSize).grace(Duration.ZERO))
.aggregate(() -> new ArrayList<>(), (k, v, agg) -> {
System.out.println("========== aggregate record ==========");
log.info("k: {}, v: {}, agg: {}", k, v, agg);
if (!signal.equals(v)) {
agg.add(v);
}
aggRecordMap.put(lastMsgTimeKey, System.currentTimeMillis());
