代码内配置Savepoint重启Flink作业无效，如何解决？

直接在Configuration中设置execution.savepoint.path无法生效，因为该参数属于Flink CLI客户端配置，而非作业运行时配置。要通过代码实现从指定Savepoint或Checkpoint恢复作业，需使用Flink提供的SavepointRestoreSettings API显式配置恢复策略，具体步骤如下：

核心实现代码 #

1. 基于StreamExecutionEnvironment配置恢复 #

如果你的SQL作业基于流处理模式，先配置StreamExecutionEnvironment的恢复参数，再创建TableEnvironment：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.environment.SavepointRestoreSettings;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

public class Deployer {
    public static void main(String[] args) throws Exception {
        // 初始化StreamExecutionEnvironment
        StreamExecutionEnvironment streamEnv = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 配置已有Checkpoint参数（保留你原有的代码）
        streamEnv.enableCheckpointing(5000);
        streamEnv.getCheckpointConfig().setCheckpointStorage("file:///tmp/flink-checkpoints");
        
        // 设置从指定Checkpoint/Savepoint恢复
        String restorePath = "file:///tmp/flink-checkpoint-directory-domain/a98c68e3139041bc32e6a931e1f701e1/chk-24/_metadata";
        SavepointRestoreSettings restoreSettings = SavepointRestoreSettings.forPath(restorePath);
        streamEnv.setSavepointRestoreSettings(restoreSettings);
        
        // 基于配置好的StreamEnv创建TableEnvironment
        EnvironmentSettings tableEnvSettings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        TableEnvironment tableEnv = TableEnvironment.create(tableEnvSettings);
        
        // 加载并执行SQL文件（保留你原有的代码逻辑）
        // 示例：读取SQL文件内容执行
        String sqlContent = "..."; // 从文件读取SQL的逻辑
        tableEnv.executeSql(sqlContent);
        
        streamEnv.execute("SQL Job from Savepoint");
    }
}

2. 关键说明 #

• SavepointRestoreSettings.forPath()方法既支持Savepoint路径，也支持已完成的Checkpoint路径（Flink允许将有效Checkpoint当作Savepoint恢复）。
• 必须确保指定的恢复路径在Flink集群所有节点可访问：
  • 若使用本地文件路径，需保证所有TaskManager节点能读取该路径（分布式场景建议使用HDFS、S3等共享存储）。
  • 路径格式需符合Flink要求（如file://、hdfs://前缀）。

常见注意事项 #

• 作业拓扑必须与生成该Checkpoint/Savepoint的作业完全一致：算子数量、算子ID、状态定义等不能变更，否则恢复会失败。
• 若恢复时遇到权限问题，需检查Flink集群进程对存储路径的读写权限。

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