Update the job with details about the file split
@param job the job configuration to update
@param inputSplit the file split
2458
private void updateJobWithSplit(final JobConf job, InputSplit inputSplit) {
    if (inputSplit instanceof FileSplit) {
      FileSplit fileSplit = (FileSplit) inputSplit;
      job.set("map.input.file", fileSplit.getPath().toString());
      job.setLong("map.input.start", fileSplit.getStart());
      job.setLong("map.input.length", fileSplit.getLength());
    }
    LOG.info("split: " + inputSplit.toString());
  }
661986
synchronized void finalizeJob(JobInProgress job) {
    // Mark the 'non-running' tasks for pruning
    markCompletedJob(job);

    JobEndNotifier.registerNotification(job.getJobConf(), job.getStatus());

    // start the merge of log files
    JobID id = job.getStatus().getJobID();

    // mark the job as completed
    try {
      JobHistory.JobInfo.markCompleted(id);
    } catch (IOException ioe) {
      LOG.info("Failed to mark job " + id + " as completed!", ioe);
    }

    final JobTrackerInstrumentation metrics = getInstrumentation();
    metrics.finalizeJob(conf, id);

    long now = getClock().getTime();

    // mark the job for cleanup at all the trackers
    addJobForCleanup(id);

    try {
      File userFileForJob =
        new File(lDirAlloc.getLocalPathToRead(SUBDIR + "/" + id,
                                              conf).toString());
      if (userFileForJob != null) {
        userFileForJob.delete();
      }
    } catch (IOException ioe) {
      LOG.info("Failed to delete job id mapping for job " + id, ioe);
    }

    // add the blacklisted trackers to potentially faulty list
    if (job.getStatus().getRunState() == JobStatus.SUCCEEDED) {
      if (job.getNoOfBlackListedTrackers() > 0) {
        for (Entry<String, List<String>> e : job.getBlackListedTrackers().entrySet()) {
          String tt = e.getKey();
          String jobName = job.getJobID().toString();
          String[] exceptions = e.getValue().toArray(new String[0]);

          faultyTrackers.incrementFaults(tt, new JobFault(tt, jobName, exceptions));
        }
      }
    }

    String jobUser = job.getProfile().getUser();
    //add to the user to jobs mapping
    synchronized (userToJobsMap) {
      ArrayList<JobInProgress> userJobs = userToJobsMap.get(jobUser);
      if (userJobs == null) {
        userJobs =  new ArrayList<JobInProgress>();
        userToJobsMap.put(jobUser, userJobs);
      }
      userJobs.add(job);
    }
  }
1269847
public synchronized void addJob(JobInProgress job) {
    String poolName = getPoolName(job);
    LOG.info("Adding job " + job.getJobID() + " to pool " + poolName +
             ", originally from pool " +
             job.getJobConf().get(EXPLICIT_POOL_PROPERTY));
    getPool(poolName).addJob(job);
  }
935973
public Set<String> configureJob(Job job) throws FileNotFoundException, IOException, TupleMRException {
		Set<String> instanceFiles = new HashSet<String>();
		for(Output output : getNamedOutputs()) {
			try {
				if(output.isDefault) {
					instanceFiles.add(PangoolMultipleOutputs.setDefaultNamedOutput(job, output.outputFormat,
					    output.keyClass, output.valueClass));
				} else {
					instanceFiles.add(PangoolMultipleOutputs.addNamedOutput(job, output.name, output.outputFormat,
					    output.keyClass, output.valueClass));
				}
			} catch(URISyntaxException e1) {
				throw new TupleMRException(e1);
			}
			for(Map.Entry<String, String> contextKeyValue : output.specificContext.entrySet()) {
				PangoolMultipleOutputs.addNamedOutputContext(job, output.name, contextKeyValue.getKey(),
				    contextKeyValue.getValue());
			}
		}
		return instanceFiles;
	}
