Flink 1.10.2迁移至1.14.5：Kafka Sink时间戳问题咨询

差异原因拆解 #

1. DataStream API 自动携带Event Time的逻辑 #

Flink 1.14.x 对Kafka Sink的时间戳处理做了底层重构，废弃了旧版本的setWriteTimestampToKafka配置，同时调整了默认行为：

• 你通过env.fromSource构建的数据流，每个元素已经通过WatermarkStrategy绑定了数据源的Event Time；
• 新版KafkaRecordSerializationSchema默认会自动提取元素的Event Time，作为Kafka记录的timestamp写入Broker，不需要额外配置。这就是你看到DataStream输出的Kafka记录时间戳与数据源Event Time一致的原因。

对比1.10.2旧版本：当时必须显式调用setWriteTimestampToKafka(true)，才会将Event Time写入Kafka，默认行为是使用生产者发送时的当前时间，这是API行为的直接变更。

2. Table API 默认使用当前时间的原因 #

Table API的Kafka Sink逻辑与DataStream分离，默认不会自动传递Event Time到Kafka记录的timestamp字段：

• 若要让Table API输出的Kafka记录携带Event Time，必须在DDL中显式配置sink.timestamp.field参数，指定表中对应的Event Time字段。示例DDL：

CREATE TABLE sink_table (
  data STRING,
  event_time TIMESTAMP(3) -- 表中存储的Event Time字段
) WITH (
  'connector' = 'kafka',
  'topic' = 'sink_topic',
  'properties.bootstrap.servers' = 'bootstrap.servers',
  'sink.timestamp.field' = 'event_time' -- 关键配置：映射到Kafka记录的timestamp
);

如果缺少这个配置，Table API的Kafka Sink就会默认用发送时的当前时间填充Kafka记录的timestamp，这就是你观察到的差异点。

额外配置说明 #

如果需要在DataStream API中修改默认行为（比如改用当前时间作为Kafka记录的timestamp），可以通过自定义KafkaRecordSerializationSchema实现：

KafkaRecordSerializationSchema.builder()
    .setTopicSelector(topicSelector)
    .setValueSerializationSchema(schema)
    .setTimestampExtractor((element, context) -> System.currentTimeMillis()) // 手动指定为当前时间
    .build()

旧版本的setWriteTimestampToKafka方法在1.14.x中已被移除，取而代之的是更灵活的setTimestampExtractor，支持自定义时间戳提取逻辑。

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