org.apache.flink.runtime.checkpoint.CheckpointCoordinator

 1     /** Coordinator-wide lock to safeguard the checkpoint updates */
 2     private final Object lock = new Object();  //Coordinator范围的锁
 3 
 4     /** Lock specially to make sure that trigger requests do not overtake each other.
 5      * This is not done with the coordinator-wide lock, because as part of triggering,
 6      * blocking operations may happen (distributed atomic counters).
 7      * Using a dedicated lock, we avoid blocking the processing of ‘acknowledge/decline‘
 8      * messages during that phase. */
 9     private final Object triggerLock = new Object(); //trigger requests的专用锁，避免在获取checkpointID时阻塞对消息的处理。
10 
11     /** Tasks who need to be sent a message when a checkpoint is started */
12     private final ExecutionVertex[] tasksToTrigger;
13 
14     /** Tasks who need to acknowledge a checkpoint before it succeeds */
15     private final ExecutionVertex[] tasksToWaitFor;
16 
17     /** Tasks who need to be sent a message when a checkpoint is confirmed */
18     private final ExecutionVertex[] tasksToCommitTo;
19 
20     /** Map from checkpoint ID to the pending checkpoint */
21     private final Map<Long, PendingCheckpoint> pendingCheckpoints;//待处理的checkpoint
22 
23     /** Actor that receives status updates from the execution graph this coordinator works for */
24     private JobStatusListener jobStatusListener;//Actor实例，监听Job状态变化并根据变化启停定时任务
25 
26     /** Flag whether a trigger request could not be handled immediately. Non-volatile, because only
27      * accessed in synchronized scope */
28     private boolean triggerRequestQueued;//标记一个触发请求是否不能被立即处理。
29 
30     /** Flag marking the coordinator as shut down (not accepting any messages any more) */
31     private volatile boolean shutdown;//coordinator的关闭标志

 1     private final class ScheduledTrigger implements Runnable {
 2         @Override
 3         public void run() {
 4             try {
 5                 triggerCheckpoint(System.currentTimeMillis(), true);
 6             }
 7             catch (Exception e) {
 8                 LOG.error("Exception while triggering checkpoint.", e);
 9             }
10         }
11     }

