/*** Submits a job to the job manager. The job is registered at the libraryCacheManager which* creates the job's class loader. The job graph is appended to the corresponding execution* graph and the execution vertices are queued for scheduling.** @param jobGraph representing the Flink job* @param jobInfo the job info* @param isRecovery Flag indicating whether this is a recovery or initial submission*/private def submitJob(jobGraph: JobGraph, jobInfo: JobInfo, isRecovery: Boolean = false): Unit = {if (jobGraph == null) {jobInfo.notifyClients(decorateMessage(JobResultFailure(new SerializedThrowable(new JobSubmissionException(null, "JobGraph must not be null.")))))}else {val jobId = jobGraph.getJobIDval jobName = jobGraph.getNamevar executionGraph: ExecutionGraph = nulltry {// Important: We need to make sure that the library registration is the first action,// because this makes sure that the uploaded jar files are removed in case of// unsuccessfultry {libraryCacheManager.registerJob(jobGraph.getJobID, jobGraph.getUserJarBlobKeys,jobGraph.getClasspaths)}var userCodeLoader = libraryCacheManager.getClassLoader(jobGraph.getJobID) //加载Jar
val restartStrategy = //加载重启策略
          Option(jobGraph.getSerializedExecutionConfig().deserializeValue(userCodeLoader).getRestartStrategy()).map(RestartStrategyFactory.createRestartStrategy).filter(p => p != null) match {case Some(strategy) => strategycase None => restartStrategyFactory.createRestartStrategy()}val jobMetrics = jobManagerMetricGroup match { //生成job manager metric groupcase Some(group) =>group.addJob(jobGraph) match {case (jobGroup:Any) => jobGroupcase null => new UnregisteredMetricsGroup()}case None =>new UnregisteredMetricsGroup()}val numSlots = scheduler.getTotalNumberOfSlots() //现有的slots数目// see if there already exists an ExecutionGraph for the corresponding job IDval registerNewGraph = currentJobs.get(jobGraph.getJobID) match {case Some((graph, currentJobInfo)) =>executionGraph = graphcurrentJobInfo.setLastActive()falsecase None =>true}executionGraph = ExecutionGraphBuilder.buildGraph( //build ExecutionGraph
          executionGraph,jobGraph,flinkConfiguration,futureExecutor,ioExecutor,userCodeLoader,checkpointRecoveryFactory,Time.of(timeout.length, timeout.unit),restartStrategy,jobMetrics,numSlots,log.logger)if (registerNewGraph) { //如果是新的JobGraph，注册到currentJobs
          currentJobs.put(jobGraph.getJobID, (executionGraph, jobInfo))}// get notified about job status changesexecutionGraph.registerJobStatusListener( //jobmananger加到通知listenersnew StatusListenerMessenger(self, leaderSessionID.orNull))jobInfo.clients foreach { //client加到通知listeners// the sender wants to be notified about state changescase (client, ListeningBehaviour.EXECUTION_RESULT_AND_STATE_CHANGES) =>val listener  = new StatusListenerMessenger(client, leaderSessionID.orNull)executionGraph.registerExecutionListener(listener)executionGraph.registerJobStatusListener(listener)case _ => // do nothing
        }} catch { //失败case t: Throwable =>log.error(s"Failed to submit job $jobId ($jobName)", t)libraryCacheManager.unregisterJob(jobId)currentJobs.remove(jobId)if (executionGraph != null) {executionGraph.fail(t) //fail executionGraph
          }val rt: Throwable = if (t.isInstanceOf[JobExecutionException]) {t} else {new JobExecutionException(jobId, s"Failed to submit job $jobId ($jobName)", t)}jobInfo.notifyClients(decorateMessage(JobResultFailure(new SerializedThrowable(rt)))) //通知提交失败return}//上面是准备executionGraph，下面是异步提交// execute the recovery/writing the jobGraph into the SubmittedJobGraphStore asynchronously// because it is a blocking operation
      future {try {if (isRecovery) {// this is a recovery of a master failure (this master takes over)executionGraph.restoreLatestCheckpointedState(false, false) //加载checkpoint状态
          }else {// load a savepoint only if this is not starting from a newer checkpoint// as part of an master failure recoveryval savepointSettings = jobGraph.getSavepointRestoreSettingsif (savepointSettings.restoreSavepoint()) { //处理savePointtry {val savepointPath = savepointSettings.getRestorePath()val allowNonRestored = savepointSettings.allowNonRestoredState()log.info(s"Starting job from savepoint '$savepointPath'" +(if (allowNonRestored) " (allowing non restored state)" else "") + ".")// load the savepoint as a checkpoint into the systemval savepoint: CompletedCheckpoint = SavepointLoader.loadAndValidateSavepoint(jobId,executionGraph.getAllVertices,savepointPath,executionGraph.getUserClassLoader,allowNonRestored)executionGraph.getCheckpointCoordinator.getCheckpointStore.addCheckpoint(savepoint)// Reset the checkpoint ID counterval nextCheckpointId: Long = savepoint.getCheckpointID + 1log.info(s"Reset the checkpoint ID to $nextCheckpointId")executionGraph.getCheckpointCoordinator.getCheckpointIdCounter.setCount(nextCheckpointId)executionGraph.restoreLatestCheckpointedState(true, allowNonRestored)} catch {case e: Exception =>jobInfo.notifyClients(decorateMessage(JobResultFailure(new SerializedThrowable(e))))throw new SuppressRestartsException(e)}}try {submittedJobGraphs.putJobGraph(new SubmittedJobGraph(jobGraph, jobInfo)) //存储该JobGraph到zk，ZooKeeperSubmittedJobGraphStore} catch {case t: Throwable =>// Don't restart the execution if this fails. Otherwise, the// job graph will skip ZooKeeper in case of HA.
                jobInfo.notifyClients(decorateMessage(JobResultFailure(new SerializedThrowable(t))))throw new SuppressRestartsException(t)}}jobInfo.notifyClients(decorateMessage(JobSubmitSuccess(jobGraph.getJobID))) //通知clients提交成功if (leaderElectionService.hasLeadership) {// There is a small chance that multiple job managers schedule the same job after if// they try to recover at the same time. This will eventually be noticed, but can not be// ruled out from the beginning.// NOTE: Scheduling the job for execution is a separate action from the job submission.// The success of submitting the job must be independent from the success of scheduling// the job.log.info(s"Scheduling job $jobId ($jobName).")executionGraph.scheduleForExecution(scheduler) //开始调度} else {// Remove the job graph. Otherwise it will be lingering around and possibly removed from// ZooKeeper by this JM.self ! decorateMessage(RemoveJob(jobId, removeJobFromStateBackend = false))log.warn(s"Submitted job $jobId, but not leader. The other leader needs to recover " +"this. I am not scheduling the job for execution.")}} catch {case t: Throwable => try {executionGraph.fail(t)} catch {case tt: Throwable =>log.error("Error while marking ExecutionGraph as failed.", tt)}}}(context.dispatcher)}}

private boolean tryRestartOrFail() {JobStatus currentState = state;if (currentState == JobStatus.FAILING || currentState == JobStatus.RESTARTING) {synchronized (progressLock) { //锁final boolean isFailureCauseAllowingRestart = !(failureCause instanceof SuppressRestartsException);final boolean isRestartStrategyAllowingRestart = restartStrategy.canRestart(); //重启策略是否允许重启boolean isRestartable = isFailureCauseAllowingRestart && isRestartStrategyAllowingRestart;if (isRestartable && transitionState(currentState, JobStatus.RESTARTING)) {restartStrategy.restart(this);return true;} else if (!isRestartable && transitionState(currentState, JobStatus.FAILED, failureCause)) { //如果不允许重启，就failedfinal List<String> reasonsForNoRestart = new ArrayList<>(2);if (!isFailureCauseAllowingRestart) {reasonsForNoRestart.add("a type of SuppressRestartsException was thrown");}if (!isRestartStrategyAllowingRestart) {reasonsForNoRestart.add("the restart strategy prevented it");}LOG.info("Could not restart the job {} ({}) because {}.", getJobName(), getJobID(),StringUtils.join(reasonsForNoRestart, " and "), failureCause);postRunCleanup();return true;} else {// we must have changed the state concurrently, thus we cannot complete this operationreturn false;}}} else {// this operation is only allowed in the state FAILING or RESTARTINGreturn false;}}

void jobVertexInFinalState() {synchronized (progressLock) {if (numFinishedJobVertices >= verticesInCreationOrder.size()) {throw new IllegalStateException("All vertices are already finished, cannot transition vertex to finished.");}numFinishedJobVertices++;if (numFinishedJobVertices == verticesInCreationOrder.size()) { //当所有的vertices都已经finished// we are done, transition to the final state
            JobStatus current;while (true) {current = this.state;if (current == JobStatus.RUNNING) {if (transitionState(current, JobStatus.FINISHED)) {postRunCleanup();break;}}else if (current == JobStatus.CANCELLING) {if (transitionState(current, JobStatus.CANCELED)) {postRunCleanup();break;}}else if (current == JobStatus.FAILING) {if (tryRestartOrFail()) { //如果failing，调用tryRestartOrFailbreak;}// concurrent job status change, let's check again}

} else if (current == JobStatus.RESTARTING) {this.failureCause = t;if (tryRestartOrFail()) {return;}// concurrent job status change, let's check again
}

@Overridepublic void restart(final ExecutionGraph executionGraph) {currentRestartAttempt++;FlinkFuture.supplyAsync(ExecutionGraphRestarter.restartWithDelay(executionGraph, delayBetweenRestartAttempts), executionGraph.getFutureExecutor());}

public void restart() {try {synchronized (progressLock) {this.currentExecutions.clear();Collection<CoLocationGroup> colGroups = new HashSet<>();for (ExecutionJobVertex jv : this.verticesInCreationOrder) {CoLocationGroup cgroup = jv.getCoLocationGroup();if(cgroup != null && !colGroups.contains(cgroup)){cgroup.resetConstraints();colGroups.add(cgroup);}jv.resetForNewExecution();}for (int i = 0; i < stateTimestamps.length; i++) {if (i != JobStatus.RESTARTING.ordinal()) {// Only clear the non restarting state in order to preserve when the job was// restarted. This is needed for the restarting time gaugestateTimestamps[i] = 0;}}numFinishedJobVertices = 0;transitionState(JobStatus.RESTARTING, JobStatus.CREATED);// if we have checkpointed state, reload it into the executionsif (checkpointCoordinator != null) {checkpointCoordinator.restoreLatestCheckpointedState(getAllVertices(), false, false);}}scheduleForExecution(slotProvider); //加入schedule
        }catch (Throwable t) {LOG.warn("Failed to restart the job.", t);fail(t);}}

private void configureCheckpointing() {CheckpointConfig cfg = streamGraph.getCheckpointConfig();long interval = cfg.getCheckpointInterval();if (interval > 0) {// check if a restart strategy has been set, if not then set the FixedDelayRestartStrategyif (streamGraph.getExecutionConfig().getRestartStrategy() == null) {// if the user enabled checkpointing, the default number of exec retries is infinite.
                streamGraph.getExecutionConfig().setRestartStrategy(RestartStrategies.fixedDelayRestart(Integer.MAX_VALUE, DEFAULT_RESTART_DELAY));}}