994987
public synchronized void initTasks()
  throws IOException, KillInterruptedException {
    if (tasksInited.get() || isComplete()) {
      return;
    }
    synchronized(jobInitKillStatus){
      if(jobInitKillStatus.killed || jobInitKillStatus.initStarted) {
        return;
      }
      jobInitKillStatus.initStarted = true;
    }

    LOG.info("Initializing " + jobId);

    // log job info
    JobHistory.JobInfo.logSubmitted(getJobID(), conf, jobFile.toString(),
                                    this.startTime, hasRestarted());
    // log the job priority
    setPriority(this.priority);

    //
    // read input splits and create a map per a split
    //
    String jobFile = profile.getJobFile();

    Path sysDir = new Path(this.jobtracker.getSystemDir());
    FileSystem fs = sysDir.getFileSystem(conf);
    DataInputStream splitFile =
      fs.open(new Path(conf.get("mapred.job.split.file")));
    JobClient.RawSplit[] splits;
    try {
      splits = JobClient.readSplitFile(splitFile);
    } finally {
      splitFile.close();
    }
    numMapTasks = splits.length;


    // if the number of splits is larger than a configured value
    // then fail the job.
    int maxTasks = jobtracker.getMaxTasksPerJob();
    if (maxTasks > 0 && numMapTasks + numReduceTasks > maxTasks) {
      throw new IOException(
                "The number of tasks for this job " +
                (numMapTasks + numReduceTasks) +
                " exceeds the configured limit " + maxTasks);
    }
    jobtracker.getInstrumentation().addWaitingMaps(getJobID(), numMapTasks);
    jobtracker.getInstrumentation().addWaitingReduces(getJobID(), numReduceTasks);

    maps = new TaskInProgress[numMapTasks];
    for(int i=0; i < numMapTasks; ++i) {
      inputLength += splits[i].getDataLength();
      maps[i] = new TaskInProgress(jobId, jobFile,
                                   splits[i],
                                   conf, this, i, numSlotsPerMap);
    }
    LOG.info("Input size for job " + jobId + " = " + inputLength
        + ". Number of splits = " + splits.length);
    if (numMapTasks > 0) {
      nonRunningMapCache = createCache(splits, maxLevel);
    }

    // set the launch time
    this.launchTime = JobTracker.getClock().getTime();
    jobtracker.getInstrumentation().addLaunchedJobs(
      this.launchTime - this.startTime);

    //
    // Create reduce tasks
    //
    this.reduces = new TaskInProgress[numReduceTasks];
    for (int i = 0; i < numReduceTasks; i++) {
      reduces[i] = new TaskInProgress(jobId, jobFile,
                                      numMapTasks, i,
                                      conf, this, numSlotsPerReduce);
      nonRunningReduces.add(reduces[i]);
    }

    // Calculate the minimum number of maps to be complete before
    // we should start scheduling reduces
    completedMapsForReduceSlowstart =
      (int)Math.ceil(
          (conf.getFloat("mapred.reduce.slowstart.completed.maps",
                         DEFAULT_COMPLETED_MAPS_PERCENT_FOR_REDUCE_SLOWSTART) *
           numMapTasks));
    // The thresholds of total maps and reduces for scheduling reducers
    // immediately.
    rushReduceMaps =
      conf.getInt(RUSH_REDUCER_MAP_THRESHOLD, rushReduceMaps);
    rushReduceReduces =
      conf.getInt(RUSH_REDUCER_REDUCE_THRESHOLD, rushReduceReduces);
    maxFetchFailuresPerMapper = conf.getInt(MAX_FETCH_FAILURES_PER_MAP_KEY,
        MAX_FETCH_FAILURES_PER_MAP_DEFAULT);

    initSetupCleanupTasks(jobFile);

    synchronized(jobInitKillStatus){
      jobInitKillStatus.initDone = true;
      if(jobInitKillStatus.killed) {
        throw new KillInterruptedException("Job " + jobId + " killed in init");
      }
    }

    tasksInited.set(true);
    JobHistory.JobInfo.logInited(profile.getJobID(), this.launchTime,
        numMapTasks, numReduceTasks);

    // Log the number of map and reduce tasks
    LOG.info("Job " + jobId + " initialized successfully with " + numMapTasks
        + " map tasks and " + numReduceTasks + " reduce tasks.");
    refreshIfNecessary();
  }
