传递给env.from_source的Watermark策略未被使用，此现象是否正常？

环境配置 #

• flink 1.19.1
• python 3.10
• Apache flink 1.19.1 python包
• java 11
• Kafka flink connector 3.4.0

问题描述 #

我创建了一个KafkaSource，从包含消息头时间戳的Kafka主题读取数据。为其配置了使用AvroRowDeserializationSchema类创建的反序列化Schema，随后将该KafkaSource与带有自定义时间戳分配器的Watermark策略、源名称一起传入env.from_source API以创建数据流，但发现该源从未使用自定义时间戳分配器，而是直接将Kafka消息头中的时间戳附加到每个元素上。此现象是否符合预期？

有趣的是，当我在env.from_source返回的流上调用assign_watermarks_and_timestamps方法时，自定义时间戳分配器能够正常获取事件时间戳。

解答 #

这种现象是符合预期的，核心原因在于Flink Kafka Source的时间戳优先级设计：

1. KafkaSource原生时间戳的优先级
   Flink的KafkaSource默认会优先采用Kafka消息头自带的时间戳（即主题配置的CreateTime或LogAppendTime）。当你通过env.from_source传入Watermark策略时，KafkaSource会先检查是否能直接获取到Kafka原生时间戳——如果可以，就会自动使用这个时间戳，跳过自定义时间戳分配器的逻辑。

2. 两种时间戳配置方式的区别

• 绑定在env.from_source中的Watermark策略属于Source层面配置，KafkaSource会优先使用自身能获取的原生时间戳，仅当Kafka消息无时间戳或你显式配置忽略原生时间戳时，才会触发自定义分配器。
• 在数据流上调用assign_watermarks_and_timestamps属于数据流层面配置，会直接覆盖Source的时间戳逻辑，强制使用自定义分配器提取事件时间，因此能正常生效。

若要让自定义时间戳分配器在Source层面生效，需要显式禁用KafkaSource对原生时间戳的使用：可以在构建KafkaSource时通过set_timestamp_extractor指定自定义提取器，或者在配置Watermark策略时明确忽略Kafka的原生时间戳。

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