Hi guys, I'm confused about watermarkers. Now I have data stream A, then split A to B, C, D with a filter transform , but only A has data, all of them will do a window aggretator. Now the question is: Do I have set a idleness duration to this stream? Cause I found there is no watermarks on A, and can't trigger window

I am not sure what you mean by splitting the stream. I think creating side outputs splits the streams, and that cannot be done with a Filter operator. https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/dev/datastream/side_output/

like this, split the stream element into multiple elements of the same kind

final SingleOutputStreamOperator<SameBean> b = a
      .filter(e->"b".equals(e.getType()));
final SingleOutputStreamOperator<SameBean> c = a
      .filter(e->"c".equals(e.getType()));

'b' and 'c' have different timestamp delay, I want make b's window OutOfOrderness longer than a . then I do :

b.assignTimestampsAndWatermarks(WatermarkStrategy.<SameBean>forBoundedOutOfOrderness(Duration.ofSeconds(120)).withTimestampAssigner((event, timestamp)->event.getTimeStamp()).withIdleness(Duration.ofSeconds(10)))
.keyBy(input -> (input.getKey()))
.window(TumblingEventTimeWindows.of(Time.seconds(interval)))
.aggregate(new TagAggAggregator());

c.assignTimestampsAndWatermarks(WatermarkStrategy.<SameBean>forBoundedOutOfOrderness(Duration.ofSeconds(30)).withTimestampAssigner((event, timestamp)->event.getTimeStamp()).withIdleness(Duration.ofSeconds(10)))
.keyBy(input -> (input.getKey()))
.window(TumblingEventTimeWindows.of(Time.seconds(interval)))
.aggregate(new TagAggAggregator());

The only difference is the forBoundedOutOfOrderness(Duration.ofSeconds(30/120)) . I have some test, after add withIdleness, the wartermark is working. or is there any other better practice for this situation?

What is your datasource that lead to “a”?

flink consume kafka as "a". but "a" has some different type, and I need to split "a" into "b" and "c"