1     //触发一个新的标准检查点。timestamp为触发检查点的时间戳，isPeriodic标志是否是周期性的触发
2     public boolean triggerCheckpoint(long timestamp, boolean isPeriodic) {
3         return triggerCheckpoint(timestamp, checkpointProperties, checkpointDirectory, isPeriodic).isSuccess();
4     }

  1     CheckpointTriggerResult triggerCheckpoint(
  2             long timestamp,
  3             CheckpointProperties props,
  4             String targetDirectory,
  5             boolean isPeriodic) {
  6 
  7         // Sanity check 如果检查点是存储在外部系统中且targetDirectory为空，报错
  8         if (props.externalizeCheckpoint() && targetDirectory == null) {
  9             throw new IllegalStateException("No target directory specified to persist checkpoint to.");
 10         }
 11 
 12         // make some eager pre-checks 一些checkpoint之前的预检查
 13         synchronized (lock) {
 14             // abort if the coordinator has been shutdown in the meantime
 15             if (shutdown) {
 16                 return new CheckpointTriggerResult(CheckpointDeclineReason.COORDINATOR_SHUTDOWN);
 17             }
 18 
 19             // Don‘t allow periodic checkpoint if scheduling has been disabled
 20             if (isPeriodic && !periodicScheduling) {
 21                 return new CheckpointTriggerResult(CheckpointDeclineReason.PERIODIC_SCHEDULER_SHUTDOWN);
 22             }
 23 
 24             // validate whether the checkpoint can be triggered, with respect to the limit of
 25             // concurrent checkpoints, and the minimum time between checkpoints.
 26             // these checks are not relevant for savepoints
 27             // 验证checkpoint是否能被触发，关于并发检查点的限制和检查点之间的最小时间。
 28             // 判断checkpoint是否被强制。强制checkpoint不受并发检查点最大数量和检查点之间最小时间的限制。
 29             if (!props.forceCheckpoint()) {
 30                 // sanity check: there should never be more than one trigger request queued
 31                 if (triggerRequestQueued) {
 32                     //如果不能被立即触发，直接返回异常
 33                     LOG.warn("Trying to trigger another checkpoint while one was queued already");
 34                     return new CheckpointTriggerResult(CheckpointDeclineReason.ALREADY_QUEUED);
 35                 }
 36 
 37                 // if too many checkpoints are currently in progress, we need to mark that a request is queued
 38                 if (pendingCheckpoints.size() >= maxConcurrentCheckpointAttempts) {
 39                     //如果未完成的检查点太多，大于配置的并发检查点最大数量，则将当前检查点的触发请求设置为不能立即执行。
 40                     triggerRequestQueued = true;
 41                     //如果定时任务已经启动，则取消定时任务的执行。
 42                     if (currentPeriodicTrigger != null) {
 43                         currentPeriodicTrigger.cancel(false);
 44                         currentPeriodicTrigger = null;
 45                     }
 46                     return new CheckpointTriggerResult(CheckpointDeclineReason.TOO_MANY_CONCURRENT_CHECKPOINTS);
 47                 }
 48 
 49                 // make sure the minimum interval between checkpoints has passed
 50                 //检查是否满足checkpoint之间的最小时间间隔的条件
 51                 final long earliestNext = lastCheckpointCompletionNanos + minPauseBetweenCheckpointsNanos;
 52                 final long durationTillNextMillis = (earliestNext - System.nanoTime()) / 1_000_000;
 53 
 54                 if (durationTillNextMillis > 0) {
 55                     if (currentPeriodicTrigger != null) {
 56                         currentPeriodicTrigger.cancel(false);
 57                         currentPeriodicTrigger = null;
 58                     }
 59                     // Reassign the new trigger to the currentPeriodicTrigger
 60                     //此时延迟时间设置为durationTillNextMillis
 61                     currentPeriodicTrigger = timer.scheduleAtFixedRate(
 62                             new ScheduledTrigger(),
 63                             durationTillNextMillis, baseInterval, TimeUnit.MILLISECONDS);
 64 
 65                     return new CheckpointTriggerResult(CheckpointDeclineReason.MINIMUM_TIME_BETWEEN_CHECKPOINTS);
 66                 }
 67             }
 68         }
 69 
 70         // check if all tasks that we need to trigger are running.
 71         // if not, abort the checkpoint
 72         // 检查需要触发checkpoint的所有Tasks是否处于运行状态，如果有一个不满足条件，则不触发检查点
 73         Execution[] executions = new Execution[tasksToTrigger.length];
 74         for (int i = 0; i < tasksToTrigger.length; i++) {
 75             Execution ee = tasksToTrigger[i].getCurrentExecutionAttempt();
 76             if (ee != null && ee.getState() == ExecutionState.RUNNING) {
 77                 executions[i] = ee;
 78             } else {
 79                 LOG.info("Checkpoint triggering task {} is not being executed at the moment. Aborting checkpoint.",
 80                         tasksToTrigger[i].getTaskNameWithSubtaskIndex());
 81                 return new CheckpointTriggerResult(CheckpointDeclineReason.NOT_ALL_REQUIRED_TASKS_RUNNING);
 82             }
 83         }
 84 
 85         // next, check if all tasks that need to acknowledge the checkpoint are running.
 86         // if not, abort the checkpoint
 87         //检查所有需要ack的tasks是否都处于运行状态，如果有一个不满足条件，则不触发检查点。
 88         Map<ExecutionAttemptID, ExecutionVertex> ackTasks = new HashMap<>(tasksToWaitFor.length);
 89 
 90         for (ExecutionVertex ev : tasksToWaitFor) {
 91             Execution ee = ev.getCurrentExecutionAttempt();
 92             if (ee != null) {
 93                 ackTasks.put(ee.getAttemptId(), ev);
 94             } else {
 95                 LOG.info("Checkpoint acknowledging task {} is not being executed at the moment. Aborting checkpoint.",
 96                         ev.getTaskNameWithSubtaskIndex());
 97                 return new CheckpointTriggerResult(CheckpointDeclineReason.NOT_ALL_REQUIRED_TASKS_RUNNING);
 98             }
 99         }
100 
101         // we will actually trigger this checkpoint!
102 
103         // we lock with a special lock to make sure that trigger requests do not overtake each other.
104         // this is not done with the coordinator-wide lock, because the ‘checkpointIdCounter‘
105         // may issue blocking operations. Using a different lock than the coordinator-wide lock,
106         // we avoid blocking the processing of ‘acknowledge/decline‘ messages during that time.
107         // 触发检查点，在triggerLock同步代码块中完成，而不是使用coordinator范围的锁。
108         synchronized (triggerLock) {
109             final long checkpointID;
110             //首先获取checkpointID
111             try {
112                 // this must happen outside the coordinator-wide lock, because it communicates
113                 // with external services (in HA mode) and may block for a while.
114                 checkpointID = checkpointIdCounter.getAndIncrement();
115             }
116             catch (Throwable t) {
117                 int numUnsuccessful = numUnsuccessfulCheckpointsTriggers.incrementAndGet();
118                 LOG.warn("Failed to trigger checkpoint (" + numUnsuccessful + " consecutive failed attempts so far)", t);
119                 return new CheckpointTriggerResult(CheckpointDeclineReason.EXCEPTION);
120             }
121 
122             //创建PendingCheckpoint实例，表示待处理检查点
123             final PendingCheckpoint checkpoint = new PendingCheckpoint(
124                 job,
125                 checkpointID,
126                 timestamp,
127                 ackTasks,
128                 props,
129                 targetDirectory,
130                 executor);
131 
132             if (statsTracker != null) {
133                 PendingCheckpointStats callback = statsTracker.reportPendingCheckpoint(
134                     checkpointID,
135                     timestamp,
136                     props);
137 
138                 checkpoint.setStatsCallback(callback);
139             }
140 
141             // schedule the timer that will clean up the expired checkpoints
142             // 针对当前checkpoints超时进行资源清理的canceller
143             final Runnable canceller = new Runnable() {
144                 @Override
145                 public void run() {
146                     synchronized (lock) {
147                         // only do the work if the checkpoint is not discarded anyways
148                         // note that checkpoint completion discards the pending checkpoint object
149                         if (!checkpoint.isDiscarded()) {
150                             LOG.info("Checkpoint " + checkpointID + " expired before completing.");
151 
152                             checkpoint.abortExpired();
153                             pendingCheckpoints.remove(checkpointID);
154                             rememberRecentCheckpointId(checkpointID);
155 
156                             triggerQueuedRequests();
157                         }
158                     }
159                 }
160             };
161 
162             try {
163                 //重新请求coordinator-wide lock
164                 // re-acquire the coordinator-wide lock
165                 synchronized (lock) {
166                     // since we released the lock in the meantime, we need to re-check
167                     // that the conditions still hold.
168                     // 重新检查触发条件，防止产生竞态条件。这里做二次检查的原因是，中间有一段关于获得checkpointId的代码，不在同步块中。
169                     if (shutdown) {
170                         return new CheckpointTriggerResult(CheckpointDeclineReason.COORDINATOR_SHUTDOWN);
171                     }
172                     else if (!props.forceCheckpoint()) {
173                         if (triggerRequestQueued) {
174                             LOG.warn("Trying to trigger another checkpoint while one was queued already");
175                             return new CheckpointTriggerResult(CheckpointDeclineReason.ALREADY_QUEUED);
176                         }
177 
178                         if (pendingCheckpoints.size() >= maxConcurrentCheckpointAttempts) {
179                             triggerRequestQueued = true;
180                             if (currentPeriodicTrigger != null) {
181                                 currentPeriodicTrigger.cancel(false);
182                                 currentPeriodicTrigger = null;
183                             }
184                             return new CheckpointTriggerResult(CheckpointDeclineReason.TOO_MANY_CONCURRENT_CHECKPOINTS);
185                         }
186 
187                         // make sure the minimum interval between checkpoints has passed
188                         final long earliestNext = lastCheckpointCompletionNanos + minPauseBetweenCheckpointsNanos;
189                         final long durationTillNextMillis = (earliestNext - System.nanoTime()) / 1_000_000;
190 
191                         if (durationTillNextMillis > 0) {
192                             if (currentPeriodicTrigger != null) {
193                                 currentPeriodicTrigger.cancel(false);
194                                 currentPeriodicTrigger = null;
195                             }
196 
197                             // Reassign the new trigger to the currentPeriodicTrigger
198                             currentPeriodicTrigger = timer.scheduleAtFixedRate(
199                                     new ScheduledTrigger(),
200                                     durationTillNextMillis, baseInterval, TimeUnit.MILLISECONDS);
201 
202                             return new CheckpointTriggerResult(CheckpointDeclineReason.MINIMUM_TIME_BETWEEN_CHECKPOINTS);
203                         }
204                     }
205 
206                     LOG.info("Triggering checkpoint " + checkpointID + " @ " + timestamp);
207 
208                     //将checkpoint加入到pendingCheckpoints中
209                     pendingCheckpoints.put(checkpointID, checkpoint);
210 
211                     //启动超时canceller，延迟checkpointTimeout执行
212                     ScheduledFuture<?> cancellerHandle = timer.schedule(
213                             canceller,
214                             checkpointTimeout, TimeUnit.MILLISECONDS);
215 
216                     if (!checkpoint.setCancellerHandle(cancellerHandle)) {
217                         // checkpoint is already disposed!
218                         cancellerHandle.cancel(false);
219                     }
220 
221                     // trigger the master hooks for the checkpoint
222                     final List<MasterState> masterStates = MasterHooks.triggerMasterHooks(masterHooks.values(),
223                             checkpointID, timestamp, executor, Time.milliseconds(checkpointTimeout));
224                     for (MasterState s : masterStates) {
225                         checkpoint.addMasterState(s);
226                     }
227                 }
228                 // end of lock scope
229 
230                 CheckpointOptions checkpointOptions;
231                 if (!props.isSavepoint()) {
232                     checkpointOptions = CheckpointOptions.forFullCheckpoint();
233                 } else {
234                     checkpointOptions = CheckpointOptions.forSavepoint(targetDirectory);
235                 }
236 
237                 // send the messages to the tasks that trigger their checkpoint
238                 // 向tasks发送消息，触发它们的检查点
239                 for (Execution execution: executions) {
240                     execution.triggerCheckpoint(checkpointID, timestamp, checkpointOptions);
241                 }
242 
243                 numUnsuccessfulCheckpointsTriggers.set(0);
244                 return new CheckpointTriggerResult(checkpoint);
245             }
246             catch (Throwable t) {
247                 // guard the map against concurrent modifications
248                 synchronized (lock) {
249                     pendingCheckpoints.remove(checkpointID);
250                 }
251 
252                 int numUnsuccessful = numUnsuccessfulCheckpointsTriggers.incrementAndGet();
253                 LOG.warn("Failed to trigger checkpoint {}. ({} consecutive failed attempts so far)",
254                         checkpointID, numUnsuccessful, t);
255 
256                 if (!checkpoint.isDiscarded()) {
257                     checkpoint.abortError(new Exception("Failed to trigger checkpoint", t));
258                 }
259                 return new CheckpointTriggerResult(CheckpointDeclineReason.EXCEPTION);
260             }
261 
262         } // end trigger lock
263     }

 1     public void startCheckpointScheduler() {
 2         synchronized (lock) {
 3             if (shutdown) {
 4                 throw new IllegalArgumentException("Checkpoint coordinator is shut down");
 5             }
 6             //保证所有以前的timer被取消
 7             stopCheckpointScheduler();
 8 
 9             periodicScheduling = true;
10             //scheduleAtFixedRate方法是以固定延迟和固定时间间隔周期性的执行任务
11             currentPeriodicTrigger = timer.scheduleAtFixedRate(
12                     new ScheduledTrigger(), 
13                     baseInterval, baseInterval, TimeUnit.MILLISECONDS);
14         }
15     }

 1     //重置一些标记变量，释放资源
 2     public void stopCheckpointScheduler() {
 3         synchronized (lock) {
 4             triggerRequestQueued = false;
 5             periodicScheduling = false;
 6 
 7             if (currentPeriodicTrigger != null) {
 8                 currentPeriodicTrigger.cancel(false);//取消当前周期的触发任务
 9                 currentPeriodicTrigger = null;
10             }
11 
12             //pendingCheckpoints中存的是待执行的检查点
13             for (PendingCheckpoint p : pendingCheckpoints.values()) {
14                 p.abortError(new Exception("Checkpoint Coordinator is suspending."));
15             }
16             pendingCheckpoints.clear();//清空pendingCheckpoints
17             numUnsuccessfulCheckpointsTriggers.set(0);
18         }
19     }

