Java 类org.apache.hadoop.mapreduce.v2.app.commit.CommitterJobAbortEvent 实例源码

项目：hadoop-plus 文件：JobImpl.java

protected JobStateInternal checkJobAfterTaskCompletion(JobImpl job) {
  //check for Job failure
  if (job.failedMapTaskCount*100 > 
    job.allowedMapFailuresPercent*job.numMapTasks ||
    job.failedReduceTaskCount*100 > 
    job.allowedReduceFailuresPercent*job.numReduceTasks) {
    job.setFinishTime();

    String diagnosticMsg = "Job failed as tasks failed. " +
        "failedMaps:" + job.failedMapTaskCount + 
        " failedReduces:" + job.failedReduceTaskCount;
    LOG.info(diagnosticMsg);
    job.addDiagnostic(diagnosticMsg);
    job.eventHandler.handle(new CommitterJobAbortEvent(job.jobId,
        job.jobContext,
        org.apache.hadoop.mapreduce.JobStatus.State.FAILED));
    return JobStateInternal.FAIL_ABORT;
  }

  return job.checkReadyForCommit();
}

项目：hadoop 文件：JobImpl.java

@Override
public void transition(JobImpl job, JobEvent event) {
  job.metrics.endRunningJob(job);
  job.addDiagnostic("Job setup failed : "
      + ((JobSetupFailedEvent) event).getMessage());
  job.eventHandler.handle(new CommitterJobAbortEvent(job.jobId,
      job.jobContext,
      org.apache.hadoop.mapreduce.JobStatus.State.FAILED));
}

项目：hadoop 文件：JobImpl.java

@Override
public JobStateInternal transition(JobImpl job, JobEvent event) {
  if(!job.failWaitTriggerScheduledFuture.isCancelled()) {
    for(Task task: job.tasks.values()) {
      if(!task.isFinished()) {
        return JobStateInternal.FAIL_WAIT;
      }
    }
  }
  //Finished waiting. All tasks finished / were killed
  job.failWaitTriggerScheduledFuture.cancel(false);
  job.eventHandler.handle(new CommitterJobAbortEvent(job.jobId,
    job.jobContext, org.apache.hadoop.mapreduce.JobStatus.State.FAILED));
  return JobStateInternal.FAIL_ABORT;
}

项目：hadoop 文件：JobImpl.java

@Override
public void transition(JobImpl job, JobEvent event) {
  LOG.info("Timeout expired in FAIL_WAIT waiting for tasks to get killed."
    + " Going to fail job anyway");
  job.failWaitTriggerScheduledFuture.cancel(false);
  job.eventHandler.handle(new CommitterJobAbortEvent(job.jobId,
    job.jobContext, org.apache.hadoop.mapreduce.JobStatus.State.FAILED));
}

项目：hadoop 文件：JobImpl.java

@Override
public void transition(JobImpl job, JobEvent event) {
  job.addDiagnostic("Job received Kill in INITED state.");
  job.eventHandler.handle(new CommitterJobAbortEvent(job.jobId,
      job.jobContext,
      org.apache.hadoop.mapreduce.JobStatus.State.KILLED));
}

项目：hadoop 文件：JobImpl.java

@Override
public void transition(JobImpl job, JobEvent event) {
  job.metrics.endRunningJob(job);
  job.addDiagnostic("Job received kill in SETUP state.");
  job.eventHandler.handle(new CommitterJobAbortEvent(job.jobId,
      job.jobContext,
      org.apache.hadoop.mapreduce.JobStatus.State.KILLED));
}

项目：hadoop 文件：JobImpl.java

@Override
public void transition(JobImpl job, JobEvent event) {
  JobCommitFailedEvent jcfe = (JobCommitFailedEvent)event;
  job.addDiagnostic("Job commit failed: " + jcfe.getMessage());
  job.eventHandler.handle(new CommitterJobAbortEvent(job.jobId,
      job.jobContext,
      org.apache.hadoop.mapreduce.JobStatus.State.FAILED));
}

项目：hadoop 文件：JobImpl.java

@Override
public void transition(JobImpl job, JobEvent event) {
  job.setFinishTime();
  job.eventHandler.handle(new CommitterJobAbortEvent(job.jobId,
      job.jobContext,
      org.apache.hadoop.mapreduce.JobStatus.State.KILLED));
}

项目：hadoop 文件：JobImpl.java

@Override
protected JobStateInternal checkJobAfterTaskCompletion(JobImpl job) {
  if (job.completedTaskCount == job.tasks.size()) {
    job.setFinishTime();
    job.eventHandler.handle(new CommitterJobAbortEvent(job.jobId,
        job.jobContext,
        org.apache.hadoop.mapreduce.JobStatus.State.KILLED));
    return JobStateInternal.KILL_ABORT;
  }
  //return the current state, Job not finished yet
  return job.getInternalState();
}