Checkpoint管理类：org.apache.flink.runtime.checkpoint.CheckpointCoordinator

从Savepoint恢复

  1.     /**
  2.      * Restore the state with given savepoint.
  3.      *
  4.      * @param savepointPointer The pointer to the savepoint.
  5.      * @param allowNonRestored True if allowing checkpoint state that cannot be mapped to any job
  6.      *     vertex in tasks.
  7.      * @param tasks Map of job vertices to restore. State for these vertices is restored via {@link
  8.      *     Execution#setInitialState(JobManagerTaskRestore)}.
  9.      * @param userClassLoader The class loader to resolve serialized classes in legacy savepoint
  10.      *     versions.
  11.      */
  12.     public boolean restoreSavepoint(
  13.             String savepointPointer,
  14.             boolean allowNonRestored,
  15.             Map<JobVertexID, ExecutionJobVertex> tasks,
  16.             ClassLoader userClassLoader)
  17.             throws Exception {
  19.         Preconditions.checkNotNull(savepointPointer, "The savepoint path cannot be null.");
  21.         LOG.info(
  22.                 "Starting job {} from savepoint {} ({})",
  23.                 job,
  24.                 savepointPointer,
  25.                 (allowNonRestored ? "allowing non restored state" : ""));
  27.         final CompletedCheckpointStorageLocation checkpointLocation =
  28.                 checkpointStorageView.resolveCheckpoint(savepointPointer);
  30.         // Load the savepoint as a checkpoint into the system
  31.         CompletedCheckpoint savepoint =
  32.                 Checkpoints.loadAndValidateCheckpoint(
  33.                         job, tasks, checkpointLocation, userClassLoader, allowNonRestored);
  35.         completedCheckpointStore.addCheckpoint(
  36.                 savepoint, checkpointsCleaner, this::scheduleTriggerRequest);
  38.         // Reset the checkpoint ID counter
  39.         long nextCheckpointId = savepoint.getCheckpointID() + 1;
  40.         checkpointIdCounter.setCount(nextCheckpointId);
  42.         LOG.info("Reset the checkpoint ID of job {} to {}.", job, nextCheckpointId);
  44.         final OptionalLong restoredCheckpointId =
  45.                 restoreLatestCheckpointedStateInternal(
  46.                         new HashSet<>(tasks.values()),
  47.                         OperatorCoordinatorRestoreBehavior.RESTORE_IF_CHECKPOINT_PRESENT,
  48.                         true,
  49.                         allowNonRestored,
  50.                         true);
  52.         return restoredCheckpointId.isPresent();
  53.     }