mapreduce job提交流程源码级分析（三）

 1  public JobStatus submitJob(JobID jobId, String jobSubmitDir, Credentials ts)
 2       throws IOException {
 3     JobInfo jobInfo = null;
 4     UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
 5     synchronized (this) {
 6       if (jobs.containsKey(jobId)) {
 7         // job already running, don‘t start twice
 8         return jobs.get(jobId).getStatus();
 9       }
10       jobInfo = new JobInfo(jobId, new Text(ugi.getShortUserName()),
11           new Path(jobSubmitDir));
12     }
13     
14     // Create the JobInProgress, do not lock the JobTracker since
15     // we are about to copy job.xml from HDFS
16     //当JobTracker接收到新的job请求（即submitJob()函数被调用）后，
17     //会创建一个JobInProgress对象并通过它来管理和调度任务。
18     //JobInProgress在创建的时候会初始化一系列与任务有关的参数，调用到FileSystem，
19     //把在JobClient端上传的所有任务文件下载到本地的文件系统中的临时目录里。这其中包括上传的*.jar文件包、
20     //记录配置信息的xml、记录分割信息的文件。
21     JobInProgress job = null;
22     try {
23       job = new JobInProgress(this, this.conf, jobInfo, 0, ts);
24     } catch (Exception e) {
25       throw new IOException(e);
26     }
27     
28     synchronized (this) {
29       // check if queue is RUNNING
30       String queue = job.getProfile().getQueueName();
31       if (!queueManager.isRunning(queue)) {
32         throw new IOException("Queue \"" + queue + "\" is not running");
33       }
34       try {
35         aclsManager.checkAccess(job, ugi, Operation.SUBMIT_JOB);
36       } catch (IOException ioe) {
37         LOG.warn("Access denied for user " + job.getJobConf().getUser()
38             + ". Ignoring job " + jobId, ioe);
39         job.fail();
40         throw ioe;
41       }
42 
43       // Check the job if it cannot run in the cluster because of invalid memory
44       // requirements.
45       try {
46         checkMemoryRequirements(job);
47       } catch (IOException ioe) {
48         throw ioe;
49       }
50       boolean recovered = true; // TODO: Once the Job recovery code is there,
51       // (MAPREDUCE-873) we
52       // must pass the "recovered" flag accurately.
53       // This is handled in the trunk/0.22
54       if (!recovered) {
55         // Store the information in a file so that the job can be recovered
56         // later (if at all)
57         Path jobDir = getSystemDirectoryForJob(jobId);
58         FileSystem.mkdirs(fs, jobDir, new FsPermission(SYSTEM_DIR_PERMISSION));
59         FSDataOutputStream out = fs.create(getSystemFileForJob(jobId));
60         jobInfo.write(out);
61         out.close();
62       }
63       
64       // Submit the job
65       JobStatus status;
66       try {
67         status = addJob(jobId, job);
68       } catch (IOException ioe) {
69         LOG.info("Job " + jobId + " submission failed!", ioe);
70         status = job.getStatus();
71         status.setFailureInfo(StringUtils.stringifyException(ioe));
72         failJob(job);
73         throw ioe;
74       }
75       
76       return status;
77     }
78   }