426887
@Override
    protected void updateFaxJobFromInputDataImpl(HTTPRequest inputData,FaxJob faxJob)
    {
        //get parameters text as map
        Map<String,String> queryStringMap=this.convertParametersTextToMap(inputData);

        //update fax job
        this.updateFaxJobFromRequestImpl(inputData,faxJob,queryStringMap);
    }
485149
private void read() throws IOException {
        lastModified = jobFile.lastModified();
        if (jobFile.exists()) {
            try (final InputStream in = new FileInputStream(jobFile)) {
                final XMLReader parser = XMLUtils.getXMLReader();
                parser.setContentHandler(new JobHandler(prop, files));

                parser.parse(new InputSource(in));
            } catch (final SAXException e) {
                throw new IOException("Failed to read job file: " + e.getMessage());
            }
        } else {
            // defaults
            prop.put(PROPERTY_GENERATE_COPY_OUTER, Generate.NOT_GENERATEOUTTER.toString());
            prop.put(PROPERTY_ONLY_TOPIC_IN_MAP, Boolean.toString(false));
            prop.put(PROPERTY_OUTER_CONTROL, OutterControl.WARN.toString());
        }
    }
829355
public void add(InputSplit s) throws IOException {
    if (null == splits) {
      throw new IOException("Uninitialized InputSplit");
    }
    if (fill == splits.length) {
      throw new IOException("Too many splits");
    }
    splits[fill++] = s;
    totsize += s.getLength();
  }
676684
@Override
  public void readFields(DataInput in) throws IOException {
    this.filename = Text.readString(in);
    this.jobid = Text.readString(in);
    this.isJobConfFile = in.readBoolean();
    this.isJobHistoryFile = in.readBoolean();
  }
1078971
public synchronized void failJob(JobInProgress job) {
    if (null == job) {
      LOG.info("Fail on null job is not valid");
      return;
    }

    JobStatus prevStatus = (JobStatus)job.getStatus().clone();
    LOG.info("Failing job " + job.getJobID());
    job.fail();

    // Inform the listeners if the job state has changed
    JobStatus newStatus = (JobStatus)job.getStatus().clone();
    if (prevStatus.getRunState() != newStatus.getRunState()) {
      JobStatusChangeEvent event =
        new JobStatusChangeEvent(job, EventType.RUN_STATE_CHANGED, prevStatus,
            newStatus);
      updateJobInProgressListeners(event);
    }
  }
1040245
public void addJobEntry(BackupJob job)
   {
      // jobEntries
      try
      {
         JobEntryInfo info = new JobEntryInfo();
         info.setDate(Calendar.getInstance());
         info.setType(job.getType());
         info.setState(job.getState());
         info.setURL(job.getStorageURL());

         logWriter.write(info, config);
      }
      catch (IOException e)
      {
         logger.error("Can't add job", e);
      }
      catch (XMLStreamException e)
      {
         logger.error("Can't add job", e);
      }
      catch (BackupOperationException e)
      {
         logger.error("Can't add job", e);
      }
   }
24018
public void initTasks() throws IOException {
    // log job info
    jobHistory.logSubmitted(jobFile.toString(), this.startTime, this.jobTrackerId);
    // log the job priority
    JobClient.RawSplit[] splits = null;
    splits = JobClient.getAndRemoveCachedSplits(jobId);
    if (splits == null) {
      FileSystem fs = jobFile.getFileSystem(jobConf);
      Path splitFile = new Path(jobFile.getParent(), "job.split");
      LOG.info("Reading splits from " + splitFile);
      DataInputStream splitFileIn = fs.open(splitFile);
      try {
        splits = JobClient.readSplitFile(splitFileIn);
      } finally {
        splitFileIn.close();
      }
    }
    initTasksFromSplits(splits);
    jobHistory.logInited(this.launchTime,
                         numMapTasks, numReduceTasks);
  }
