public static void restoreMasterHooks ( final Map < String , MasterTriggerRestoreHook < ? > > masterHooks , final Collection < MasterState > states , final long checkpointId , final boolean allowUnmatchedState , final Logger log ) throws FlinkException { // early out if ( states == null || states . isEmpty ( ) || masterHooks == null || masterHooks . isEmpty ( ) ) { log . info ( "No master state to restore" ) ; return ; } log . info ( "Calling master restore hooks" ) ; // collect the hooks final LinkedHashMap < String , MasterTriggerRestoreHook < ? > > allHooks = new LinkedHashMap <> ( masterHooks ) ; // first, deserialize all hook state final ArrayList < Tuple2 < MasterTriggerRestoreHook < ? > , Object > > hooksAndStates = new ArrayList <> ( ) ; for ( MasterState state : states ) { if ( state != null ) { final String name = state . name ( ) ; final MasterTriggerRestoreHook < ? > hook = allHooks . remove ( name ) ; if ( hook != null ) { log . debug ( "Found state to restore for hook '{}'" , name ) ; Object deserializedState = deserializeState ( state , hook ) ; hooksAndStates . add ( new Tuple2 <> ( hook , deserializedState ) ) ; } else if ( ! allowUnmatchedState ) { throw new IllegalStateException ( "Found state '" + state . name ( ) + "' which is not resumed by any hook." ) ; } else { log . info ( "Dropping unmatched state from '{}'" , name ) ; } } } // now that all is deserialized, call the hooks for ( Tuple2 < MasterTriggerRestoreHook < ? > , Object > hookAndState : hooksAndStates ) { restoreHook ( hookAndState . f1 , hookAndState . f0 , checkpointId ) ; } // trigger the remaining hooks without checkpointed state for ( MasterTriggerRestoreHook < ? > hook : allHooks . values ( ) ) { restoreHook ( null , hook , checkpointId ) ; } }
Calls the restore method given checkpoint master hooks and passes the given master state to them where state with a matching name is found .