Flink CheckpointCoordinator 步骤流程源码

谁借莪１个温暖的怀抱￠ 2021-09-25 03:24 1265阅读 0赞

1 内部是用ScheduledThreadPoolExecutor实现的

CheckpointCoordinator.java

/**
    	 * Triggers a new standard checkpoint and uses the given timestamp as the checkpoint
    	 * timestamp.
    	 *
    	 * @param timestamp The timestamp for the checkpoint.
    	 * @param isPeriodic Flag indicating whether this triggered checkpoint is
    	 * periodic. If this flag is true, but the periodic scheduler is disabled,
    	 * the checkpoint will be declined.
    	 * @return <code>true</code> if triggering the checkpoint succeeded.
    	 */
    	public boolean triggerCheckpoint(long timestamp, boolean isPeriodic) {
    		try {
    			triggerCheckpoint(timestamp, checkpointProperties, null, isPeriodic, false);
    			return true;
    		} catch (CheckpointException e) {
    			long latestGeneratedCheckpointId = getCheckpointIdCounter().get();
    			// here we can not get the failed pending checkpoint's id,
    			// so we pass the negative latest generated checkpoint id as a special flag
    			failureManager.handleJobLevelCheckpointException(e, -1 * latestGeneratedCheckpointId);
    			return false;
    		}
    	}
    
    	@VisibleForTesting
    	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.");
    		}
    
    		// make some eager pre-checks
    		synchronized (lock) {
    			preCheckBeforeTriggeringCheckpoint(isPeriodic, props.forceCheckpoint());
    		}
    
    		// check if all tasks that we need to trigger are running.
    		// if not, abort the checkpoint
    		Execution[] executions = new Execution[tasksToTrigger.length];
    		for (int i = 0; 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);
    			}
    		}
    
    		// next, check if all tasks that need to acknowledge the checkpoint are running.
    		// if not, abort the 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);
    			}
    		}
    
    		// we will actually trigger this checkpoint!
    
    		// we lock with a special lock to make sure that trigger requests do not overtake each other.
    		// this is not done with the coordinator-wide lock, because the 'checkpointIdCounter'
    		// may issue blocking operations. Using a different lock than the coordinator-wide lock,
    		// we avoid blocking the processing of 'acknowledge/decline' messages during that time.
    		synchronized (triggerLock) {
    
    			final CheckpointStorageLocation checkpointStorageLocation;
    			final long checkpointID;
    
    			try {
    				// this must happen outside the coordinator-wide lock, because it communicates
    				// with external services (in HA mode) and may block for a while.
    				checkpointID = checkpointIdCounter.getAndIncrement();
    
    				checkpointStorageLocation = props.isSavepoint() ?
    						checkpointStorage.initializeLocationForSavepoint(checkpointID, externalSavepointLocation) :
    						checkpointStorage.initializeLocationForCheckpoint(checkpointID);
    			}
    			catch (Throwable t) {
    				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);
    			}
    
    			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);
    			}
    
    			// schedule the timer that will clean up the expired checkpoints
    			final Runnable canceller = () -> {
    				synchronized (lock) {
    					// only do the work if the checkpoint is not discarded anyways
    					// note that checkpoint completion discards the pending checkpoint object
    					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 {
    				// re-acquire the coordinator-wide lock
    				synchronized (lock) {
    					// since we released the lock in the meantime, we need to re-check
    					// that the conditions still hold.
    					preCheckBeforeTriggeringCheckpoint(isPeriodic, props.forceCheckpoint());
    
    					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 is already disposed!
    						cancellerHandle.cancel(false);
    					}
    
    					// trigger the master hooks for the 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());
    
    				// send the messages to the tasks that trigger their checkpoint
    				for (Execution execution: executions) {
    					if (props.isSynchronous()) {
    						execution.triggerSynchronousSavepoint(checkpointID, timestamp, checkpointOptions, advanceToEndOfTime);
    					} else {
    						execution.triggerCheckpoint(checkpointID, timestamp, checkpointOptions);
    					}
    				}
    
    				numUnsuccessfulCheckpointsTriggers.set(0);
    				return checkpoint;
    			}
    			catch (Throwable t) {
    				// guard the map against concurrent modifications
    				synchronized (lock) {
    					pendingCheckpoints.remove(checkpointID);
    				}
    
    				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 {
    					checkpointStorageLocation.disposeOnFailure();
    				} catch (Throwable t2) {
    					LOG.warn("Cannot dispose failed checkpoint storage location {}", checkpointStorageLocation, t2);
    				}
    
    				// rethrow the CheckpointException directly.
    				if (t instanceof CheckpointException) {
    					throw (CheckpointException) t;
    				}
    				throw new CheckpointException(CheckpointFailureReason.EXCEPTION, t);
    			}
    		}
    	}
    	
    
    
        //做5个检查，都符合条件才继续；反之，有一个不符合都抛出异常，无法继续checkpoint
        private void preCheckBeforeTriggeringCheckpoint(boolean isPeriodic, boolean forceCheckpoint) throws CheckpointException {
    		// abort if the coordinator has been shutdown in the meantime
    		if (shutdown) {
    			throw new CheckpointException(CheckpointFailureReason.CHECKPOINT_COORDINATOR_SHUTDOWN);
    		}
    
    		// Don't allow periodic checkpoint if scheduling has been disabled
    		if (isPeriodic && !periodicScheduling) {
    			throw new CheckpointException(CheckpointFailureReason.PERIODIC_SCHEDULER_SHUTDOWN);
    		}
    
    		if (!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();
    		}
    	}

