Flink Kafka Connector指定时间戳无数据时偏移量重置失败的解决方案咨询

问题背景 #

需要从指定时间戳开始消费Kafka Topic数据，但不确定该时间戳是否存在对应数据。当无匹配数据时，连接器直接报错，期望此时能自动重置偏移量到最早可用位置，而非终止作业。

配置信息 #

当前使用的建表SQL：

CREATE TABLE TEST_TABLE (
     code_type_id STRING,
     create_user STRING,
     create_time STRING,
     create_name STRING
) WITH (
    'connector' = 'kafka',
    'topic' = 'TEST_TOPIC',
    'properties.bootstrap.servers' = '10.xx.xx.xx:9092',
    'properties.group.id' = 'test_group',
    'scan.startup.mode' = 'timestamp',
    'scan.startup.timestamp-millis' = '1734074100000',
    'properties.auto.offset.reset' = 'earliest',
    'format' = 'csv',
    'csv.ignore-parse-errors' = 'true',
    'csv.allow-comments' = 'true',
    'csv.field-delimiter' = '\u0001',
    'properties.security.protocol' = 'SASL_PLAINTEXT',
    'properties.sasl.mechanism' = 'SCRAM-SHA-256',
    'properties.sasl.jaas.config' = 'org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username="xxx" password="xxxxxxx";',
    'properties.key.deserializer' = 'org.apache.kafka.common.serialization.StringDeserializer',
    'properties.value.deserializer' = 'org.apache.kafka.common.serialization.StringDeserializer'
);

错误情况 #

指定时间戳无对应数据时，连接器触发偏移量查找失败的错误，配置的properties.auto.offset.reset = 'earliest'未生效，作业直接终止。

已尝试操作 #

• 设置scan.startup.mode为timestamp，指定scan.startup.timestamp-millis作为启动时间戳
• 配置properties.auto.offset.reset为earliest，期望无数据时自动 fallback 到最早偏移量

解决方案 #

核心原因 #

Flink 1.16版本中，properties.auto.offset.reset仅针对消费者无已提交偏移量的场景生效，不适用于timestamp启动模式下主动查找时间戳对应偏移量失败的情况。此时Flink会直接抛出错误，而非触发该配置的 fallback 逻辑。

可行解决方式 #

1. 自定义KafkaSource实现fallback逻辑（推荐）

放弃DDL方式，改用Java/Scala代码自定义KafkaSource，通过withFallbackOffset配置时间戳查找失败时的 fallback 策略：

KafkaSource<String> source = KafkaSource.<String>builder()
    .setBootstrapServers("10.xx.xx.xx:9092")
    .setTopics("TEST_TOPIC")
    .setGroupId("test_group")
    .setDeserializer(KafkaRecordDeserializationSchema.valueOnly(StringDeserializer.class))
    // 指定时间戳启动，找不到则 fallback 到最早偏移量
    .setStartingOffsets(OffsetsInitializer.timestamp(1734074100000L)
        .withFallbackOffset(OffsetsInitializer.earliest()))
    // SASL安全配置
    .setProperty("security.protocol", "SASL_PLAINTEXT")
    .setProperty("sasl.mechanism", "SCRAM-SHA-256")
    .setProperty("sasl.jaas.config", "org.apache.flink.kafka.shaded.org.apache.kafka.common.security.scram.ScramLoginModule required username=\"xxx\" password=\"xxxxxxx\";")
    .build();

// 将Source转换为DataStream后注册为Table使用
DataStream<String> stream = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source");
Table table = tableEnv.fromDataStream(stream);

2. 预处理校验时间戳有效性

在提交作业前，通过Kafka命令行工具校验指定时间戳是否在Topic的时间范围内：

# 查询Topic的最早消息时间戳
kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 10.xx.xx.xx:9092 --topic TEST_TOPIC --time -2

# 查询Topic的最晚消息时间戳
kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list 10.xx.xx.xx:9092 --topic TEST_TOPIC --time -1

如果指定时间戳晚于最晚时间戳，动态修改DDL的scan.startup.mode为earliest后再提交作业。

3. 升级Flink版本（可选）

Flink 1.17及以上版本新增了scan.startup.timestamp-millis.fallback-offset配置，可直接在DDL中指定时间戳查找失败时的 fallback 策略，无需自定义Source：

CREATE TABLE TEST_TABLE (
    -- 字段定义不变
) WITH (
    -- 原有配置不变
    'scan.startup.mode' = 'timestamp',
    'scan.startup.timestamp-millis' = '1734074100000',
    -- 新增fallback配置
    'scan.startup.timestamp-millis.fallback-offset' = 'earliest'
);

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