import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType; //导入依赖的package包/类
@Override
public synchronized void jobUpdated(JobChangeEvent event) {
  JobInProgress job = event.getJobInProgress();
  if (event instanceof JobStatusChangeEvent) {
    // Check if the ordering of the job has changed
    // For now priority and start-time can change the job ordering
    JobStatusChangeEvent statusEvent = (JobStatusChangeEvent)event;
    JobSchedulingInfo oldInfo =  
      new JobSchedulingInfo(statusEvent.getOldStatus());
    if (statusEvent.getEventType() == EventType.PRIORITY_CHANGED 
        || statusEvent.getEventType() == EventType.START_TIME_CHANGED) {
      // Make a priority change
      reorderJobs(job, oldInfo);
    } else if (statusEvent.getEventType() == EventType.RUN_STATE_CHANGED) {
      // Check if the job is complete
      int runState = statusEvent.getNewStatus().getRunState();
      if (runState == JobStatus.SUCCEEDED
          || runState == JobStatus.FAILED
          || runState == JobStatus.KILLED) {
        jobCompleted(oldInfo);
      }
    }
  }
}
 

import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType; //导入依赖的package包/类
private synchronized void killJob(JobInProgress job) {
  LOG.info("Killing job " + job.getJobID());
  JobStatus prevStatus = (JobStatus)job.getStatus().clone();
  job.kill();
  
  // Inform the listeners if the job is killed
  // Note : 
  //   If the job is killed in the PREP state then the listeners will be 
  //   invoked
  //   If the job is killed in the RUNNING state then cleanup tasks will be 
  //   launched and the updateTaskStatuses() will take care of it
  JobStatus newStatus = (JobStatus)job.getStatus().clone();
  if (prevStatus.getRunState() != newStatus.getRunState()
      && newStatus.getRunState() == JobStatus.KILLED) {
    JobStatusChangeEvent event = 
      new JobStatusChangeEvent(job, EventType.RUN_STATE_CHANGED, prevStatus, 
          newStatus);
    updateJobInProgressListeners(event);
  }
}
 

import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType; //导入依赖的package包/类
/**
 * Fail a job and inform the listeners. Other components in the framework 
 * should use this to fail a job.
 */
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);
  }
}
 

import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType; //导入依赖的package包/类
/**
 * Change the run-time priority of the given job.
 * 
 * @param jobId job id
 * @param priority new {@link JobPriority} for the job
 * @throws IOException
 * @throws AccessControlException
 */
synchronized void setJobPriority(JobID jobId, JobPriority priority)
    throws AccessControlException, IOException {
  JobInProgress job = jobs.get(jobId);
  if (job != null) {

    // check both queue-level and job-level access
    aclsManager.checkAccess(job, UserGroupInformation.getCurrentUser(),
        Operation.SET_JOB_PRIORITY);

    synchronized (taskScheduler) {
      JobStatus oldStatus = (JobStatus)job.getStatus().clone();
      job.setPriority(priority);
      JobStatus newStatus = (JobStatus)job.getStatus().clone();
      JobStatusChangeEvent event = 
        new JobStatusChangeEvent(job, EventType.PRIORITY_CHANGED, oldStatus, 
                                 newStatus);
      updateJobInProgressListeners(event);
    }
  } else {
    LOG.warn("Trying to change the priority of an unknown job: " + jobId);
  }
}
 

