flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-runtime/src/main/java/org/apache/flink/runtime/checkpoint/CheckpointCoordinator.java
/**
	 * Restores the latest checkpointed state.
	 *
	 * @param tasks Map of job vertices to restore. State for these vertices is
	 * restored via {@link Execution#setInitialState(TaskStateSnapshot)}.
	 * @param errorIfNoCheckpoint Fail if no completed checkpoint is available to
	 * restore from.
	 * @param allowNonRestoredState Allow checkpoint state that cannot be mapped
	 * to any job vertex in tasks.
	 * @return <code>true</code> if state was restored, <code>false</code> otherwise.
	 * @throws IllegalStateException If the CheckpointCoordinator is shut down.
	 * @throws IllegalStateException If no completed checkpoint is available and
	 *                               the <code>failIfNoCheckpoint</code> flag has been set.
	 * @throws IllegalStateException If the checkpoint contains state that cannot be
	 *                               mapped to any job vertex in <code>tasks</code> and the
	 *                               <code>allowNonRestoredState</code> flag has not been set.
	 * @throws IllegalStateException If the max parallelism changed for an operator
	 *                               that restores state from this checkpoint.
	 * @throws IllegalStateException If the parallelism changed for an operator
	 *                               that restores <i>non-partitioned</i> state from this
	 *                               checkpoint.
	 */
Restores the latest checkpointed state.
@param tasks Map of job vertices to restore. State for these vertices is
restored via  Execution#setInitialState(TaskStateSnapshot).
@param errorIfNoCheckpoint Fail if no completed checkpoint is available to
restore from.
@param allowNonRestoredState Allow checkpoint state that cannot be mapped
to any job vertex in tasks.
@return true if state was restored, false otherwise.
@throws IllegalStateException If the CheckpointCoordinator is shut down.
@throws IllegalStateException If no completed checkpoint is available and
the failIfNoCheckpoint flag has been set.
@throws IllegalStateException If the checkpoint contains state that cannot be
mapped to any job vertex in tasks and the
allowNonRestoredState flag has not been set.
@throws IllegalStateException If the max parallelism changed for an operator
that restores state from this checkpoint.
@throws IllegalStateException If the parallelism changed for an operator
that restores non-partitioned state from this
checkpoint.
Restores the latest checkpointed state . @param tasks Map of job vertices to restore . State for these vertices is restored via Execution#setInitialState ( TaskStateSnapshot ) . @param errorIfNoCheckpoint Fail if no completed checkpoint is available to restore from . @param allowNonRestoredState Allow checkpoint state that cannot be mapped to any job vertex in tasks . @return true if state was restored false otherwise . @throws IllegalStateException If the CheckpointCoordinator is shut down . @throws IllegalStateException If no completed checkpoint is available and the failIfNoCheckpoint flag has been set . @throws IllegalStateException If the checkpoint contains state that cannot be mapped to any job vertex in tasks and the allowNonRestoredState flag has not been set . @throws IllegalStateException If the max parallelism changed for an operator that restores state from this checkpoint . @throws IllegalStateException If the parallelism changed for an operator that restores non - partitioned state from this checkpoint .
public boolean restoreLatestCheckpointedState(
			Map<JobVertexID, ExecutionJobVertex> tasks,
			boolean errorIfNoCheckpoint,
			boolean allowNonRestoredState) throws Exception {
		synchronized (lock) {
			if (shutdown) {
				throw new IllegalStateException("CheckpointCoordinator is shut down");
			}
			sharedStateRegistry.close();
			sharedStateRegistry = sharedStateRegistryFactory.create(executor);
			completedCheckpointStore.recover();
			for (CompletedCheckpoint completedCheckpoint : completedCheckpointStore.getAllCheckpoints()) {
				completedCheckpoint.registerSharedStatesAfterRestored(sharedStateRegistry);
			}
			LOG.debug("Status of the shared state registry after restore: {}.", sharedStateRegistry);
			CompletedCheckpoint latest = completedCheckpointStore.getLatestCheckpoint();
			if (latest == null) {
				if (errorIfNoCheckpoint) {
					throw new IllegalStateException("No completed checkpoint available");
				} else {
					return false;
				}
			}
			LOG.info("Restoring from latest valid checkpoint: {}.", latest);
			final Map<OperatorID, OperatorState> operatorStates = latest.getOperatorStates();
			StateAssignmentOperation stateAssignmentOperation =
					new StateAssignmentOperation(tasks, operatorStates, allowNonRestoredState);
			stateAssignmentOperation.assignStates();
			MasterHooks.restoreMasterHooks(
					masterHooks,
					latest.getMasterHookStates(),
					latest.getCheckpointID(),
					allowNonRestoredState,
					LOG);
			if (statsTracker != null) {
				long restoreTimestamp = System.currentTimeMillis();
				RestoredCheckpointStats restored = new RestoredCheckpointStats(
					latest.getCheckpointID(),
					latest.getProperties(),
					restoreTimestamp,
					latest.getExternalPointer());
				statsTracker.reportRestoredCheckpoint(restored);
			}
			return true;
		}
	}
public boolean restoreLatestCheckpointedState ( Map < JobVertexID , ExecutionJobVertex > tasks , boolean errorIfNoCheckpoint , boolean allowNonRestoredState ) throws Exception { synchronized ( lock ) { if ( shutdown ) { throw new IllegalStateException ( "CheckpointCoordinator is shut down" ) ; } sharedStateRegistry . close ( ) ; sharedStateRegistry = sharedStateRegistryFactory . create ( executor ) ; completedCheckpointStore . recover ( ) ; for ( CompletedCheckpoint completedCheckpoint : completedCheckpointStore . getAllCheckpoints ( ) ) { completedCheckpoint . registerSharedStatesAfterRestored ( sharedStateRegistry ) ; } LOG . debug ( "Status of the shared state registry after restore: {}." , sharedStateRegistry ) ; CompletedCheckpoint latest = completedCheckpointStore . getLatestCheckpoint ( ) ; if ( latest == null ) { if ( errorIfNoCheckpoint ) { throw new IllegalStateException ( "No completed checkpoint available" ) ; } else { return false ; } } LOG . info ( "Restoring from latest valid checkpoint: {}." , latest ) ; final Map < OperatorID , OperatorState > operatorStates = latest . getOperatorStates ( ) ; StateAssignmentOperation stateAssignmentOperation = new StateAssignmentOperation ( tasks , operatorStates , allowNonRestoredState ) ; stateAssignmentOperation . assignStates ( ) ; MasterHooks . restoreMasterHooks ( masterHooks , latest . getMasterHookStates ( ) , latest . getCheckpointID ( ) , allowNonRestoredState , LOG ) ; if ( statsTracker != null ) { long restoreTimestamp = System . currentTimeMillis ( ) ; RestoredCheckpointStats restored = new RestoredCheckpointStats ( latest . getCheckpointID ( ) , latest . getProperties ( ) , restoreTimestamp , latest . getExternalPointer ( ) ) ; statsTracker . reportRestoredCheckpoint ( restored ) ; } return true ; } }
Restores the latest checkpointed state.
Restores the latest checkpointed state .
// We create a new shared state registry object, so that all pending async disposal requests from previous // runs will go against the old object (were they can do no harm). // This must happen under the checkpoint lock. // Recover the checkpoints, TODO this could be done only when there is a new leader, not on each recovery // Now, we re-register all (shared) states from the checkpoint store with the new registry // Restore from the latest checkpoint // re-assign the task states // call master hooks for restore // update metrics