isRestored 方法返回值依赖 restoredCheckpointId 是否为空：

restoredCheckpointId 在算子状态句柄（StreamOperatorStateHandler）中从 StreamOperatorStateContext 获取并赋值给 StateInitializationContext（该 context 就是 initializeState 方法中使用的 context）：

StreamOperatorStateContext 在 StreamOperatorStateHandler 初始化时传入：

StreamOperatorStateContext的 checkpointId 来自 TaskStateMangerImpl：

TaskStateMangerImpl 实现中会从 jobMangerTaskRestore 中获取 checkpointId：

jobMangerTaskRestore 实例化的唯一方式是从 TaskStateMangerImpl 构造函数中构造：

submitTask 是 JM 通过 Akka 向 TM 提交 Task 执行的方法。JM 会从状态文件中获取 source 算子的状态交由 TM 执行恢复动作，checkpointId默认赋值。

任务从状态启动时(allowNonRestoredState=true)，每个算子的 subTask 都具有状态：这里的状态指的是表达算子状态的 OperatorSubtaskStates 实例不会为空，只是说真正有状态算子的大小不为0，本身没状态算子的大小为0。

以 task 纬度分配状态，当开启算子链合并时，每个 task 里面包含多个算子：jobVertex 为 job 执行图的顶点：多个算子组成。

里面有两次 for 循环，外层处理单个 task，内层处理 task 再分别处理每个算子（封装为 OperatorState）。