使用Flink Table消费Kafka消息无控制台输出问题排查

问题背景 #

需求为每10分钟统计amount字段总和，基于Flink SQL消费Kafka数据实现，代码如下：

StreamExecutionEnvironment env =  StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

String create_kafka_table = "create table orders(" +
        "    order_id string," +
        "    order_item string," +
        "    amount int," +
        "    order_time timestamp(0)," +
        "    watermark for order_time as order_time - interval '1' minute" +
        ") with (" +
        "  'connector' = 'kafka'," +
        "  'topic' = 'order_flow'," +
        "  'properties.bootstrap.servers' = '127.0.0.1:9093'," +
        "  'properties.group.id' = 'order_consumer'," +
        "  'scan.startup.mode' = 'earliest-offset'," +
        "  'format' = 'csv'" +
        ")";

tenv.executeSql(create_kafka_table);

tenv.executeSql("select window_start, window_end, sum(amount)  " +
        " FROM TABLE(" +
        "    TUMBLE(TABLE orders, DESCRIPTOR(order_time), INTERVAL '10' MINUTES))" +
        "  GROUP BY window_start, window_end"
).print();

Kafka测试数据：

1,a,10,2022-11-05 10:35:00
1,a,10,2022-11-05 10:38:00
1,a,10,2022-11-05 10:40:00
1,a,10,2022-11-05 10:50:00
1,a,10,2022-11-05 10:52:00
1,a,10,2022-11-05 10:55:00
1,a,10,2022-11-05 10:55:00
1,a,10,2022-11-05 11:00:00

现象：WebUI显示GlobalWindowAggregate任务已接收记录，但无控制台输出。

排查与解决方法 #

• 核心原因：水位线未推进到窗口结束时间
  Flink事件时间滚动窗口的触发条件是水位线（Watermark）超过窗口结束时间。当前定义的水位线规则为order_time - interval '1' minute，即水位线始终比最新事件时间晚1分钟：

  • 10:30-10:40的窗口，需要水位线≥10:40才会触发，但现有数据中最晚事件时间为10:40时，水位线仅为10:39，未达触发条件；
  • 10:40-10:50的窗口，最晚事件时间为10:50时，水位线为10:49，同样未达触发条件；
  • 10:50-11:00的窗口，最晚事件时间为11:00时，水位线为10:59，仍未触发。

• 解决方案1：调整水位线延迟
  修改水位线定义，将延迟设为0分钟，这样当有窗口结束时间的事件进入时，水位线会直接推进到窗口结束时间，触发计算：

  "    watermark for order_time as order_time - interval '0' minute"
  

• 解决方案2：补充触发窗口的测试数据
  在Kafka中添加晚于窗口结束时间1分钟的事件，比如：

  1,a,10,2022-11-05 10:41:00  // 触发10:30-10:40窗口
  1,a,10,2022-11-05 11:01:00  // 触发10:50-11:00窗口
  

• 额外验证点
  确认Kafka数据中的order_time格式符合timestamp(0)要求（秒级时间字符串），确保Flink能正确解析事件时间，避免因时间解析错误导致水位线无法推进。

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