使用COUNT OVER PARTITION时总是缺失最新Kafka记录

问题场景 #

本地Linux环境测试Flink 1.21时，使用仅携带Header的Kafka消息作为数据源，执行带OVER窗口的COUNT聚合查询，发现最后一条发送到Kafka的消息始终无法出现在查询结果中。

相关定义与语句 #

Kafka源表定义 #

CREATE TEMPORARY TABLE person
(
    payload STRING,
    `headers` MAP<STRING, BYTES> METADATA,
    `record_time` TIMESTAMP_LTZ(3) METADATA FROM 'timestamp',
    WATERMARK FOR record_time AS record_time - INTERVAL '5' SECOND
) WITH (
      'connector' = 'kafka',
      'properties.bootstrap.servers' = 'localhost:9092', 
      'properties.group.id' = 'test1',
      'topic' = 'person',
      'scan.startup.mode' = 'earliest-offset',
      'properties.enableAutoCommit' = 'true',
      'properties.isolationLevel' = 'read_committed',
      'properties.maxWaitTimeInSeconds' = '12',
      'format' = 'raw');


create view person_view as select 
    cast(headers['id'] as varchar) as id,
    cast(headers['name'] as varchar) as name,
    record_time from person;

测试Kafka消息（仅含Header） #

{
    "id": "xyz1",
    "name": "test"
}

执行的查询语句 #

SELECT id, COUNT(*) OVER (PARTITION BY id order by record_time) AS r_count FROM person_view

原因分析 #

问题核心在于事件时间驱动的OVER窗口依赖水印推进触发计算：

• 定义的水印规则为record_time - INTERVAL '5' SECOND，只有当水印时间超过某条消息的record_time + 5秒时，这条消息对应的窗口计算才会被触发并输出结果。
• 最后一条消息发送后，没有后续事件产生更大的record_time来推动水印超过它的record_time + 5秒，因此窗口无法完成计算，结果不会输出。
• Kafka消费者的maxWaitTimeInSeconds仅控制拉取消息的等待时长，无法触发事件时间窗口的计算逻辑。

解决方法 #

1. 改用处理时间驱动（适合对时间精度要求不高的场景） #

修改源表定义，使用处理时间替代事件时间，水印基于处理时间生成：

CREATE TEMPORARY TABLE person
(
    payload STRING,
    `headers` MAP<STRING, BYTES> METADATA,
    `proc_time` AS PROCTIME(), -- 生成处理时间
    WATERMARK FOR proc_time AS proc_time - INTERVAL '5' SECOND
) WITH (
      'connector' = 'kafka',
      'properties.bootstrap.servers' = 'localhost:9092', 
      'properties.group.id' = 'test1',
      'topic' = 'person',
      'scan.startup.mode' = 'earliest-offset',
      'properties.enableAutoCommit' = 'true',
      'properties.isolationLevel' = 'read_committed',
      'properties.maxWaitTimeInSeconds' = '12',
      'format' = 'raw');

create view person_view as select 
    cast(headers['id'] as varchar) as id,
    cast(headers['name'] as varchar) as name,
    proc_time from person;

查询语句调整为按处理时间排序：

SELECT id, COUNT(*) OVER (PARTITION BY id order by proc_time) AS r_count FROM person_view

2. 触发水印推进（事件时间场景） #

手动发送一条事件时间远大于最后一条消息的测试消息（比如比最后一条的record_time大6秒以上），推动水印超过目标消息的record_time + 5秒，触发窗口计算。

3. 启用窗口提前触发 #

在Flink SQL客户端配置窗口提前触发规则，即使水印未达最终时间，也会定期输出中间结果：

SET table.exec.emit.early-fire.enabled = true;
SET table.exec.emit.early-fire.delay = 1s; -- 每1秒触发一次中间结果输出

4. 验证消息时间戳有效性 #

确认Kafka消息的timestamp字段已正确设置，无时间倒退或无效值，避免水印无法正常推进。

内容的提问来源于stack exchange，提问作者王子1986