import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType; //导入依赖的package包/类
public void jobUpdated(JobChangeEvent event) {
  LOG.info("Job " + event.getJobInProgress().getJobID().toString() + " updated");
  // remove the job is the event is for a completed job
  if (event instanceof JobStatusChangeEvent) {
    JobStatusChangeEvent statusEvent = (JobStatusChangeEvent)event;
    if (statusEvent.getEventType() == EventType.RUN_STATE_CHANGED) {
      // check if the state changes from 
      // RUNNING->COMPLETE(SUCCESS/KILLED/FAILED)
      JobInProgress jip = event.getJobInProgress();
      String jobId = jip.getJobID().toString();
      if (jip.isComplete()) {
        LOG.info("Job " +  jobId + " deleted from the running queue");
        if (statusEvent.getOldStatus().getRunState() == JobStatus.PREP) {
          wjobs.remove(jip);
        } else {
          jobs.remove(jip);
        }
      } else {
        // PREP->RUNNING
        LOG.info("Job " +  jobId + " deleted from the waiting queue");
        wjobs.remove(jip);
        jobs.add(jip);
      }
    }
  }
}
 

import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType; //导入依赖的package包/类
public void initJob(JobInProgress job) {
  try {
    JobStatus prevStatus = (JobStatus)job.getStatus().clone();
    job.initTasks();
    JobStatus newStatus = (JobStatus)job.getStatus().clone();
    if (prevStatus.getRunState() != newStatus.getRunState()) {
      JobStatusChangeEvent event = 
        new JobStatusChangeEvent(job, EventType.RUN_STATE_CHANGED, prevStatus, 
            newStatus);
      for (JobInProgressListener listener : listeners) {
        listener.jobUpdated(event);
      }
    }
  } catch (Exception ioe) {
    failJob(job);
  }
}
 

import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType; //导入依赖的package包/类
public void setStartTime(FakeJobInProgress fjob, long start) {
  // take a snapshot of the status before changing it
  JobStatus oldStatus = (JobStatus)fjob.getStatus().clone();
  
  fjob.startTime = start; // change the start time of the job
  fjob.status.setStartTime(start); // change the start time of the jobstatus
  
  JobStatus newStatus = (JobStatus)fjob.getStatus().clone();
  
  JobStatusChangeEvent event = 
    new JobStatusChangeEvent (fjob, EventType.START_TIME_CHANGED, oldStatus,
                              newStatus);
  for (JobInProgressListener listener : mylisteners) {
    listener.jobUpdated(event);
  }
}
 

import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType; //导入依赖的package包/类
private synchronized void killJob(JobInProgress job) {
  LOG.info("Killing job " + job.getJobID());
  JobStatus prevStatus = (JobStatus)job.getStatus().clone();
  job.kill();

  // Inform the listeners if the job is killed
  // Note :
  //   If the job is killed in the PREP state then the listeners will be
  //   invoked
  //   If the job is killed in the RUNNING state then cleanup tasks will be
  //   launched and the updateTaskStatuses() will take care of it
  JobStatus newStatus = (JobStatus)job.getStatus().clone();
  if (prevStatus.getRunState() != newStatus.getRunState()
      && newStatus.getRunState() == JobStatus.KILLED) {
    JobStatusChangeEvent event =
      new JobStatusChangeEvent(job, EventType.RUN_STATE_CHANGED, prevStatus,
          newStatus);
    updateJobInProgressListeners(event);
  }
}
 

import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType; //导入依赖的package包/类
/**
 * Fail a job and inform the listeners. Other components in the framework
 * should use this to fail a job.
 */
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);
  }
}
 

import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType; //导入依赖的package包/类
/**
 * Change the run-time priority of the given job.
 * @param jobId job id
 * @param priority new {@link JobPriority} for the job
 */
synchronized void setJobPriority(JobID jobId, JobPriority priority) {
  JobInProgress job = jobs.get(jobId);
  if (job != null) {
    synchronized (taskScheduler) {
      JobStatus oldStatus = (JobStatus)job.getStatus().clone();
      job.setPriority(priority);
      JobStatus newStatus = (JobStatus)job.getStatus().clone();
      JobStatusChangeEvent event =
        new JobStatusChangeEvent(job, EventType.PRIORITY_CHANGED, oldStatus,
                                 newStatus);
      updateJobInProgressListeners(event);
    }
  } else {
    LOG.warn("Trying to change the priority of an unknown job: " + jobId);
  }
}
 

