配置Flink使用ProcessingTime时，context.timestamp()为何返回Kafka摄入时间？

原因分析 #

你遇到的情况是正常的，核心逻辑如下：

• ctx.timestamp() 返回的是元素的事件时间戳，和ProcessingTime（处理时间）是两个独立概念。
• 新版Flink的KafkaSource（你使用的1.16.1版本属于新Source API）默认会自动提取Kafka消息的CreateTime（消息写入Kafka的时间）作为元素的事件时间戳，这个行为不受TimeCharacteristic.ProcessingTime配置的影响。
• TimeCharacteristic.ProcessingTime仅用于指定Flink窗口、定时器等算子默认使用的时间语义，不会清除元素本身携带的事件时间戳。

解决方法 #

如果希望ctx.timestamp()返回null，需要显式配置KafkaSource不要提取Kafka内置的时间戳，具体有两种实现方式：

方式1：自定义时间戳分配器返回null #

修改KafkaSource构建代码，添加set_timestamp_assigner配置，指定不设置事件时间戳：

kafka_source = KafkaSource.builder()\
        .set_properties(properties)\
        .set_topics(topic)\
        .set_starting_offsets(KafkaOffsetsInitializer.earliest())\
        .set_value_only_deserializer(SimpleStringSchema())\
        .set_timestamp_assigner(lambda context, record: None)  # 关键配置：不生成事件时间戳
        .build()

方式2：通过消费者属性禁用自动时间戳提取 #

在Kafka配置中添加属性，禁止提取内置时间戳：

properties = get_kafka_properties(args)
# 添加配置禁用Kafka消息时间戳提取
properties["kafka.timestamp.extractor"] = "org.apache.flink.kafka.shaded.org.apache.kafka.common.record.TimestampType.NO_TIMESTAMP_TYPE"

kafka_source = KafkaSource.builder()\
        .set_properties(properties)\
        .set_topics(topic)\
        .set_starting_offsets(KafkaOffsetsInitializer.earliest())\
        .set_value_only_deserializer(SimpleStringSchema())\
        .build()

补充说明 #

• TimeCharacteristic.ProcessingTime在Flink 1.12之后已被标记为过时，推荐直接通过WatermarkStrategy指定时间语义。如果要完全基于ProcessingTime开发，确保WatermarkStrategy使用no_watermarks()，同时结合上述时间戳配置即可。
• 若要获取ProcessingTime（Flink处理当前元素的时间），可以调用ctx.timer_service().current_processing_time()。

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