Flink事件时间水印疑问：处理时间窗口触发时机不符预期

我正在理解基于Event Time的Watermarks机制，参考Flink官方WordCount示例修改了代码，给事件添加时间戳并配置了允许20秒乱序的Watermark策略，事件格式为word;timestamp，Map函数生成包含word、计数1、时间戳的Tuple3。

测试发送6条test事件后结果正确，但在ProcessFunction中观察到：窗口是定义的40秒滚动窗口，Watermark计算正确（比最后一条事件时间戳少20秒），但窗口在处理时间到达窗口结束时间时立即触发，而非等待Watermark超过窗口结束时间（比如处理时间=窗口结束时间+20秒），这和窗口默认触发器的文档描述不符。

请问我哪里操作有误，或是对Watermarks的理解存在偏差？

我的代码 #

public class DataStreamJob {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        WatermarkStrategy<Tuple3<String, Integer, Long>> strategy = WatermarkStrategy
                .<Tuple3<String, Integer, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(20))
                .withTimestampAssigner((event, timestamp) -> event.f2);

        DataStream<Tuple2<String, Integer>> dataStream = env
                .socketTextStream("localhost", 9999)
                .map(new Splitter())
                .assignTimestampsAndWatermarks(strategy)
                .keyBy(value -> value.f0)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(40)))
                .process(new MyProcessWindowFunction());

        dataStream.print();

        env.execute("Window WordCount");
    }

    public static class Splitter extends RichMapFunction<String, Tuple3<String, Integer, Long>> {

        @Override
        public Tuple3<String, Integer, Long> map(String value) throws Exception {
            String[] word = value.split(";");
            return new Tuple3<String, Integer, Long>(word[0], 1, Long.parseLong(word[1]));
        }
    }

    public static class MyProcessWindowFunction extends ProcessWindowFunction<Tuple3<String, Integer, Long>, Tuple2<String, Integer>, String, TimeWindow> {

        @Override
        public void process(String s, ProcessWindowFunction<Tuple3<String, Integer, Long>, Tuple2<String, Integer>, String, TimeWindow>.Context context, Iterable<Tuple3<String, Integer, Long>> elements, Collector<Tuple2<String, Integer>> out) throws Exception {
            Integer sum = 0;
            for (Tuple3<String, Integer, Long> in : elements) {
                sum++;
            }

            out.collect(new Tuple2<String, Integer>(s, sum));
            Date date = new Date(context.window().getStart());
            Date date2 = new Date(context.window().getEnd());
            Date watermark = new Date(context.currentWatermark());
            Date processingTime = new Date(context.currentProcessingTime());
            System.out.println(context.currentWatermark());
            System.out.println("Processing Time: " + processingTime);
            Format format = new SimpleDateFormat("yyyy MM dd HH:mm:ss");
            System.out.println("Watermark: " + watermark);
            System.out.println("Start Window: " + format.format(date) + " End Window: " + format.format(date2));
        }
    }
}

问题原因与解决方案 #

核心问题在于你使用了处理时间窗口（TumblingProcessingTimeWindows），而非事件时间窗口：

• 处理时间窗口的触发逻辑完全依赖当前系统的处理时间，当处理时间到达窗口结束时间时就会触发计算，和Watermark、事件时间没有任何关系。
• 如果你想要基于Event Time和Watermark来触发窗口，需要使用事件时间窗口，即TumblingEventTimeWindows.of(Time.seconds(40))。

额外注意事项：

• 对于Flink 1.12之前的版本，需要显式设置时间语义为Event Time：env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);；1.12+版本已默认支持事件时间，无需额外配置。
• 事件时间窗口的默认触发器是EventTimeTrigger，它的触发条件是Watermark超过窗口的结束时间，这才是你预期的行为。

修改后的窗口配置代码片段：

.window(TumblingEventTimeWindows.of(Time.seconds(40)))

内容的提问来源于stack exchange，提问作者FDS