Flink Table API读取Kafka做窗口聚合无输出问题排查

两个核心问题导致窗口无输出、程序挂起：

• 阻塞式执行调用导致窗口逻辑未实际运行：Flink Table API的execute()方法是阻塞调用，启动作业后会持续运行直到作业主动结束。你代码中先调用了inputTable.execute().print()，这个作业对接的是无界的Kafka数据源，消费完现有4条数据后会持续等待新数据，永远不会主动退出，因此后续的窗口聚合作业代码根本没有被执行。
• 水位线无法推进导致窗口不触发：注释掉源表打印的execute之后，窗口作业启动但依然不会输出结果，根源是你的窗口触发依赖水位线。你定义的水位线策略为WATERMARK FOR ts AS ts，即水位线始终等于当前已处理数据的最大事件时间；而5秒滚动窗口要输出结果，必须等水位线超过窗口的结束时间。你现有4条测试数据的最大事件时间是2021-08-13 14:59:56.00，对应包含该数据的窗口是[2021-08-13 14:59:55, 2021-08-13 15:00:00)，窗口结束时间为15:00:00，消费完4条数据后Kafka Source没有新数据进入，水位线会一直停在14:59:56，达不到窗口触发阈值，因此窗口不会计算输出，程序也会持续挂起等待新数据。

• 移除独立的源表打印执行逻辑：不要在同一段代码中调用多个execute()方法，如果需要验证源表数据解析是否正确，单独写测试逻辑验证，不要和窗口聚合逻辑放在同一个执行流程中。
• 配置Source空闲超时，允许无新数据时推进水位线：修改Kafka表的WITH参数，添加空闲超时配置，同时可以根据需要设置合理的水位线乱序延迟，示例建表语句片段如下：

'scan.watermark.idle-timeout' = '3s',
WATERMARK FOR ts AS ts - INTERVAL '0' SECOND

该配置生效后，如果Kafka分区持续3秒没有新数据流入，Flink会将该分区标记为空闲，自动将全局水位线推进到当前已接收数据的最大事件时间，触发所有满足条件的窗口计算。

• 本地快速验证的临时方案：直接向Kafka的test主题写入一条事件时间大于等于2021-08-13 15:00:00.00的测试数据，手动推高水位线越过窗口结束时间，即可立即看到窗口计算结果。

• 开启本地WebUI观测作业状态：初始化执行环境时传入配置开启WebUI端口，启动作业后可以直接在浏览器查看各算子的消费进度、当前水位线值、窗口状态等信息，配置代码如下：

import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

// 初始化带WebUI的执行环境
Configuration conf = new Configuration();
conf.set(RestOptions.PORT, 8081);
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(conf);
// 设置水位线周期性生成间隔，默认200ms，本地调试可以设为1s减少开销
env.getConfig().setAutoWatermarkInterval(1000);
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

启动作业后访问localhost:8081即可进入Flink WebUI，在算子详情页可以直接看到当前水位线数值，快速判断窗口不触发是否由水位线问题导致。

• IDEA断点调试：直接在自定义函数、Watermark生成逻辑、窗口处理逻辑对应的代码行打断点，本地运行时可以直接查看当前流入的数据、水位线数值、状态缓存内容，定位逻辑问题。
• 单元测试优先用有界数据源：验证窗口聚合逻辑时，可以先使用本地内存的有界数据源做测试，逻辑跑通后再替换为Kafka等无界数据源，减少调试等待时间。

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