Flink事件时间窗口提前触发与非确定性结果问题排查

场景说明 #

读取Kafka存储的小样本数据，未在数据源直接设置Watermark，而是先对数据做预处理再提取事件时间，随后设置5秒的事件时间滑动窗口。多数情况下结果正常，但有时窗口会提前触发，部分时间戳在窗口范围内的事件未被纳入（仅测试单个窗口）。

相关代码 #

// seconds
int windowSize = 5;
int windowSlide = 5;
int windowOffset = 0;
int envParallelism = 5;

// read data from kafkasource
DataStream<String> eventstream =
        env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source").setParallelism(1);

// some preprocessing, produce inValue tuples
eventstream = eventstream.flatMap(new JsonFlatMap()); 

WatermarkStrategy watermarkStrategy = WatermarkStrategy
        .forBoundedOutOfOrderness(Duration.ofSeconds(maxOutOfOrderness));

WatermarkStrategy watermarkStrategyWithTimestampAssigner = watermarkStrategy
        .withTimestampAssigner(
                new SerializableTimestampAssigner<Event>() {
                    @Override
                    public long extractTimestamp(Event event, long recordTimestamp) {
                        return event.time;
                    }
                });
eventStream = eventStream.assignTimestampsAndWatermarks(watermarkStrategyWithTimestampAssigner).setParallelism(1);

eventStream.print();

// --for every "eventName"(key), add "inValue" to "inValueList" and "outValue" to "outValueList"　
DataStream<Model> output = eventStream
        .keyBy(new EventKeySelector("eventName"))
        .window(SlidingEventTimeWindows.of(
                Time.seconds(windowSize), Time.seconds(windowSlide),
                Time.seconds(windowOffset)))
        .process(new GraphProcessWindowFunction());
output.print();

测试数据（Kafka存储） #

{"eventName":1,"outValue":2,"time":1743663199798} 
{"eventName":2,"outValue":1,"time":1743663199908}
{"eventName":2,"outValue":6,"time":1743663199911}
{"eventName":3,"outValue":6,"time":1743663199914} 
{"eventName":6,"outValue":1,"time":1743663199918}
{"eventName":6,"outValue":3,"time":1743663199920} 
{"eventName":2,"outValue":6,"time":1743663207928}
{"eventName":3,"outValue":6,"time":1743663207933}

注：最后两条事件间隔8000ms生成，本不应被纳入窗口。

异常现象与已尝试方案 #

• 异常表现：部分时间戳符合窗口范围的事件未被分配到窗口；设置allowedLateness(>3秒)时，未被分配的事件会触发窗口二次输出，但无需处理二次输出。
• 已尝试操作：将forBoundedOutOfOrderness的maxOutOfOrderness设为5秒、调整env.getConfig().setAutoWatermarkInterval(1)，均无效果；各窗口的currentwatermark()始终固定为1743663202933。

疑问 #

1. 未在数据源直接设置Watermark是否会导致该问题？
2. 如何解决窗口提前触发、事件漏处理的问题？

环境：Flink v1.16.0，Java 11

内容的提问来源于stack exchange，提问作者K.M