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）。