752479
public Split<JobXMLDescriptor> getOrCreateSplit()
   {
      List<Node> nodeList = model.get("split");
      if (nodeList != null &&  nodeList.size() > 0)
      {
         return new SplitImpl<JobXMLDescriptor>(this, "split", model, nodeList.get(0));
      }
      return createSplit();
   }
149243
@Override
  public void addJob(final Job job) throws JobExistsException {
    log.info("adding job: {}", job);
    final JobId id = job.getId();
    final UUID operationId = UUID.randomUUID();
    final String creationPath = Paths.configJobCreation(id, operationId);
    final ZooKeeperClient client = provider.get("addJob");
    try {
      try {
        client.ensurePath(Paths.historyJob(id));
        client.transaction(create(Paths.configJob(id), job),
            create(Paths.configJobRefShort(id), id),
            create(Paths.configJobHosts(id)),
            create(creationPath),
            // Touch the jobs root node so that its version is bumped on every job
            // change down the tree. Effectively, make it that version == cVersion.
            set(Paths.configJobs(), UUID.randomUUID().toString().getBytes()));
      } catch (final NodeExistsException e) {
        if (client.exists(creationPath) != null) {
          // The job was created, we're done here
          return;
        }
        throw new JobExistsException(id.toString());
      }
    } catch (NoNodeException e) {
      throw new HeliosRuntimeException("adding job " + job + " failed due to missing ZK path: "
                                       + e.getPath(), e);
    } catch (final KeeperException e) {
      throw new HeliosRuntimeException("adding job " + job + " failed", e);
    }
  }
559175
public List<InputSplit> getSplits(JobContext job
                                    ) throws IOException {
    long minSize = Math.max(getFormatMinSplitSize(), getMinSplitSize(job));
    long maxSize = getMaxSplitSize(job);

    // generate splits
    List<InputSplit> splits = new ArrayList<InputSplit>();
    for (LocatedFileStatus file: listLocatedStatus(job)) {
      Path path = file.getPath();
      long length = file.getLen();
      BlockLocation[] blkLocations = file.getBlockLocations();

      if ((length != 0) && isSplitable(job, path)) { 
        long blockSize = file.getBlockSize();
        long splitSize = computeSplitSize(blockSize, minSize, maxSize);

        long bytesRemaining = length;
        while (((double) bytesRemaining)/splitSize > SPLIT_SLOP) {
          int blkIndex = getBlockIndex(blkLocations, length-bytesRemaining);
          splits.add(new FileSplit(path, length-bytesRemaining, splitSize, 
                                   blkLocations[blkIndex].getHosts()));
          bytesRemaining -= splitSize;
        }
        
        if (bytesRemaining != 0) {
          splits.add(new FileSplit(path, length-bytesRemaining, bytesRemaining, 
                     blkLocations[blkLocations.length-1].getHosts()));
        }
      } else if (length != 0) {
        splits.add(new FileSplit(path, 0, length, blkLocations[0].getHosts()));
      } else { 
        //Create empty hosts array for zero length files
        splits.add(new FileSplit(path, 0, length, new String[0]));
      }
    }
    LOG.debug("Total # of splits: " + splits.size());
    return splits;
  }
716714
public static void addInputPath(Job job, 
                                  Path path) throws IOException {
    Configuration conf = job.getConfiguration();
    FileSystem fs = FileSystem.get(conf);
    path = path.makeQualified(fs);
    String dirStr = StringUtils.escapeString(path.toString());
    String dirs = conf.get("mapred.input.dir");
    conf.set("mapred.input.dir", dirs == null ? dirStr : dirs + "," + dirStr);
  }
1168118
public void configure(JobConf job) {
    this.inputFile = job.get("map.input.file");
    maxNumItems = job.getLong("aggregate.max.num.unique.values",
                              Long.MAX_VALUE);
  }