org.apache.flink.runtime.checkpoint.CheckpointCoordinatorDeActivator

 1     //actor的实现类，监听JobStatus的变化，激活和取消周期性的checkpoint调度任务。
 2     public class CheckpointCoordinatorDeActivator implements JobStatusListener {
 3 
 4         private final CheckpointCoordinator coordinator;
 5 
 6         public CheckpointCoordinatorDeActivator(CheckpointCoordinator coordinator) {
 7             this.coordinator = checkNotNull(coordinator);
 8         }
 9 
10         @Override
11         public void jobStatusChanges(JobID jobId, JobStatus newJobStatus, long timestamp, Throwable error) {
12             if (newJobStatus == JobStatus.RUNNING) {
13                 // start the checkpoint scheduler
14                 // 一旦监听到JobStatus变为RUNNING，就会启动定时任务
15                 coordinator.startCheckpointScheduler();
16             } else {
17                 // anything else should stop the trigger for now
18                 coordinator.stopCheckpointScheduler();
19             }
20         }
21     }

 1     public JobStatusListener createActivatorDeactivator() {
 2         synchronized (lock) {
 3             if (shutdown) {
 4                 throw new IllegalArgumentException("Checkpoint coordinator is shut down");
 5             }
 6 
 7             if (jobStatusListener == null) {
 8                 jobStatusListener = new CheckpointCoordinatorDeActivator(this);
 9             }
10 
11             return jobStatusListener;
12         }
13     }

