Flink CDC 3.0 启动后咋们如何提交从savepoint恢复？

在 Flink CDC 3.0 中，提交从 savepoint 恢复的操作可以通过以下步骤完成：

1. 首先，确保你已经创建了一个 savepoint。你可以使用 savepoint() 方法来创建一个 savepoint，例如：

   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   // ... 其他代码 ...
   
   env.execute("Flink CDC Job");
   
   // 创建 savepoint
   Savepoint savepoint = env.getCheckpointConfig().getSavepointRetentionTimeout(TimeUnit.MINUTES);
   env.execute("Flink CDC Job with Savepoint");
   

2. 然后，你需要将 savepoint 文件保存到某个位置，以便稍后恢复。你可以使用 savepoint() 方法的重载版本来指定 savepoint 的文件路径和名称，例如：

   env.execute("Flink CDC Job");
   
   // 创建 savepoint
   Savepoint savepoint = env.getCheckpointConfig().getSavepointRetentionTimeout(TimeUnit.MINUTES);
   savepoint.setDirectory("path/to/savepoint/directory");
   savepoint.setFilenamePrefix("savepoint-");
   env.execute("Flink CDC Job with Savepoint");
   

3. 最后，当你需要从 savepoint 恢复时，可以使用 resumeFromSavepoint() 方法来启动 Flink CDC 作业并从指定的 savepoint 恢复，例如：

   StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
   // ... 其他代码 ...
   
   env.execute("Flink CDC Job");
   
   // 创建 savepoint
   Savepoint savepoint = env.getCheckpointConfig().getSavepointRetentionTimeout(TimeUnit.MINUTES);
   savepoint.setDirectory("path/to/savepoint/directory");
   savepoint.setFilenamePrefix("savepoint-");
   env.execute("Flink CDC Job with Savepoint");
   
   // 从 savepoint 恢复
   env.resumeFromSavepoint(savepoint);
   env.execute("Flink CDC Job Resumed from Savepoint");
   

通过以上步骤，你可以成功提交从 savepoint 恢复操作。请确保将 "path/to/savepoint/directory" 替换为实际的 savepoint 文件路径。

可以参考这个链接：https://github.com/ververica/flink-cdc-connectors/issues/2940，此回答整理自钉群“Flink CDC 社区”