AWS托管Apache Flink checkpoint与应用版本不兼容及状态迁移咨询

问题场景 #

我有一个AWS托管的Apache Flink应用，包含2条输入流（A、B）和2条输出流（C、D）。每次对部分流做破坏性改动（比如修改A、C的数据格式及对应Java对象）后，重新部署时无法从最新checkpoint恢复，必须无checkpoint重启，导致未受影响的流（B、D）丢失关键历史状态。

报错信息 #

Caused by: java.util.concurrent.CompletionException: java.lang.IllegalStateException: Failed to rollback to checkpoint/savepoint s3://90adba2e0a9002371abdf5ee58da7883e350fe4d/a4aa7fa6aabe9853f35082abc0fd1320-338785721659-1663851015791/savepoints/115/attempt-1/savepoint-a4aa7f-7f08cc16a519. Cannot map checkpoint/savepoint state for operator fb8f6c016d8c769a70a72e4189372e48 to the new program, because the operator is not available in the new program.

需求：希望保留未受改动影响的流（如B->proc4->proc5->D）的状态，避免全量重启；想了解如何在AWS托管Flink中实现状态迁移，比如State Processor API的具体操作。

核心原因 #

报错本质是算子ID不匹配：Flink默认自动生成算子ID，代码变更（哪怕只改了A流的处理逻辑）会导致拓扑变化，旧checkpoint里的算子ID在新拓扑中找不到，直接触发全量恢复失败。其实B/D路径的算子状态是完好的，但被整体拓扑的ID映射失败牵连了。

分步解决方案 #

1. 给未改动算子固定ID #

要保住B/D路径的状态，第一步必须给这些算子手动指定固定UID，避免代码变更时ID自动变化：

// 对B流的proc4算子设置固定UID
streamB.keyBy(...)
       .process(new Proc4Function())
       .uid("proc4-fixed-id"); // 这个ID永远不变，哪怕其他代码改了

// proc5算子同理
.process(new Proc5Function())
.uid("proc5-fixed-id");

这样新拓扑里这些算子的ID和旧checkpoint里的完全一致，恢复时能精准匹配状态。

2. 针对改动部分做状态兼容或迁移 #

情况1：仅改数据格式/POJO类，算子逻辑没动

如果只是A流的POJO字段变了（比如新增字段、调整字段类型），不用动拓扑，直接做序列化兼容：

• 在Flink配置里加：

  state.backend.rocksdb.serializer-compatibility.enabled: true
  

  开启RocksDB的序列化兼容模式，允许新代码读取旧状态里的POJO。
• 同时在POJO类上处理兼容：比如给旧字段加@Deprecated，或者实现VersionedDeserializationSchema自定义反序列化逻辑，确保新代码能解析旧状态的数据。

情况2：算子逻辑/拓扑变了（比如新增/删除算子）

这种情况必须用State Processor API提取旧checkpoint里的有效状态，再生成新的可恢复savepoint。AWS托管Flink完全支持这个API，操作步骤如下：

1. 写一个状态导出作业：
   单独写个Flink作业，读取旧的checkpoint/savepoint，提取出proc4、proc5的状态，写入新的savepoint：

   ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
   // 加载旧savepoint（替换成你的S3路径）
   ExistingSavepoint savepoint = Savepoint.load(env, "s3://your-old-savepoint-path", new RocksDBStateBackend());
   // 提取proc4的KeyedState（替换成你的键类型和状态类型）
   DataSet<KeyValue<String, YourProc4State>> proc4State = savepoint.readKeyedState("proc4-fixed-id", String.class, YourProc4State.class);
   // 提取proc5的状态
   DataSet<KeyValue<String, YourProc5State>> proc5State = savepoint.readKeyedState("proc5-fixed-id", String.class, YourProc5State.class);
   // 生成只包含这两个算子状态的新savepoint
   Savepoint.create(env, new RocksDBStateBackend())
           .withKeyedState("proc4-fixed-id", proc4State, String.class, YourProc4State.class)
           .withKeyedState("proc5-fixed-id", proc5State, String.class, YourProc5State.class)
           .write("s3://new-savepoint-path");
   

2. 在AWS上运行导出作业：
   • 把这个作业打包成JAR，上传到S3。
   • 在AWS Flink控制台创建临时作业，指定这个JAR，给足S3读写权限（确保Flink角色能访问新旧savepoint路径）。
   • 运行作业生成新的savepoint，里面只有B/D路径的有效状态。
3. 用新savepoint启动主应用：
   部署修改后的主应用时，指定从新生成的savepoint恢复。此时B/D路径会加载旧状态继续处理，A/C路径状态为空，你可以配置A流从最新位置或指定偏移量重新消费。

3. AWS托管Flink的额外注意事项 #

• 权限要到位：确保Flink服务角色有S3读写权限，涵盖旧checkpoint、新savepoint、作业JAR的路径。
• 改代码前先存快照：每次做破坏性改动前，手动触发一次savepoint，确保旧状态完整：

  aws kinesisanalyticsv2 create-application-snapshot --application-name your-flink-app --snapshot-name pre-change-snapshot
  

• 记录拓扑和算子ID：每次部署前记录算子UID和拓扑结构，方便后续状态匹配排查问题。

针对你的示例场景的操作流程 #

1. 给proc4、proc5添加固定UID。
2. 用State Processor API导出旧savepoint中proc4、proc5的状态，生成新savepoint。
3. 部署修改后的应用，指定从新savepoint恢复；同时配置A流从最新位置重新消费，B流依赖savepoint的状态继续处理。

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