1     /** The job to which this message belongs */
2     private final JobID job;
3     /** The task execution that is source/target of the checkpoint message */  
4     private final ExecutionAttemptID taskExecutionId; //检查点的source/target task
5     /** The ID of the checkpoint that this message coordinates */
6     private final long checkpointId;

1         // send the messages to the tasks that trigger their checkpoint
2         for (Execution execution: executions) {
3             execution.triggerCheckpoint(checkpointID, timestamp, checkpointOptions);
4         }

 1         //在该execution的task上触发一个新的checkpoint
 2         public void triggerCheckpoint(long checkpointId, long timestamp, CheckpointOptions checkpointOptions) {
 3             //获取Resource
 4             final SimpleSlot slot = assignedResource;//获取Slot
 5 
 6             if (slot != null) {
 7                 //TaskManagerGateway是用于和TaskManager通信的抽象基础类
 8                 final TaskManagerGateway taskManagerGateway = slot.getTaskManagerGateway();
 9                 //通过taskManagerGateway向TaskManager发送消息
10                 taskManagerGateway.triggerCheckpoint(attemptId, getVertex().getJobId(), checkpointId, timestamp, checkpointOptions);
11             } else {
12                 LOG.debug("The execution has no slot assigned. This indicates that the execution is " +
13                     "no longer running.");
14             }
15         }

 1         public void triggerCheckpoint(
 2                 ExecutionAttemptID executionAttemptID,
 3                 JobID jobId,
 4                 long checkpointId,
 5                 long timestamp,
 6                 CheckpointOptions checkpointOptions) {
 7 
 8             Preconditions.checkNotNull(executionAttemptID);
 9             Preconditions.checkNotNull(jobId);
10             //新建了一个TriggerCheckpoint消息，通过actorGateway的tell方法（异步发送，没有返回结果）发送这个消息
11             //ActorGateway是基于actor通信的接口
12             actorGateway.tell(new TriggerCheckpoint(jobId, executionAttemptID, checkpointId, timestamp, checkpointOptions));
13         }

