使用Table API与Kafka连接器的Flink作业无法从Savepoint恢复

问题描述 #

• 通过Savepoint停止Flink作业后，使用同一jar包从该Savepoint恢复，报错无法映射Savepoint状态，提示旧算子ID在新程序中不存在
• 疑问：未修改代码为何算子ID发生变化？Table API结合Kafka连接器的作业是否支持Savepoint恢复？

错误信息 #

used by: java.util.concurrent.CompletionException: java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint file:/root/flink-savepoints/savepoint-5f285c-c2749410db07. Cannot map checkpoint/savepoint state for operator dd5fc1f28f42d777f818e2e8ea18c331 to the new program, because the operator is not available in the new program. If you want to allow to skip this, you can set the --allowNonRestoredState option on the CLI.

used by: java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint file:/root/flink-savepoints/savepoint-5f285c-c2749410db07. Cannot map checkpoint/savepoint state for operator dd5fc1f28f42d777f818e2e8ea18c331 to the new program, because the operator is not available in the new program. If you want to allow to skip this, you can set the --allowNonRestoredState option on the CLI.

作业代码 #

public final class FlinkJob {

    public static void main(String[] args) {

        final String JOB_NAME = "FlinkJob";

        final EnvironmentSettings settings = EnvironmentSettings.inStreamingMode();
        final TableEnvironment tEnv = TableEnvironment.create(settings);
        tEnv.getConfig().set("pipeline.name", JOB_NAME);
        tEnv.getConfig().setLocalTimeZone(ZoneId.of("UTC"));

        tEnv.executeSql("CREATE TEMPORARY TABLE ApiLog (" +
                "  `_timestamp` TIMESTAMP(3) METADATA FROM 'timestamp' VIRTUAL," +
                "  `_partition` INT METADATA FROM 'partition' VIRTUAL," +
                "  `_offset` BIGINT METADATA FROM 'offset' VIRTUAL," +
                "  `Data` STRING," +
                "  `Action` STRING," +
                "  `ProduceDateTime` TIMESTAMP_LTZ(6)," +
                "  `OffSet` INT" +
                ") WITH (" +
                "  'connector' = 'kafka'," +
                "  'topic' = 'api.log'," +
                "  'properties.group.id' = 'flink'," +
                "  'properties.bootstrap.servers' = '<mykafkahost...>'," +
                "  'format' = 'json'," +
                "  'json.timestamp-format.standard' = 'ISO-8601'" +
                ")");

        tEnv.executeSql("CREATE TABLE print_table (" +
                " `_timestamp` TIMESTAMP(3)," +
                " `_partition` INT," +
                " `_offset` BIGINT," +
                " `Data` STRING," +
                " `Action` STRING," +
                " `ProduceDateTime` TIMESTAMP(6)," +
                " `OffSet` INT" +
                ") WITH ('connector' = 'print')");

        tEnv.executeSql("INSERT INTO print_table" +
                " SELECT * FROM ApiLog");

    }

}

问题解答 #

1. 未修改代码但算子ID变化的原因 #

Table API是声明式API，Flink会自动生成执行计划和算子ID，即使代码无改动，以下场景仍可能导致算子ID变化：

• Kafka元数据依赖：你的Kafka表定义了_timestamp、_partition等VIRTUAL元数据字段，Flink生成算子时会依赖Kafka集群的实时元数据（如分区数、topic配置），若集群元数据发生变化（哪怕代码未改），会导致算子结构改变，进而算子ID变化。
• 执行计划隐式优化：Flink优化器可能在不同运行环境/启动时机生成不同的执行计划（如算子合并、顺序调整），这会直接导致算子ID变更。
• 临时表的不确定性：使用CREATE TEMPORARY TABLE时，临时表的元数据在作业重启时会重新解析，细微的解析差异也可能影响算子ID的生成。

2. Table API + Kafka连接器是否支持Savepoint恢复？ #

支持，但需满足以下前提：

• 作业的表结构、SQL逻辑完全一致，包括字段顺序、类型、元数据定义。
• Kafka连接器的配置完全一致（如group.id、topic、bootstrap.servers），尤其是group.id，Kafka消费位移与该配置绑定。
• 启用稳定的算子ID生成策略，避免隐式变化导致的ID不一致。

3. 解决当前问题的具体步骤 #

• 检查Kafka集群状态：确认api.log主题的分区数、配置未发生变化，无新增/删除分区操作。
• 启用稳定算子ID生成：在代码中添加配置，让Table API基于SQL逻辑和表结构生成稳定的哈希算子UID：

  tEnv.getConfig().set("table.exec.uid.generator", "HASH");
  

• 替换临时表为持久表：若业务允许，将CREATE TEMPORARY TABLE改为CREATE TABLE，避免临时表元数据解析的不确定性。
• 谨慎使用强制恢复：若确认缺失的算子状态不影响业务连续性，可添加--allowNonRestoredState参数启动作业，跳过无法映射的状态，但需注意这可能导致数据重复或丢失。
• 重新生成Savepoint：若上述方法无效，建议停止当前作业并生成新的Savepoint，再用添加了稳定UID配置的代码重启作业。

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