Flink Checkpoint原理解析

	public PendingCheckpoint triggerCheckpoint(
			long timestamp,
			CheckpointProperties props,
			@Nullable String externalSavepointLocation,
			boolean isPeriodic,
			boolean advanceToEndOfTime) throws CheckpointException {

		if (advanceToEndOfTime && !(props.isSynchronous() && props.isSavepoint())) {
			throw new IllegalArgumentException("Only synchronous savepoints are allowed to advance the watermark to MAX.");
		}

		// 做一些预先的检查
		synchronized (lock) {
			// 若此时coordinator已经被关闭则checkpoint失败
			if (shutdown) {
				throw new CheckpointException(CheckpointFailureReason.CHECKPOINT_COORDINATOR_SHUTDOWN);
			}

			// 若是周期性执行而调度器已经关闭则checkpoint失败
			if (isPeriodic && !periodicScheduling) {
				throw new CheckpointException(CheckpointFailureReason.PERIODIC_SCHEDULER_SHUTDOWN);
			}

			// 验证checkpoint是否达到大并发数，验证checkpoint之间的小时间间隔
			// 默认大并发度为1，小间隔为0，这些参数都可以提前自定义配置
			// 这些检查都和savepoint没有关系
			if (!props.forceCheckpoint()) {
				// 检查是否只有一个触发checkpoint的请求
				if (triggerRequestQueued) {
					LOG.warn("Trying to trigger another checkpoint for job {} while one was queued already.", job);
					throw new CheckpointException(CheckpointFailureReason.ALREADY_QUEUED);
				}

				// 检查当前有多少checkpoint在pending
				checkConcurrentCheckpoints();
				// 检查小时间间隔
				checkMinPauseBetweenCheckpoints();
			}
		}

		// 检查所有子任务的状态，若任务不在运行则直接抛弃这次checkpoint
		Execution[] executions = new Execution[tasksToTrigger.length];
		for (int i = ; i < tasksToTrigger.length; i++) {
			Execution ee = tasksToTrigger[i].getCurrentExecutionAttempt();
			if (ee == null) {
				LOG.info("Checkpoint triggering task {} of job {} is not being executed at the moment. Aborting checkpoint.",
						tasksToTrigger[i].getTaskNameWithSubtaskIndex(),
						job);
				throw new CheckpointException(CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING);
			} else if (ee.getState() == ExecutionState.RUNNING) {
				executions[i] = ee;
			} else {
				LOG.info("Checkpoint triggering task {} of job {} is not in state {} but {} instead. Aborting checkpoint.",
						tasksToTrigger[i].getTaskNameWithSubtaskIndex(),
						job,
						ExecutionState.RUNNING,
						ee.getState());
				throw new CheckpointException(CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING);
			}
		}


		// 检查所有待确认算子的运行状态，若不在运行中则直接抛弃这次checkpoint
		Map<ExecutionAttemptID, ExecutionVertex> ackTasks = new HashMap<>(tasksToWaitFor.length);

		for (ExecutionVertex ev : tasksToWaitFor) {
			Execution ee = ev.getCurrentExecutionAttempt();
			if (ee != null) {
				ackTasks.put(ee.getAttemptId(), ev);
			} else {
				LOG.info("Checkpoint acknowledging task {} of job {} is not being executed at the moment. Aborting checkpoint.",
						ev.getTaskNameWithSubtaskIndex(),
						job);
				throw new CheckpointException(CheckpointFailureReason.NOT_ALL_REQUIRED_TASKS_RUNNING);
			}
		}


		// 做完上述的一系列检查后开始真正触发checkpoint
		// 通过加特殊的锁来确保触发请求不会并发
		synchronized (triggerLock) {

			// checkpoint保存路径
			final CheckpointStorageLocation checkpointStorageLocation;
			final long checkpointID;

			try {
				// 这个必须使用全局的锁，因为与外部的系统交互可能会导致阻塞一段时间
				// checkpointID通常是一个递增的时间戳
				checkpointID = checkpointIdCounter.getAndIncrement();

				checkpointStorageLocation = props.isSavepoint() ?
						checkpointStorage.initializeLocationForSavepoint(checkpointID, externalSavepointLocation) :
						checkpointStorage.initializeLocationForCheckpoint(checkpointID);
			}
			catch (Throwable t) {
				// 若失败则记录失败次数，这些指标都可以通过metrics收集起来，用于checkpoint监控
				int numUnsuccessful = numUnsuccessfulCheckpointsTriggers.incrementAndGet();
				LOG.warn("Failed to trigger checkpoint for job {} ({} consecutive failed attempts so far).",
						job,
						numUnsuccessful,
						t);
				throw new CheckpointException(CheckpointFailureReason.EXCEPTION, t);
			}

			// 创建一个等待执行的Checkpoint
			final PendingCheckpoint checkpoint = new PendingCheckpoint(
				job,
				checkpointID,
				timestamp,
				ackTasks,
				props,
				checkpointStorageLocation,
				executor);

			if (statsTracker != null) {
				PendingCheckpointStats callback = statsTracker.reportPendingCheckpoint(
					checkpointID,
					timestamp,
					props);

				checkpoint.setStatsCallback(callback);
			}

			// 调度器会清理掉过期的checkpoint
			final Runnable canceller = () -> {
				synchronized (lock) {
					// 若checkpoint已经过期则直接丢弃
					if (!checkpoint.isDiscarded()) {
						LOG.info("Checkpoint {} of job {} expired before completing.", checkpointID, job);

						failPendingCheckpoint(checkpoint, CheckpointFailureReason.CHECKPOINT_EXPIRED);
						pendingCheckpoints.remove(checkpointID);
						rememberRecentCheckpointId(checkpointID);

						triggerQueuedRequests();
					}
				}
			};

			try {
				// 重新获取coordinator锁
				synchronized (lock) {
					// 因为之前做完预检查后释放了锁，所以需要再检查下上述条件
					if (shutdown) {
						throw new CheckpointException(CheckpointFailureReason.CHECKPOINT_COORDINATOR_SHUTDOWN);
					}
					else if (!props.forceCheckpoint()) {
						if (triggerRequestQueued) {
							LOG.warn("Trying to trigger another checkpoint for job {} while one was queued already.", job);
							throw new CheckpointException(CheckpointFailureReason.ALREADY_QUEUED);
						}

						checkConcurrentCheckpoints();

						checkMinPauseBetweenCheckpoints();
					}

					LOG.info("Triggering checkpoint {} @ {} for job {}.", checkpointID, timestamp, job);

					pendingCheckpoints.put(checkpointID, checkpoint);

					// 创建超时取消线程
					ScheduledFuture<?> cancellerHandle = timer.schedule(
							canceller,
							checkpointTimeout, TimeUnit.MILLISECONDS);

					if (!checkpoint.setCancellerHandle(cancellerHandle)) {
						// 若checkpoint超时则会被取消
						cancellerHandle.cancel(false);
					}

					// 提前创建钩子等待所有子任务触发checkpoint
					final List<MasterState> masterStates = MasterHooks.triggerMasterHooks(masterHooks.values(),
							checkpointID, timestamp, executor, Time.milliseconds(checkpointTimeout));
					for (MasterState s : masterStates) {
						checkpoint.addMasterState(s);
					}
				}
				// end of lock scope

				final CheckpointOptions checkpointOptions = new CheckpointOptions(
						props.getCheckpointType(),
						checkpointStorageLocation.getLocationReference());

				// 向所有子任务发送消息触发checkpoint
				for (Execution execution: executions) {
					if (props.isSynchronous()) {
						execution.triggerSynchronousSavepoint(checkpointID, timestamp, checkpointOptions, advanceToEndOfTime);
					} else {
						// 触发checkpoint,从source算子广播CheckBarrier，当所有算子都完成Checkpoint后，jobManager认为Checkpoint已完成
						execution.triggerCheckpoint(checkpointID, timestamp, checkpointOptions);
					}
				}
				// 将失败的触发器数量归0
				numUnsuccessfulCheckpointsTriggers.set();
				return checkpoint;
			}
			catch (Throwable t) {
				// checkpoint出现异常则则从pending缓存中去除
				synchronized (lock) {
					pendingCheckpoints.remove(checkpointID);
				}
				// 记录checkpoint失败次数
				int numUnsuccessful = numUnsuccessfulCheckpointsTriggers.incrementAndGet();
				LOG.warn("Failed to trigger checkpoint {} for job {}. ({} consecutive failed attempts so far)",
						checkpointID, job, numUnsuccessful, t);

				if (!checkpoint.isDiscarded()) {
					failPendingCheckpoint(checkpoint, CheckpointFailureReason.TRIGGER_CHECKPOINT_FAILURE, t);
				}

				try {
					// 失败后清除这次checkpoint的保存路径和该路径下的所有数据
					checkpointStorageLocation.disposeOnFailure();
				}
				catch (Throwable t2) {
					LOG.warn("Cannot dispose failed checkpoint storage location {}", checkpointStorageLocation, t2);
				}

				throw new CheckpointException(CheckpointFailureReason.EXCEPTION, t);
			}

		} // end trigger lock
	}