1         @Override
2         public void tell(Object message) {
3             Object newMessage = decorator.decorate(message);
4             //通过ActorRef实例actor发送消息，ActorRef是akka中的类。以后需要研究Akka的实现机制。
5             actor.tell(newMessage, ActorRef.noSender());
6         }

 1         //该方法为TaskManager的消息处理中心。接收消息，按消息的种类调用不同的方法处理。
 2         override def handleMessage: Receive = {
 3             case message: TaskMessage => handleTaskMessage(message)
 4 
 5             //这个就是处理checkpoints相关的消息
 6             case message: AbstractCheckpointMessage => handleCheckpointingMessage(message)
 7 
 8             case JobManagerLeaderAddress(address, newLeaderSessionID) =>
 9               handleJobManagerLeaderAddress(address, newLeaderSessionID)
10 
11             case message: RegistrationMessage => handleRegistrationMessage(message)
12 
13             ...
14         }

 1         //处理Checkpoint相关的消息
 2         private def handleCheckpointingMessage(actorMessage: AbstractCheckpointMessage): Unit = {
 3 
 4             actorMessage match {
 5               //触发Checkpoint消息
 6               case message: TriggerCheckpoint =>
 7                 val taskExecutionId = message.getTaskExecutionId
 8                 val checkpointId = message.getCheckpointId
 9                 val timestamp = message.getTimestamp
10                 val checkpointOptions = message.getCheckpointOptions
11 
12                 log.debug(s"Receiver TriggerCheckpoint $checkpointId@$timestamp for $taskExecutionId.")
13 
14                 val task = runningTasks.get(taskExecutionId)
15                 if (task != null) {
16                   //调用Task的triggerCheckpointBarrier方法，触发Checkpoint Barrier，Barrier实现机制的细节以后讨论。
17                   task.triggerCheckpointBarrier(checkpointId, timestamp, checkpointOptions)
18                 } else {
19                   log.debug(s"TaskManager received a checkpoint request for unknown task $taskExecutionId.")
20                 }
21               //Checkpoint完成通知消息
22               case message: NotifyCheckpointComplete =>
23                 val taskExecutionId = message.getTaskExecutionId
24                 val checkpointId = message.getCheckpointId
25                 val timestamp = message.getTimestamp
26 
27                 log.debug(s"Receiver ConfirmCheckpoint $checkpointId@$timestamp for $taskExecutionId.")
28 
29                 val task = runningTasks.get(taskExecutionId)
30                 if (task != null) {
31                   //调用Task的notifyCheckpointComplete方法，进行相关处理
32                   task.notifyCheckpointComplete(checkpointId)
33                 } else {
34                   log.debug(
35                     s"TaskManager received a checkpoint confirmation for unknown task $taskExecutionId.")
36                 }
37 
38               // unknown checkpoint message
39               case _ => unhandled(actorMessage)
40             }
41           }