package org.apache.flink.runtime.checkpoint;
    
    /**
     * Various reasons why a checkpoint was failure.
     */
    public enum CheckpointFailureReason {
    
    	PERIODIC_SCHEDULER_SHUTDOWN(true, "Periodic checkpoint scheduler is shut down."),
    
    	ALREADY_QUEUED(true, "Another checkpoint request has already been queued."),
    
    	TOO_MANY_CONCURRENT_CHECKPOINTS(true, "The maximum number of concurrent checkpoints is exceeded"),
    
    	MINIMUM_TIME_BETWEEN_CHECKPOINTS(true, "The minimum time between checkpoints is still pending. " +
    			"Checkpoint will be triggered after the minimum time."),
    
    	NOT_ALL_REQUIRED_TASKS_RUNNING(true, "Not all required tasks are currently running."),
    
    	EXCEPTION(true, "An Exception occurred while triggering the checkpoint."),
    
    	CHECKPOINT_EXPIRED(false, "Checkpoint expired before completing."),
    
    	CHECKPOINT_SUBSUMED(false, "Checkpoint has been subsumed."),
    
    	CHECKPOINT_DECLINED(false, "Checkpoint was declined."),
    
    	CHECKPOINT_DECLINED_TASK_NOT_READY(false, "Checkpoint was declined (tasks not ready)"),
    
    	CHECKPOINT_DECLINED_TASK_NOT_CHECKPOINTING(false, "Task does not support checkpointing"),
    
    	CHECKPOINT_DECLINED_SUBSUMED(false, "Checkpoint was canceled because a barrier from newer checkpoint was received."),
    
    	CHECKPOINT_DECLINED_ON_CANCELLATION_BARRIER(false, "Task received cancellation from one of its inputs"),
    
    	CHECKPOINT_DECLINED_ALIGNMENT_LIMIT_EXCEEDED(false, "The checkpoint alignment phase needed to buffer more than the configured maximum bytes"),
    
    	CHECKPOINT_DECLINED_INPUT_END_OF_STREAM(false, "Checkpoint was declined because one input stream is finished"),
    
    	CHECKPOINT_COORDINATOR_SHUTDOWN(false, "CheckpointCoordinator shutdown."),
    
    	CHECKPOINT_COORDINATOR_SUSPEND(false, "Checkpoint Coordinator is suspending."),
    
    	JOB_FAILURE(false, "The job has failed."),
    
    	JOB_FAILOVER_REGION(false, "FailoverRegion is restarting."),
    
    	TASK_CHECKPOINT_FAILURE(false, "Task local checkpoint failure."),
    
    	FINALIZE_CHECKPOINT_FAILURE(false, "Failure to finalize checkpoint."),
    
    	TRIGGER_CHECKPOINT_FAILURE(false, "Trigger checkpoint failure.");
    
    	// ------------------------------------------------------------------------
    
    	private final boolean preFlight;
    	private final String message;
    
    	CheckpointFailureReason(boolean isPreFlight, String message) {
    		this.preFlight = isPreFlight;
    		this.message = message;
    	}
    
    	public String message() {
    		return message;
    	}
    
    	/**
    	 * @return true if this value indicates a failure reason happening before a checkpoint is passed to a job's tasks.
    	 */
    	public boolean isPreFlight() {
    		return preFlight;
    	}
    }

Flink 1.9.2