import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType; //导入依赖的package包/类
private void jobStateChanged(JobStatusChangeEvent event, QueueInfo qi) {
  JobInProgress job = event.getJobInProgress();
  JobSchedulingInfo oldJobStateInfo = 
    new JobSchedulingInfo(event.getOldStatus());
  // Check if the ordering of the job has changed
  // For now priority and start-time can change the job ordering
  if (event.getEventType() == EventType.PRIORITY_CHANGED 
      || event.getEventType() == EventType.START_TIME_CHANGED) {
    // Make a priority change
    reorderJobs(job, oldJobStateInfo, qi);
  } else if (event.getEventType() == EventType.RUN_STATE_CHANGED) {
    // Check if the job is complete
    int runState = job.getStatus().getRunState();
    if (runState == JobStatus.SUCCEEDED
        || runState == JobStatus.FAILED
        || runState == JobStatus.KILLED) {
      jobCompleted(job, oldJobStateInfo, qi);
    } else if (runState == JobStatus.RUNNING) {
      makeJobRunning(job, oldJobStateInfo, qi);
    }
  }
}
 

import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType; //导入依赖的package包/类
public void setStartTime(FakeJobInProgress fjob, long start) {
  // take a snapshot of the status before changing it
  JobStatus oldStatus = (JobStatus)fjob.getStatus().clone();
  
  fjob.startTime = start; // change the start time of the job
  fjob.status.setStartTime(start); // change the start time of the jobstatus
  
  JobStatus newStatus = (JobStatus)fjob.getStatus().clone();
  
  JobStatusChangeEvent event = 
    new JobStatusChangeEvent (fjob, EventType.START_TIME_CHANGED, oldStatus,
                              newStatus);
  for (JobInProgressListener listener : listeners) {
    listener.jobUpdated(event);
  }
}
 

import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType; //导入依赖的package包/类
private JobStatusChangeEvent updateJob(JobInProgress jip, 
                                       JobHistory.JobInfo job) {
  // Change the job priority
  String jobpriority = job.get(Keys.JOB_PRIORITY);
  JobPriority priority = JobPriority.valueOf(jobpriority);
  // It's important to update this via the jobtracker's api as it will 
  // take care of updating the event listeners too
  setJobPriority(jip.getJobID(), priority);

  // Save the previous job status
  JobStatus oldStatus = (JobStatus)jip.getStatus().clone();
  
  // Set the start/launch time only if there are recovered tasks
  // Increment the job's restart count
  jip.updateJobInfo(job.getLong(JobHistory.Keys.SUBMIT_TIME), 
                    job.getLong(JobHistory.Keys.LAUNCH_TIME));

  // Save the new job status
  JobStatus newStatus = (JobStatus)jip.getStatus().clone();
  
  return new JobStatusChangeEvent(jip, EventType.START_TIME_CHANGED, oldStatus, 
                                  newStatus);
}
 

import org.apache.hadoop.mapred.JobStatusChangeEvent.EventType; //导入依赖的package包/类
/**
 * Change the run-time priority of the given job.
 * @param jobId job id
 * @param priority new {@link JobPriority} for the job
 */
synchronized void setJobPriority(JobID jobId, JobPriority priority) {
  JobInProgress job = jobs.get(jobId);
  if (job != null) {
    synchronized (taskScheduler) {
      JobStatus oldStatus = (JobStatus)job.getStatus().clone();
      job.setPriority(priority);
      JobStatus newStatus = (JobStatus)job.getStatus().clone();
      JobStatusChangeEvent event = 
        new JobStatusChangeEvent(job, EventType.PRIORITY_CHANGED, oldStatus, 
                                 newStatus);
      updateJobInProgressListeners(event);
    }
  } else {
    LOG.warn("Trying to change the priority of an unknown job: " + jobId);
  }
}