Flink事件时间temporal join仅生效数秒，如何实现持续关联？

问题根因 #

• 右表（Debezium同步维度表）的Watermark长期停滞：当Postgres表长时间没有更新时，Debezium侧没有新消息流入，右表的Watermark不会向前推进。事件时间Temporal Join要求左表事件时间必须小于等于右表的Watermark才能触发关联，右表Watermark卡住后，左表新产生的事件无法完成匹配，就会出现仅更新后几秒能关联的现象。
• 右表启动模式不合理：你配置了scan.startup.mode = 'latest-offset'，如果启动前Postgres表已经有存量数据，Debezium不会同步这些历史数据到Flink，右表初始状态缺少存量维度值，也会导致关联匹配失败。
• 右表Watermark无乱序容忍：当前配置的Watermark是严格等于事件时间，若Debezium同步消息出现轻微乱序，会导致Watermark异常推进，部分正常事件被判定为迟到丢弃。

修复方案 #

1. 配置源空闲超时参数，避免右表无数据时卡住整体Watermark
   在作业初始化阶段添加如下配置，单位为毫秒：

tEnv.getConfig().set("table.exec.source.idle-timeout", "10000");

该配置表示如果右表超过10秒没有新数据流入，会被标记为空闲源，整体Watermark计算时会忽略该源，左表Watermark可以正常推进完成关联。
2. 调整右表Watermark策略，增加乱序容忍
修改TransportNetworkEdge_Kafka表的Watermark定义，允许5秒内的消息乱序：

WATERMARK FOR `timestamp` AS `timestamp` - INTERVAL '5' SECOND

3. 调整Debezium表启动模式，同步全量存量数据
   如果需要关联Postgres表的历史存量数据，需要先保证Debezium已经对Postgres表完成全量快照，将scan.startup.mode改为initial即可，该模式仅首次启动时执行全量同步，后续重启会从消费断点续传：

'scan.startup.mode' = 'initial'

内容的提问来源于stack exchange，提问作者E. Marotti