1         public void acknowledgeCheckpoint(
2                 long checkpointId,
3                 CheckpointMetrics checkpointMetrics,
4                 SubtaskState checkpointStateHandles) {
5             //通过CheckpointResponder接口的实例checkpointResponder发送ack消息
6             checkpointResponder.acknowledgeCheckpoint(
7                     jobId, executionId, checkpointId, checkpointMetrics,
8                     checkpointStateHandles);
9         }

 1         @Override
 2         public void acknowledgeCheckpoint(
 3                 JobID jobID,
 4                 ExecutionAttemptID executionAttemptID,
 5                 long checkpointId,
 6                 CheckpointMetrics checkpointMetrics,
 7                 SubtaskState checkpointStateHandles) {
 8             //新建一个AcknowledgeCheckpoint消息
 9             AcknowledgeCheckpoint message = new AcknowledgeCheckpoint(
10                     jobID, executionAttemptID, checkpointId, checkpointMetrics,
11                     checkpointStateHandles);
12             //通过actorGateway发送出去
13             actorGateway.tell(message);
14         }

org.apache.flink.runtime.taskmanager.Task

1         try {
2             boolean success = statefulTask.triggerCheckpoint(checkpointMetaData, checkpointOptions);
3             if (!success) {
4                 //通过CheckpointResponder发送消息，类似发送AcknowledgeCheckpoint消息
5                 checkpointResponder.declineCheckpoint(
6                         getJobID(), getExecutionId(), checkpointID,
7                         new CheckpointDeclineTaskNotReadyException(taskName));
8             }
9         }