private static void checkStateMappingCompleteness ( boolean allowNonRestoredState , Map < OperatorID , OperatorState > operatorStates , Map < JobVertexID , ExecutionJobVertex > tasks ) { Set < OperatorID > allOperatorIDs = new HashSet <> ( ) ; for ( ExecutionJobVertex executionJobVertex : tasks . values ( ) ) { allOperatorIDs . addAll ( executionJobVertex . getOperatorIDs ( ) ) ; } for ( Map . Entry < OperatorID , OperatorState > operatorGroupStateEntry : operatorStates . entrySet ( ) ) { OperatorState operatorState = operatorGroupStateEntry . getValue ( ) ; //----------------------------------------find operator for state--------------------------------------------- if ( ! allOperatorIDs . contains ( operatorGroupStateEntry . getKey ( ) ) ) { if ( allowNonRestoredState ) { LOG . info ( "Skipped checkpoint state for operator {}." , operatorState . getOperatorID ( ) ) ; } else { throw new IllegalStateException ( "There is no operator for the state " + operatorState . getOperatorID ( ) ) ; } } } }
Verifies that all operator states can be mapped to an execution job vertex .