如何使用Flink Table/SQL API处理Kafka数据源水印的Idleness问题

Flink Table/SQL API 提供了和DataStream API withIdleness完全对等的配置能力，可用来解决Kafka数据源的水印空闲问题，具体实现分为两种方式：

全局生效配置 #

通过配置Table环境全局参数，所有配置了水印策略的数据源都会生效空闲检测逻辑：

• Java代码配置示例：

TableEnvironment tEnv = TableEnvironment.create(EnvironmentSettings.inStreamingMode());
// 配置空闲超时时间为1分钟
tEnv.getConfig().set("table.exec.source.idle-timeout", "1min");

• SQL客户端/动态参数配置示例：

SET table.exec.source.idle-timeout = '1min';

该参数生效后，任意数据源的分区在指定时长内没有数据流入时，会被标记为空闲分区，整体水印推进时不会再等待该分区的事件时间，避免单个分区断流导致全局水印停滞。

单Kafka源单独配置 #

如果仅需要针对指定Kafka数据源生效，不需要全局配置，可以在创建Kafka表时指定连接器级别的scan.idle-timeout参数，优先级高于全局配置：

CREATE TABLE kafka_event_table (
    event_id STRING,
    ts TIMESTAMP(3),
    -- 定义20秒乱序容忍的水印策略，和DataStream的forBoundedOutOfOrderness(Duration.ofSeconds(20))效果一致
    WATERMARK FOR ts AS ts - INTERVAL '20' SECOND
) WITH (
    'connector' = 'kafka',
    'topic' = 'your_topic_name',
    'properties.bootstrap.servers' = 'your_kafka_address:9092',
    'format' = 'json',
    -- 单独配置当前Kafka源的空闲超时时间为1分钟，和DataStream的withIdleness(Duration.ofMinutes(1))效果一致
    'scan.idle-timeout' = '1min'
);

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