Flink替换KafkaSource后Interval Join丢弃所有记录问题咨询

核心结论 #

先直接回应两个咨询问题：

• Flink 1.14.4版本中新版KafkaSource与旧版废弃FlinkKafkaConsumer在Join场景下的水印传播逻辑确实存在行为差异，是本次异常的直接诱因。
• Interval Join算子本身不存在“取两个输入流水印最大值”的逻辑，出现该现象的唯一原因是其中一条输入流被Flink运行时判定为空闲（idle），算子计算输出水印时直接忽略了该流的水印值。

根因拆解 #

• 包括Interval Join在内的所有双输入算子，原生水印计算规则为：输出水印 = 所有处于活跃状态的输入流的当前水印最小值。如果某条输入流被判定为空闲，计算时会直接跳过该流的水印，仅取剩余活跃流的水印作为输出。
• 你的场景中最终输出水印等于Kafka Source的水印（比左侧流快约1分钟），说明左侧经过长链路处理的DataStream被临时判定为空闲流，水印计算时被排除，因此输出水印直接取了Kafka流的高水印，左侧所有数据因为时间戳小于当前算子水印被直接丢弃。
• 新旧Kafka连接器的具体行为差异：
  • 旧版FlinkKafkaConsumer基于旧版SourceFunction API实现，即使没有新消息可消费，也会按照pipeline.auto-watermark-interval配置的周期（默认200ms）持续向下游发送当前水印的更新消息，下游算子永远不会因为长时间收不到某条流的消息，误将对应输入标记为空闲。
  • 新版KafkaSource基于FLIP-27新Source API实现，在1.14.x版本中默认行为与旧版不一致：当消费的分区数据间隔较长（你的场景为1分钟/条，分区最大空闲59秒）且未配置withIdleness参数时，SourceReader不会在空闲周期向下游发送心跳式的水印更新，会触发下游算子的输入空闲判定逻辑误判，最终将持续有数据处理的左侧长链路流错误标记为空闲。

修复方案 #

1. 给Kafka Source的水印策略显式配置空闲检测参数，设置的超时时间略大于分区最大空闲时长即可，示例代码：

WatermarkStrategy.<YourEventType>forBoundedOutOfOrderness(Duration.ofSeconds(1))
    // 配置65秒空闲超时，覆盖59秒的最大分区空闲场景，避免误判
    .withIdleness(Duration.ofSeconds(65))

2. 额外校验项：检查左侧长链路算子的水印传播逻辑，确认链路中没有阻塞水印传递的算子（比如未配置允许延迟的窗口算子、自定义算子中未正确转发水印），可在Join的左侧输入前加临时监控，确认左侧流水印正常按节奏推进。

问题示意图

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