Flink源码——Job 提交、部署流程源码分析之Task状态恢复

\toperator.open();

operator.initializeState(streamTaskStateInitializer);

获取到 StreamOperator

 timeServiceManager = context.internalTimerServiceManager(); 

获取 StreamOperatorStateContext， 会执行各种 state 的恢复

stateHandler.initializeOperatorState(this);

operatorChain.initializeStateAndOpenOperators(createStreamTaskStateInitializer());

创建 StreamOperatorStateHandler 实例对象

接：Flink源码——Job 提交、部署流程源码分析之Task 部署、提交、執行（一）

StreamOperatorStateHandler 初始化

状态恢复入口

执行用户定义的算子open方法

readRecoveredChannelState();

初始化得到一个 StreamTaskStateInitializerImpl 组件

createStreamTaskStateInitializer()

runtimeContext.setKeyedStateStore(stateHandler.getKeyedStateStore().orElse(null));

遍历每个 StreamOperator

actionExecutor.runThrowing(() -> { *  注释： 状态恢复入口operatorChain.initializeStateAndOpenOperators(createStreamTaskStateInitializer()); *  注释： 初始化 Mail    *  这个地方主要是初始化 InputGate 等输入相关的细节readRecoveredChannelState();}

初始化 Mail，这个地方主要是初始化 InputGate 等输入相关的细节

初始化状态这当中，包含 state 恢复

 operator.initializeState(streamTaskStateInitializer); 

Flink源码（1.11.x）Flink源码——Job 提交、部署流程源码分析之Task状态恢复