You are viewing a plain text version of this content. The canonical link for it is here.
Posted to mapreduce-commits@hadoop.apache.org by sz...@apache.org on 2013/01/14 04:44:52 UTC
svn commit: r1432796 [2/5] - in
/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project: ./ conf/
hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapred/
hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/mai...
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java?rev=1432796&r1=1432795&r2=1432796&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java Mon Jan 14 03:44:35 2013
@@ -47,7 +47,6 @@ import org.apache.hadoop.mapreduce.Count
import org.apache.hadoop.mapreduce.JobACL;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.MRJobConfig;
-import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.TypeConverter;
import org.apache.hadoop.mapreduce.jobhistory.JobFinishedEvent;
import org.apache.hadoop.mapreduce.jobhistory.JobHistoryEvent;
@@ -77,14 +76,20 @@ import org.apache.hadoop.mapreduce.v2.ap
import org.apache.hadoop.mapreduce.v2.api.records.TaskType;
import org.apache.hadoop.mapreduce.v2.app.AppContext;
import org.apache.hadoop.mapreduce.v2.app.TaskAttemptListener;
+import org.apache.hadoop.mapreduce.v2.app.commit.CommitterJobAbortEvent;
+import org.apache.hadoop.mapreduce.v2.app.commit.CommitterJobCommitEvent;
+import org.apache.hadoop.mapreduce.v2.app.commit.CommitterJobSetupEvent;
import org.apache.hadoop.mapreduce.v2.app.job.JobStateInternal;
import org.apache.hadoop.mapreduce.v2.app.job.Task;
import org.apache.hadoop.mapreduce.v2.app.job.TaskAttempt;
+import org.apache.hadoop.mapreduce.v2.app.job.event.JobAbortCompletedEvent;
+import org.apache.hadoop.mapreduce.v2.app.job.event.JobCommitFailedEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobCounterUpdateEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobDiagnosticsUpdateEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobEventType;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobFinishEvent;
+import org.apache.hadoop.mapreduce.v2.app.job.event.JobSetupFailedEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobTaskAttemptCompletedEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobTaskAttemptFetchFailureEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobTaskEvent;
@@ -138,7 +143,6 @@ public class JobImpl implements org.apac
private final Clock clock;
private final JobACLsManager aclsManager;
private final String username;
- private final OutputCommitter committer;
private final Map<JobACL, AccessControlList> jobACLs;
private float setupWeight = 0.05f;
private float cleanupWeight = 0.05f;
@@ -176,6 +180,7 @@ public class JobImpl implements org.apac
private Counters fullCounters = null;
private Counters finalMapCounters = null;
private Counters finalReduceCounters = null;
+
// FIXME:
//
// Can then replace task-level uber counters (MR-2424) with job-level ones
@@ -245,7 +250,7 @@ public class JobImpl implements org.apac
DIAGNOSTIC_UPDATE_TRANSITION)
.addTransition(JobStateInternal.INITED, JobStateInternal.INITED,
JobEventType.JOB_COUNTER_UPDATE, COUNTER_UPDATE_TRANSITION)
- .addTransition(JobStateInternal.INITED, JobStateInternal.RUNNING,
+ .addTransition(JobStateInternal.INITED, JobStateInternal.SETUP,
JobEventType.JOB_START,
new StartTransition())
.addTransition(JobStateInternal.INITED, JobStateInternal.KILLED,
@@ -257,19 +262,43 @@ public class JobImpl implements org.apac
// Ignore-able events
.addTransition(JobStateInternal.INITED, JobStateInternal.INITED,
JobEventType.JOB_UPDATED_NODES)
-
+
+ // Transitions from SETUP state
+ .addTransition(JobStateInternal.SETUP, JobStateInternal.SETUP,
+ JobEventType.JOB_DIAGNOSTIC_UPDATE,
+ DIAGNOSTIC_UPDATE_TRANSITION)
+ .addTransition(JobStateInternal.SETUP, JobStateInternal.SETUP,
+ JobEventType.JOB_COUNTER_UPDATE, COUNTER_UPDATE_TRANSITION)
+ .addTransition(JobStateInternal.SETUP, JobStateInternal.RUNNING,
+ JobEventType.JOB_SETUP_COMPLETED,
+ new SetupCompletedTransition())
+ .addTransition(JobStateInternal.SETUP, JobStateInternal.FAIL_ABORT,
+ JobEventType.JOB_SETUP_FAILED,
+ new SetupFailedTransition())
+ .addTransition(JobStateInternal.SETUP, JobStateInternal.KILL_ABORT,
+ JobEventType.JOB_KILL,
+ new KilledDuringSetupTransition())
+ .addTransition(JobStateInternal.SETUP, JobStateInternal.ERROR,
+ JobEventType.INTERNAL_ERROR,
+ INTERNAL_ERROR_TRANSITION)
+ // Ignore-able events
+ .addTransition(JobStateInternal.SETUP, JobStateInternal.SETUP,
+ JobEventType.JOB_UPDATED_NODES)
+
// Transitions from RUNNING state
.addTransition(JobStateInternal.RUNNING, JobStateInternal.RUNNING,
JobEventType.JOB_TASK_ATTEMPT_COMPLETED,
TASK_ATTEMPT_COMPLETED_EVENT_TRANSITION)
.addTransition
(JobStateInternal.RUNNING,
- EnumSet.of(JobStateInternal.RUNNING, JobStateInternal.SUCCEEDED, JobStateInternal.FAILED),
+ EnumSet.of(JobStateInternal.RUNNING,
+ JobStateInternal.COMMITTING, JobStateInternal.FAIL_ABORT),
JobEventType.JOB_TASK_COMPLETED,
new TaskCompletedTransition())
.addTransition
(JobStateInternal.RUNNING,
- EnumSet.of(JobStateInternal.RUNNING, JobStateInternal.SUCCEEDED, JobStateInternal.FAILED),
+ EnumSet.of(JobStateInternal.RUNNING,
+ JobStateInternal.COMMITTING),
JobEventType.JOB_COMPLETED,
new JobNoTasksCompletedTransition())
.addTransition(JobStateInternal.RUNNING, JobStateInternal.KILL_WAIT,
@@ -296,7 +325,8 @@ public class JobImpl implements org.apac
// Transitions from KILL_WAIT state.
.addTransition
(JobStateInternal.KILL_WAIT,
- EnumSet.of(JobStateInternal.KILL_WAIT, JobStateInternal.KILLED),
+ EnumSet.of(JobStateInternal.KILL_WAIT,
+ JobStateInternal.KILL_ABORT),
JobEventType.JOB_TASK_COMPLETED,
new KillWaitTaskCompletedTransition())
.addTransition(JobStateInternal.KILL_WAIT, JobStateInternal.KILL_WAIT,
@@ -318,6 +348,35 @@ public class JobImpl implements org.apac
JobEventType.JOB_MAP_TASK_RESCHEDULED,
JobEventType.JOB_TASK_ATTEMPT_FETCH_FAILURE))
+ // Transitions from COMMITTING state
+ .addTransition(JobStateInternal.COMMITTING,
+ JobStateInternal.SUCCEEDED,
+ JobEventType.JOB_COMMIT_COMPLETED,
+ new CommitSucceededTransition())
+ .addTransition(JobStateInternal.COMMITTING,
+ JobStateInternal.FAIL_ABORT,
+ JobEventType.JOB_COMMIT_FAILED,
+ new CommitFailedTransition())
+ .addTransition(JobStateInternal.COMMITTING,
+ JobStateInternal.KILL_ABORT,
+ JobEventType.JOB_KILL,
+ new KilledDuringCommitTransition())
+ .addTransition(JobStateInternal.COMMITTING,
+ JobStateInternal.COMMITTING,
+ JobEventType.JOB_DIAGNOSTIC_UPDATE,
+ DIAGNOSTIC_UPDATE_TRANSITION)
+ .addTransition(JobStateInternal.COMMITTING,
+ JobStateInternal.COMMITTING,
+ JobEventType.JOB_COUNTER_UPDATE, COUNTER_UPDATE_TRANSITION)
+ .addTransition(JobStateInternal.COMMITTING,
+ JobStateInternal.ERROR, JobEventType.INTERNAL_ERROR,
+ INTERNAL_ERROR_TRANSITION)
+ // Ignore-able events
+ .addTransition(JobStateInternal.COMMITTING,
+ JobStateInternal.COMMITTING,
+ EnumSet.of(JobEventType.JOB_UPDATED_NODES,
+ JobEventType.JOB_TASK_ATTEMPT_FETCH_FAILURE))
+
// Transitions from SUCCEEDED state
.addTransition(JobStateInternal.SUCCEEDED, JobStateInternal.SUCCEEDED,
JobEventType.JOB_DIAGNOSTIC_UPDATE,
@@ -334,6 +393,61 @@ public class JobImpl implements org.apac
JobEventType.JOB_UPDATED_NODES,
JobEventType.JOB_TASK_ATTEMPT_FETCH_FAILURE))
+ // Transitions from FAIL_ABORT state
+ .addTransition(JobStateInternal.FAIL_ABORT,
+ JobStateInternal.FAIL_ABORT,
+ JobEventType.JOB_DIAGNOSTIC_UPDATE,
+ DIAGNOSTIC_UPDATE_TRANSITION)
+ .addTransition(JobStateInternal.FAIL_ABORT,
+ JobStateInternal.FAIL_ABORT,
+ JobEventType.JOB_COUNTER_UPDATE, COUNTER_UPDATE_TRANSITION)
+ .addTransition(JobStateInternal.FAIL_ABORT, JobStateInternal.FAILED,
+ JobEventType.JOB_ABORT_COMPLETED,
+ new JobAbortCompletedTransition())
+ .addTransition(JobStateInternal.FAIL_ABORT, JobStateInternal.KILLED,
+ JobEventType.JOB_KILL,
+ new KilledDuringAbortTransition())
+ .addTransition(JobStateInternal.FAIL_ABORT,
+ JobStateInternal.ERROR, JobEventType.INTERNAL_ERROR,
+ INTERNAL_ERROR_TRANSITION)
+ // Ignore-able events
+ .addTransition(JobStateInternal.FAIL_ABORT,
+ JobStateInternal.FAIL_ABORT,
+ EnumSet.of(JobEventType.JOB_UPDATED_NODES,
+ JobEventType.JOB_TASK_COMPLETED,
+ JobEventType.JOB_TASK_ATTEMPT_COMPLETED,
+ JobEventType.JOB_MAP_TASK_RESCHEDULED,
+ JobEventType.JOB_TASK_ATTEMPT_FETCH_FAILURE,
+ JobEventType.JOB_COMMIT_COMPLETED,
+ JobEventType.JOB_COMMIT_FAILED))
+
+ // Transitions from KILL_ABORT state
+ .addTransition(JobStateInternal.KILL_ABORT,
+ JobStateInternal.KILL_ABORT,
+ JobEventType.JOB_DIAGNOSTIC_UPDATE,
+ DIAGNOSTIC_UPDATE_TRANSITION)
+ .addTransition(JobStateInternal.KILL_ABORT,
+ JobStateInternal.KILL_ABORT,
+ JobEventType.JOB_COUNTER_UPDATE, COUNTER_UPDATE_TRANSITION)
+ .addTransition(JobStateInternal.KILL_ABORT, JobStateInternal.KILLED,
+ JobEventType.JOB_ABORT_COMPLETED,
+ new JobAbortCompletedTransition())
+ .addTransition(JobStateInternal.KILL_ABORT, JobStateInternal.KILLED,
+ JobEventType.JOB_KILL,
+ new KilledDuringAbortTransition())
+ .addTransition(JobStateInternal.KILL_ABORT,
+ JobStateInternal.ERROR, JobEventType.INTERNAL_ERROR,
+ INTERNAL_ERROR_TRANSITION)
+ // Ignore-able events
+ .addTransition(JobStateInternal.KILL_ABORT,
+ JobStateInternal.KILL_ABORT,
+ EnumSet.of(JobEventType.JOB_UPDATED_NODES,
+ JobEventType.JOB_TASK_ATTEMPT_FETCH_FAILURE,
+ JobEventType.JOB_SETUP_COMPLETED,
+ JobEventType.JOB_SETUP_FAILED,
+ JobEventType.JOB_COMMIT_COMPLETED,
+ JobEventType.JOB_COMMIT_FAILED))
+
// Transitions from FAILED state
.addTransition(JobStateInternal.FAILED, JobStateInternal.FAILED,
JobEventType.JOB_DIAGNOSTIC_UPDATE,
@@ -351,7 +465,12 @@ public class JobImpl implements org.apac
JobEventType.JOB_TASK_COMPLETED,
JobEventType.JOB_TASK_ATTEMPT_COMPLETED,
JobEventType.JOB_MAP_TASK_RESCHEDULED,
- JobEventType.JOB_TASK_ATTEMPT_FETCH_FAILURE))
+ JobEventType.JOB_TASK_ATTEMPT_FETCH_FAILURE,
+ JobEventType.JOB_SETUP_COMPLETED,
+ JobEventType.JOB_SETUP_FAILED,
+ JobEventType.JOB_COMMIT_COMPLETED,
+ JobEventType.JOB_COMMIT_FAILED,
+ JobEventType.JOB_ABORT_COMPLETED))
// Transitions from KILLED state
.addTransition(JobStateInternal.KILLED, JobStateInternal.KILLED,
@@ -366,8 +485,14 @@ public class JobImpl implements org.apac
// Ignore-able events
.addTransition(JobStateInternal.KILLED, JobStateInternal.KILLED,
EnumSet.of(JobEventType.JOB_KILL,
+ JobEventType.JOB_START,
JobEventType.JOB_UPDATED_NODES,
- JobEventType.JOB_TASK_ATTEMPT_FETCH_FAILURE))
+ JobEventType.JOB_TASK_ATTEMPT_FETCH_FAILURE,
+ JobEventType.JOB_SETUP_COMPLETED,
+ JobEventType.JOB_SETUP_FAILED,
+ JobEventType.JOB_COMMIT_COMPLETED,
+ JobEventType.JOB_COMMIT_FAILED,
+ JobEventType.JOB_ABORT_COMPLETED))
// No transitions from INTERNAL_ERROR state. Ignore all.
.addTransition(
@@ -381,6 +506,11 @@ public class JobImpl implements org.apac
JobEventType.JOB_DIAGNOSTIC_UPDATE,
JobEventType.JOB_UPDATED_NODES,
JobEventType.JOB_TASK_ATTEMPT_FETCH_FAILURE,
+ JobEventType.JOB_SETUP_COMPLETED,
+ JobEventType.JOB_SETUP_FAILED,
+ JobEventType.JOB_COMMIT_COMPLETED,
+ JobEventType.JOB_COMMIT_FAILED,
+ JobEventType.JOB_ABORT_COMPLETED,
JobEventType.INTERNAL_ERROR))
.addTransition(JobStateInternal.ERROR, JobStateInternal.ERROR,
JobEventType.JOB_COUNTER_UPDATE, COUNTER_UPDATE_TRANSITION)
@@ -410,6 +540,8 @@ public class JobImpl implements org.apac
private Credentials fsTokens;
private Token<JobTokenIdentifier> jobToken;
private JobTokenSecretManager jobTokenSecretManager;
+
+ private JobStateInternal forcedState = null;
public JobImpl(JobId jobId, ApplicationAttemptId applicationAttemptId,
Configuration conf, EventHandler eventHandler,
@@ -417,8 +549,9 @@ public class JobImpl implements org.apac
JobTokenSecretManager jobTokenSecretManager,
Credentials fsTokenCredentials, Clock clock,
Map<TaskId, TaskInfo> completedTasksFromPreviousRun, MRAppMetrics metrics,
- OutputCommitter committer, boolean newApiCommitter, String userName,
- long appSubmitTime, List<AMInfo> amInfos, AppContext appContext) {
+ boolean newApiCommitter, String userName,
+ long appSubmitTime, List<AMInfo> amInfos, AppContext appContext,
+ JobStateInternal forcedState, String forcedDiagnostic) {
this.applicationAttemptId = applicationAttemptId;
this.jobId = jobId;
this.jobName = conf.get(JobContext.JOB_NAME, "<missing job name>");
@@ -442,7 +575,6 @@ public class JobImpl implements org.apac
this.fsTokens = fsTokenCredentials;
this.jobTokenSecretManager = jobTokenSecretManager;
- this.committer = committer;
this.aclsManager = new JobACLsManager(conf);
this.username = System.getProperty("user.name");
@@ -450,6 +582,10 @@ public class JobImpl implements org.apac
// This "this leak" is okay because the retained pointer is in an
// instance variable.
stateMachine = stateMachineFactory.make(this);
+ this.forcedState = forcedState;
+ if(forcedDiagnostic != null) {
+ this.diagnostics.add(forcedDiagnostic);
+ }
}
protected StateMachine<JobStateInternal, JobEventType, JobEvent> getStateMachine() {
@@ -461,11 +597,6 @@ public class JobImpl implements org.apac
return jobId;
}
- // Getter methods that make unit testing easier (package-scoped)
- OutputCommitter getCommitter() {
- return this.committer;
- }
-
EventHandler getEventHandler() {
return this.eventHandler;
}
@@ -694,7 +825,7 @@ public class JobImpl implements org.apac
public JobState getState() {
readLock.lock();
try {
- return getExternalState(getStateMachine().getCurrentState());
+ return getExternalState(getInternalState());
} finally {
readLock.unlock();
}
@@ -744,6 +875,9 @@ public class JobImpl implements org.apac
public JobStateInternal getInternalState() {
readLock.lock();
try {
+ if(forcedState != null) {
+ return forcedState;
+ }
return getStateMachine().getCurrentState();
} finally {
readLock.unlock();
@@ -751,9 +885,16 @@ public class JobImpl implements org.apac
}
private static JobState getExternalState(JobStateInternal smState) {
- if (smState == JobStateInternal.KILL_WAIT) {
+ switch (smState) {
+ case KILL_WAIT:
+ case KILL_ABORT:
return JobState.KILLED;
- } else {
+ case SETUP:
+ case COMMITTING:
+ return JobState.RUNNING;
+ case FAIL_ABORT:
+ return JobState.FAILED;
+ default:
return JobState.valueOf(smState.name());
}
}
@@ -799,22 +940,15 @@ public class JobImpl implements org.apac
return FileSystem.get(conf);
}
- static JobStateInternal checkJobCompleteSuccess(JobImpl job) {
- // check for Job success
- if (job.completedTaskCount == job.tasks.size()) {
- try {
- // Commit job & do cleanup
- job.getCommitter().commitJob(job.getJobContext());
- } catch (IOException e) {
- LOG.error("Could not do commit for Job", e);
- job.addDiagnostic("Job commit failed: " + e.getMessage());
- job.abortJob(org.apache.hadoop.mapreduce.JobStatus.State.FAILED);
- return job.finished(JobStateInternal.FAILED);
- }
- job.logJobHistoryFinishedEvent();
- return job.finished(JobStateInternal.SUCCEEDED);
+ protected JobStateInternal checkReadyForCommit() {
+ JobStateInternal currentState = getInternalState();
+ if (completedTaskCount == tasks.size()
+ && currentState == JobStateInternal.RUNNING) {
+ eventHandler.handle(new CommitterJobCommitEvent(jobId, getJobContext()));
+ return JobStateInternal.COMMITTING;
}
- return null;
+ // return the current state as job not ready to commit yet
+ return getInternalState();
}
JobStateInternal finished(JobStateInternal finalState) {
@@ -916,6 +1050,10 @@ public class JobImpl implements org.apac
conf.getInt(MRJobConfig.MR_AM_VMEM_MB,
MRJobConfig.DEFAULT_MR_AM_VMEM_MB);
+ long sysCPUSizeForUberSlot =
+ conf.getInt(MRJobConfig.MR_AM_CPU_VCORES,
+ MRJobConfig.DEFAULT_MR_AM_CPU_VCORES);
+
boolean uberEnabled =
conf.getBoolean(MRJobConfig.JOB_UBERTASK_ENABLE, false);
boolean smallNumMapTasks = (numMapTasks <= sysMaxMaps);
@@ -927,6 +1065,13 @@ public class JobImpl implements org.apac
conf.getLong(MRJobConfig.REDUCE_MEMORY_MB, 0))
<= sysMemSizeForUberSlot)
|| (sysMemSizeForUberSlot == JobConf.DISABLED_MEMORY_LIMIT));
+ boolean smallCpu =
+ (
+ Math.max(
+ conf.getInt(MRJobConfig.MAP_CPU_VCORES, 1),
+ conf.getInt(MRJobConfig.REDUCE_CPU_VCORES, 1)) <
+ sysCPUSizeForUberSlot
+ );
boolean notChainJob = !isChainJob(conf);
// User has overall veto power over uberization, or user can modify
@@ -937,7 +1082,8 @@ public class JobImpl implements org.apac
// while "uber-AM" (MR AM + LocalContainerLauncher) loops over tasks
// and thus requires sequential execution.
isUber = uberEnabled && smallNumMapTasks && smallNumReduceTasks
- && smallInput && smallMemory && notChainJob && isValidUberMaxReduces;
+ && smallInput && smallMemory && smallCpu
+ && notChainJob && isValidUberMaxReduces;
if (isUber) {
LOG.info("Uberizing job " + jobId + ": " + numMapTasks + "m+"
@@ -1104,25 +1250,21 @@ public class JobImpl implements org.apac
job.allowedReduceFailuresPercent =
job.conf.getInt(MRJobConfig.REDUCE_FAILURES_MAXPERCENT, 0);
- // do the setup
- job.committer.setupJob(job.jobContext);
- job.setupProgress = 1.0f;
-
// create the Tasks but don't start them yet
createMapTasks(job, inputLength, taskSplitMetaInfo);
createReduceTasks(job);
job.metrics.endPreparingJob(job);
return JobStateInternal.INITED;
- //TODO XXX Should JobInitedEvent be generated here (instead of in StartTransition)
-
} catch (IOException e) {
LOG.warn("Job init failed", e);
+ job.metrics.endPreparingJob(job);
job.addDiagnostic("Job init failed : "
+ StringUtils.stringifyException(e));
- job.abortJob(org.apache.hadoop.mapreduce.JobStatus.State.FAILED);
- job.metrics.endPreparingJob(job);
- return job.finished(JobStateInternal.FAILED);
+ job.eventHandler.handle(new CommitterJobAbortEvent(job.jobId,
+ job.jobContext,
+ org.apache.hadoop.mapreduce.JobStatus.State.FAILED));
+ return JobStateInternal.FAILED;
}
}
@@ -1174,7 +1316,7 @@ public class JobImpl implements org.apac
job.remoteJobConfFile,
job.conf, splits[i],
job.taskAttemptListener,
- job.committer, job.jobToken, job.fsTokens,
+ job.jobToken, job.fsTokens,
job.clock, job.completedTasksFromPreviousRun,
job.applicationAttemptId.getAttemptId(),
job.metrics, job.appContext);
@@ -1191,7 +1333,7 @@ public class JobImpl implements org.apac
job.eventHandler,
job.remoteJobConfFile,
job.conf, job.numMapTasks,
- job.taskAttemptListener, job.committer, job.jobToken,
+ job.taskAttemptListener, job.jobToken,
job.fsTokens, job.clock,
job.completedTasksFromPreviousRun,
job.applicationAttemptId.getAttemptId(),
@@ -1224,6 +1366,35 @@ public class JobImpl implements org.apac
}
} // end of InitTransition
+ private static class SetupCompletedTransition
+ implements SingleArcTransition<JobImpl, JobEvent> {
+ @Override
+ public void transition(JobImpl job, JobEvent event) {
+ job.setupProgress = 1.0f;
+ job.scheduleTasks(job.mapTasks); // schedule (i.e., start) the maps
+ job.scheduleTasks(job.reduceTasks);
+
+ // If we have no tasks, just transition to job completed
+ if (job.numReduceTasks == 0 && job.numMapTasks == 0) {
+ job.eventHandler.handle(new JobEvent(job.jobId,
+ JobEventType.JOB_COMPLETED));
+ }
+ }
+ }
+
+ private static class SetupFailedTransition
+ implements SingleArcTransition<JobImpl, JobEvent> {
+ @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));
+ }
+ }
+
public static class StartTransition
implements SingleArcTransition<JobImpl, JobEvent> {
/**
@@ -1233,43 +1404,45 @@ public class JobImpl implements org.apac
@Override
public void transition(JobImpl job, JobEvent event) {
job.startTime = job.clock.getTime();
- job.scheduleTasks(job.mapTasks); // schedule (i.e., start) the maps
- job.scheduleTasks(job.reduceTasks);
JobInitedEvent jie =
new JobInitedEvent(job.oldJobId,
job.startTime,
job.numMapTasks, job.numReduceTasks,
job.getState().toString(),
- job.isUber()); //Will transition to state running. Currently in INITED
+ job.isUber());
job.eventHandler.handle(new JobHistoryEvent(job.jobId, jie));
JobInfoChangeEvent jice = new JobInfoChangeEvent(job.oldJobId,
job.appSubmitTime, job.startTime);
job.eventHandler.handle(new JobHistoryEvent(job.jobId, jice));
job.metrics.runningJob(job);
- // If we have no tasks, just transition to job completed
- if (job.numReduceTasks == 0 && job.numMapTasks == 0) {
- job.eventHandler.handle(new JobEvent(job.jobId, JobEventType.JOB_COMPLETED));
- }
+ job.eventHandler.handle(new CommitterJobSetupEvent(
+ job.jobId, job.jobContext));
}
}
- protected void abortJob(
- org.apache.hadoop.mapreduce.JobStatus.State finalState) {
- try {
- committer.abortJob(jobContext, finalState);
- } catch (IOException e) {
- LOG.warn("Could not abortJob", e);
+ private void unsuccessfulFinish(JobStateInternal finalState) {
+ if (finishTime == 0) setFinishTime();
+ cleanupProgress = 1.0f;
+ JobUnsuccessfulCompletionEvent unsuccessfulJobEvent =
+ new JobUnsuccessfulCompletionEvent(oldJobId,
+ finishTime,
+ succeededMapTaskCount,
+ succeededReduceTaskCount,
+ finalState.toString());
+ eventHandler.handle(new JobHistoryEvent(jobId,
+ unsuccessfulJobEvent));
+ finished(finalState);
+ }
+
+ private static class JobAbortCompletedTransition
+ implements SingleArcTransition<JobImpl, JobEvent> {
+ @Override
+ public void transition(JobImpl job, JobEvent event) {
+ JobStateInternal finalState = JobStateInternal.valueOf(
+ ((JobAbortCompletedEvent) event).getFinalState().name());
+ job.unsuccessfulFinish(finalState);
}
- if (finishTime == 0) setFinishTime();
- cleanupProgress = 1.0f;
- JobUnsuccessfulCompletionEvent unsuccessfulJobEvent =
- new JobUnsuccessfulCompletionEvent(oldJobId,
- finishTime,
- succeededMapTaskCount,
- succeededReduceTaskCount,
- finalState.toString());
- eventHandler.handle(new JobHistoryEvent(jobId, unsuccessfulJobEvent));
}
// JobFinishedEvent triggers the move of the history file out of the staging
@@ -1343,9 +1516,22 @@ public class JobImpl implements org.apac
implements SingleArcTransition<JobImpl, JobEvent> {
@Override
public void transition(JobImpl job, JobEvent event) {
- job.abortJob(org.apache.hadoop.mapreduce.JobStatus.State.KILLED);
job.addDiagnostic("Job received Kill in INITED state.");
- job.finished(JobStateInternal.KILLED);
+ job.eventHandler.handle(new CommitterJobAbortEvent(job.jobId,
+ job.jobContext,
+ org.apache.hadoop.mapreduce.JobStatus.State.KILLED));
+ }
+ }
+
+ private static class KilledDuringSetupTransition
+ implements SingleArcTransition<JobImpl, JobEvent> {
+ @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));
}
}
@@ -1470,10 +1656,10 @@ public class JobImpl implements org.apac
taskKilled(job, task);
}
- return checkJobForCompletion(job);
+ return checkJobAfterTaskCompletion(job);
}
- protected JobStateInternal checkJobForCompletion(JobImpl job) {
+ protected JobStateInternal checkJobAfterTaskCompletion(JobImpl job) {
//check for Job failure
if (job.failedMapTaskCount*100 >
job.allowedMapFailuresPercent*job.numMapTasks ||
@@ -1486,17 +1672,13 @@ public class JobImpl implements org.apac
" failedReduces:" + job.failedReduceTaskCount;
LOG.info(diagnosticMsg);
job.addDiagnostic(diagnosticMsg);
- job.abortJob(org.apache.hadoop.mapreduce.JobStatus.State.FAILED);
- return job.finished(JobStateInternal.FAILED);
+ job.eventHandler.handle(new CommitterJobAbortEvent(job.jobId,
+ job.jobContext,
+ org.apache.hadoop.mapreduce.JobStatus.State.FAILED));
+ return JobStateInternal.FAIL_ABORT;
}
- JobStateInternal jobCompleteSuccess = JobImpl.checkJobCompleteSuccess(job);
- if (jobCompleteSuccess != null) {
- return jobCompleteSuccess;
- }
-
- //return the current state, Job not finished yet
- return job.getInternalState();
+ return job.checkReadyForCommit();
}
private void taskSucceeded(JobImpl job, Task task) {
@@ -1529,18 +1711,52 @@ public class JobImpl implements org.apac
}
// Transition class for handling jobs with no tasks
- static class JobNoTasksCompletedTransition implements
+ private static class JobNoTasksCompletedTransition implements
MultipleArcTransition<JobImpl, JobEvent, JobStateInternal> {
@Override
public JobStateInternal transition(JobImpl job, JobEvent event) {
- JobStateInternal jobCompleteSuccess = JobImpl.checkJobCompleteSuccess(job);
- if (jobCompleteSuccess != null) {
- return jobCompleteSuccess;
- }
-
- // Return the current state, Job not finished yet
- return job.getInternalState();
+ return job.checkReadyForCommit();
+ }
+ }
+
+ private static class CommitSucceededTransition implements
+ SingleArcTransition<JobImpl, JobEvent> {
+ @Override
+ public void transition(JobImpl job, JobEvent event) {
+ job.logJobHistoryFinishedEvent();
+ job.finished(JobStateInternal.SUCCEEDED);
+ }
+ }
+
+ private static class CommitFailedTransition implements
+ SingleArcTransition<JobImpl, JobEvent> {
+ @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));
+ }
+ }
+
+ private static class KilledDuringCommitTransition implements
+ SingleArcTransition<JobImpl, JobEvent> {
+ @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));
+ }
+ }
+
+ private static class KilledDuringAbortTransition implements
+ SingleArcTransition<JobImpl, JobEvent> {
+ @Override
+ public void transition(JobImpl job, JobEvent event) {
+ job.unsuccessfulFinish(JobStateInternal.KILLED);
}
}
@@ -1557,11 +1773,13 @@ public class JobImpl implements org.apac
private static class KillWaitTaskCompletedTransition extends
TaskCompletedTransition {
@Override
- protected JobStateInternal checkJobForCompletion(JobImpl job) {
+ protected JobStateInternal checkJobAfterTaskCompletion(JobImpl job) {
if (job.completedTaskCount == job.tasks.size()) {
job.setFinishTime();
- job.abortJob(org.apache.hadoop.mapreduce.JobStatus.State.KILLED);
- return job.finished(JobStateInternal.KILLED);
+ 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();
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/MapTaskImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/MapTaskImpl.java?rev=1432796&r1=1432795&r2=1432796&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/MapTaskImpl.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/MapTaskImpl.java Mon Jan 14 03:44:35 2013
@@ -24,7 +24,6 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.MapTaskAttemptImpl;
import org.apache.hadoop.mapreduce.MRJobConfig;
-import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.jobhistory.JobHistoryParser.TaskInfo;
import org.apache.hadoop.mapreduce.security.token.JobTokenIdentifier;
import org.apache.hadoop.mapreduce.split.JobSplit.TaskSplitMetaInfo;
@@ -47,13 +46,13 @@ public class MapTaskImpl extends TaskImp
public MapTaskImpl(JobId jobId, int partition, EventHandler eventHandler,
Path remoteJobConfFile, JobConf conf,
TaskSplitMetaInfo taskSplitMetaInfo,
- TaskAttemptListener taskAttemptListener, OutputCommitter committer,
+ TaskAttemptListener taskAttemptListener,
Token<JobTokenIdentifier> jobToken,
Credentials credentials, Clock clock,
Map<TaskId, TaskInfo> completedTasksFromPreviousRun, int startCount,
MRAppMetrics metrics, AppContext appContext) {
super(jobId, TaskType.MAP, partition, eventHandler, remoteJobConfFile,
- conf, taskAttemptListener, committer, jobToken, credentials, clock,
+ conf, taskAttemptListener, jobToken, credentials, clock,
completedTasksFromPreviousRun, startCount, metrics, appContext);
this.taskSplitMetaInfo = taskSplitMetaInfo;
}
@@ -68,7 +67,7 @@ public class MapTaskImpl extends TaskImp
return new MapTaskAttemptImpl(getID(), nextAttemptNumber,
eventHandler, jobFile,
partition, taskSplitMetaInfo, conf, taskAttemptListener,
- committer, jobToken, credentials, clock, appContext);
+ jobToken, credentials, clock, appContext);
}
@Override
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/ReduceTaskImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/ReduceTaskImpl.java?rev=1432796&r1=1432795&r2=1432796&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/ReduceTaskImpl.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/ReduceTaskImpl.java Mon Jan 14 03:44:35 2013
@@ -24,7 +24,6 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.ReduceTaskAttemptImpl;
import org.apache.hadoop.mapreduce.MRJobConfig;
-import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.jobhistory.JobHistoryParser.TaskInfo;
import org.apache.hadoop.mapreduce.security.token.JobTokenIdentifier;
import org.apache.hadoop.mapreduce.v2.api.records.JobId;
@@ -46,12 +45,12 @@ public class ReduceTaskImpl extends Task
public ReduceTaskImpl(JobId jobId, int partition,
EventHandler eventHandler, Path jobFile, JobConf conf,
int numMapTasks, TaskAttemptListener taskAttemptListener,
- OutputCommitter committer, Token<JobTokenIdentifier> jobToken,
+ Token<JobTokenIdentifier> jobToken,
Credentials credentials, Clock clock,
Map<TaskId, TaskInfo> completedTasksFromPreviousRun, int startCount,
MRAppMetrics metrics, AppContext appContext) {
super(jobId, TaskType.REDUCE, partition, eventHandler, jobFile, conf,
- taskAttemptListener, committer, jobToken, credentials, clock,
+ taskAttemptListener, jobToken, credentials, clock,
completedTasksFromPreviousRun, startCount, metrics, appContext);
this.numMapTasks = numMapTasks;
}
@@ -66,7 +65,7 @@ public class ReduceTaskImpl extends Task
return new ReduceTaskAttemptImpl(getID(), nextAttemptNumber,
eventHandler, jobFile,
partition, numMapTasks, conf, taskAttemptListener,
- committer, jobToken, credentials, clock, appContext);
+ jobToken, credentials, clock, appContext);
}
@Override
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java?rev=1432796&r1=1432795&r2=1432796&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java Mon Jan 14 03:44:35 2013
@@ -39,7 +39,6 @@ import java.util.regex.Pattern;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
@@ -57,7 +56,6 @@ import org.apache.hadoop.mapreduce.Count
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.JobCounter;
import org.apache.hadoop.mapreduce.MRJobConfig;
-import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TaskCounter;
import org.apache.hadoop.mapreduce.TypeConverter;
@@ -76,6 +74,7 @@ import org.apache.hadoop.mapreduce.v2.ap
import org.apache.hadoop.mapreduce.v2.api.records.TaskType;
import org.apache.hadoop.mapreduce.v2.app.AppContext;
import org.apache.hadoop.mapreduce.v2.app.TaskAttemptListener;
+import org.apache.hadoop.mapreduce.v2.app.commit.CommitterTaskAbortEvent;
import org.apache.hadoop.mapreduce.v2.app.job.TaskAttemptStateInternal;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobCounterUpdateEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobDiagnosticsUpdateEvent;
@@ -99,7 +98,6 @@ import org.apache.hadoop.mapreduce.v2.ap
import org.apache.hadoop.mapreduce.v2.app.rm.ContainerAllocatorEvent;
import org.apache.hadoop.mapreduce.v2.app.rm.ContainerRequestEvent;
import org.apache.hadoop.mapreduce.v2.app.speculate.SpeculatorEvent;
-import org.apache.hadoop.mapreduce.v2.app.taskclean.TaskCleanupEvent;
import org.apache.hadoop.mapreduce.v2.util.MRApps;
import org.apache.hadoop.net.NetUtils;
import org.apache.hadoop.security.Credentials;
@@ -157,7 +155,6 @@ public abstract class TaskAttemptImpl im
private final Clock clock;
private final org.apache.hadoop.mapred.JobID oldJobId;
private final TaskAttemptListener taskAttemptListener;
- private final OutputCommitter committer;
private final Resource resourceCapability;
private final String[] dataLocalHosts;
private final List<String> diagnostics = new ArrayList<String>();
@@ -168,6 +165,7 @@ public abstract class TaskAttemptImpl im
private Token<JobTokenIdentifier> jobToken;
private static AtomicBoolean initialClasspathFlag = new AtomicBoolean();
private static String initialClasspath = null;
+ private static String initialAppClasspath = null;
private static Object commonContainerSpecLock = new Object();
private static ContainerLaunchContext commonContainerSpec = null;
private static final Object classpathLock = new Object();
@@ -501,7 +499,7 @@ public abstract class TaskAttemptImpl im
public TaskAttemptImpl(TaskId taskId, int i,
EventHandler eventHandler,
TaskAttemptListener taskAttemptListener, Path jobFile, int partition,
- JobConf conf, String[] dataLocalHosts, OutputCommitter committer,
+ JobConf conf, String[] dataLocalHosts,
Token<JobTokenIdentifier> jobToken,
Credentials credentials, Clock clock,
AppContext appContext) {
@@ -525,13 +523,15 @@ public abstract class TaskAttemptImpl im
this.credentials = credentials;
this.jobToken = jobToken;
this.eventHandler = eventHandler;
- this.committer = committer;
this.jobFile = jobFile;
this.partition = partition;
//TODO:create the resource reqt for this Task attempt
this.resourceCapability = recordFactory.newRecordInstance(Resource.class);
- this.resourceCapability.setMemory(getMemoryRequired(conf, taskId.getTaskType()));
+ this.resourceCapability.setMemory(
+ getMemoryRequired(conf, taskId.getTaskType()));
+ this.resourceCapability.setVirtualCores(
+ getCpuRequired(conf, taskId.getTaskType()));
this.dataLocalHosts = dataLocalHosts;
RackResolver.init(conf);
@@ -555,6 +555,21 @@ public abstract class TaskAttemptImpl im
return memory;
}
+ private int getCpuRequired(Configuration conf, TaskType taskType) {
+ int vcores = 1;
+ if (taskType == TaskType.MAP) {
+ vcores =
+ conf.getInt(MRJobConfig.MAP_CPU_VCORES,
+ MRJobConfig.DEFAULT_MAP_CPU_VCORES);
+ } else if (taskType == TaskType.REDUCE) {
+ vcores =
+ conf.getInt(MRJobConfig.REDUCE_CPU_VCORES,
+ MRJobConfig.DEFAULT_REDUCE_CPU_VCORES);
+ }
+
+ return vcores;
+ }
+
/**
* Create a {@link LocalResource} record with all the given parameters.
*/
@@ -585,6 +600,7 @@ public abstract class TaskAttemptImpl im
Map<String, String> env = new HashMap<String, String>();
MRApps.setClasspath(env, conf);
initialClasspath = env.get(Environment.CLASSPATH.name());
+ initialAppClasspath = env.get(Environment.APP_CLASSPATH.name());
initialClasspathFlag.set(true);
return initialClasspath;
}
@@ -683,6 +699,13 @@ public abstract class TaskAttemptImpl im
environment,
Environment.CLASSPATH.name(),
getInitialClasspath(conf));
+
+ if (initialAppClasspath != null) {
+ Apps.addToEnvironment(
+ environment,
+ Environment.APP_CLASSPATH.name(),
+ initialAppClasspath);
+ }
} catch (IOException e) {
throw new YarnException(e);
}
@@ -1436,10 +1459,8 @@ public abstract class TaskAttemptImpl im
TaskAttemptContext taskContext =
new TaskAttemptContextImpl(taskAttempt.conf,
TypeConverter.fromYarn(taskAttempt.attemptId));
- taskAttempt.eventHandler.handle(new TaskCleanupEvent(
- taskAttempt.attemptId,
- taskAttempt.committer,
- taskContext));
+ taskAttempt.eventHandler.handle(new CommitterTaskAbortEvent(
+ taskAttempt.attemptId, taskContext));
}
}
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskImpl.java?rev=1432796&r1=1432795&r2=1432796&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskImpl.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskImpl.java Mon Jan 14 03:44:35 2013
@@ -37,7 +37,6 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.MRConfig;
-import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.TaskAttemptID;
import org.apache.hadoop.mapreduce.TypeConverter;
import org.apache.hadoop.mapreduce.jobhistory.JobHistoryEvent;
@@ -100,7 +99,6 @@ public abstract class TaskImpl implement
protected final JobConf conf;
protected final Path jobFile;
- protected final OutputCommitter committer;
protected final int partition;
protected final TaskAttemptListener taskAttemptListener;
protected final EventHandler eventHandler;
@@ -278,7 +276,7 @@ public abstract class TaskImpl implement
public TaskImpl(JobId jobId, TaskType taskType, int partition,
EventHandler eventHandler, Path remoteJobConfFile, JobConf conf,
- TaskAttemptListener taskAttemptListener, OutputCommitter committer,
+ TaskAttemptListener taskAttemptListener,
Token<JobTokenIdentifier> jobToken,
Credentials credentials, Clock clock,
Map<TaskId, TaskInfo> completedTasksFromPreviousRun, int startCount,
@@ -301,7 +299,6 @@ public abstract class TaskImpl implement
this.partition = partition;
this.taskAttemptListener = taskAttemptListener;
this.eventHandler = eventHandler;
- this.committer = committer;
this.credentials = credentials;
this.jobToken = jobToken;
this.metrics = metrics;
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/recover/RecoveryService.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/recover/RecoveryService.java?rev=1432796&r1=1432795&r2=1432796&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/recover/RecoveryService.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/recover/RecoveryService.java Mon Jan 14 03:44:35 2013
@@ -31,6 +31,7 @@ import org.apache.hadoop.conf.Configurat
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.OutputCommitter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;
@@ -48,6 +49,8 @@ import org.apache.hadoop.mapreduce.v2.ap
import org.apache.hadoop.mapreduce.v2.api.records.TaskId;
import org.apache.hadoop.mapreduce.v2.api.records.TaskState;
import org.apache.hadoop.mapreduce.v2.app.ControlledClock;
+import org.apache.hadoop.mapreduce.v2.app.commit.CommitterTaskAbortEvent;
+import org.apache.hadoop.mapreduce.v2.app.commit.CommitterEventType;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobDiagnosticsUpdateEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobEvent;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobEventType;
@@ -65,8 +68,6 @@ import org.apache.hadoop.mapreduce.v2.ap
import org.apache.hadoop.mapreduce.v2.app.launcher.ContainerRemoteLaunchEvent;
import org.apache.hadoop.mapreduce.v2.app.rm.ContainerAllocator;
import org.apache.hadoop.mapreduce.v2.app.rm.ContainerAllocatorEvent;
-import org.apache.hadoop.mapreduce.v2.app.taskclean.TaskCleaner;
-import org.apache.hadoop.mapreduce.v2.app.taskclean.TaskCleanupEvent;
import org.apache.hadoop.mapreduce.v2.jobhistory.JobHistoryUtils;
import org.apache.hadoop.mapreduce.v2.util.MRBuilderUtils;
import org.apache.hadoop.yarn.Clock;
@@ -100,6 +101,7 @@ public class RecoveryService extends Com
private final ApplicationAttemptId applicationAttemptId;
private final OutputCommitter committer;
+ private final boolean newApiCommitter;
private final Dispatcher dispatcher;
private final ControlledClock clock;
@@ -113,10 +115,11 @@ public class RecoveryService extends Com
private volatile boolean recoveryMode = false;
public RecoveryService(ApplicationAttemptId applicationAttemptId,
- Clock clock, OutputCommitter committer) {
+ Clock clock, OutputCommitter committer, boolean newApiCommitter) {
super("RecoveringDispatcher");
this.applicationAttemptId = applicationAttemptId;
this.committer = committer;
+ this.newApiCommitter = newApiCommitter;
this.dispatcher = createRecoveryDispatcher();
this.clock = new ControlledClock(clock);
addService((Service) dispatcher);
@@ -205,18 +208,18 @@ public class RecoveryService extends Com
throws IOException {
FSDataInputStream in = null;
Path historyFile = null;
- String jobName =
+ String jobId =
TypeConverter.fromYarn(applicationAttemptId.getApplicationId())
.toString();
String jobhistoryDir =
- JobHistoryUtils.getConfiguredHistoryStagingDirPrefix(conf);
+ JobHistoryUtils.getConfiguredHistoryStagingDirPrefix(conf, jobId);
Path histDirPath =
FileContext.getFileContext(conf).makeQualified(new Path(jobhistoryDir));
FileContext fc = FileContext.getFileContext(histDirPath.toUri(), conf);
// read the previous history file
historyFile =
fc.makeQualified(JobHistoryUtils.getStagingJobHistoryFile(histDirPath,
- jobName, (applicationAttemptId.getAttemptId() - 1)));
+ jobId, (applicationAttemptId.getAttemptId() - 1)));
LOG.info("History file is at " + historyFile);
in = fc.open(historyFile);
return in;
@@ -339,8 +342,8 @@ public class RecoveryService extends Com
return;
}
- else if (event.getType() == TaskCleaner.EventType.TASK_CLEAN) {
- TaskAttemptId aId = ((TaskCleanupEvent) event).getAttemptID();
+ else if (event.getType() == CommitterEventType.TASK_ABORT) {
+ TaskAttemptId aId = ((CommitterTaskAbortEvent) event).getAttemptID();
LOG.debug("TASK_CLEAN");
actualHandler.handle(new TaskAttemptEvent(aId,
TaskAttemptEventType.TA_CLEANUP_DONE));
@@ -360,8 +363,17 @@ public class RecoveryService extends Com
switch (state) {
case SUCCEEDED:
//recover the task output
- TaskAttemptContext taskContext = new TaskAttemptContextImpl(getConfig(),
- attInfo.getAttemptId());
+
+ // check the committer type and construct corresponding context
+ TaskAttemptContext taskContext = null;
+ if(newApiCommitter) {
+ taskContext = new TaskAttemptContextImpl(getConfig(),
+ attInfo.getAttemptId());
+ } else {
+ taskContext = new org.apache.hadoop.mapred.TaskAttemptContextImpl(new JobConf(getConfig()),
+ TypeConverter.fromYarn(aId));
+ }
+
try {
TaskType type = taskContext.getTaskAttemptID().getTaskID().getTaskType();
int numReducers = taskContext.getConfiguration().getInt(MRJobConfig.NUM_REDUCES, 1);
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMCommunicator.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMCommunicator.java?rev=1432796&r1=1432795&r2=1432796&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMCommunicator.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/rm/RMCommunicator.java Mon Jan 14 03:44:35 2013
@@ -22,6 +22,7 @@ import java.io.IOException;
import java.net.InetSocketAddress;
import java.security.PrivilegedAction;
import java.util.Map;
+import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.commons.logging.Log;
@@ -62,7 +63,8 @@ import org.apache.hadoop.yarn.service.Ab
/**
* Registers/unregisters to RM and sends heartbeats to RM.
*/
-public abstract class RMCommunicator extends AbstractService {
+public abstract class RMCommunicator extends AbstractService
+ implements RMHeartbeatHandler {
private static final Log LOG = LogFactory.getLog(RMContainerAllocator.class);
private int rmPollInterval;//millis
protected ApplicationId applicationId;
@@ -77,6 +79,8 @@ public abstract class RMCommunicator ext
private Resource minContainerCapability;
private Resource maxContainerCapability;
protected Map<ApplicationAccessType, String> applicationACLs;
+ private volatile long lastHeartbeatTime;
+ private ConcurrentLinkedQueue<Runnable> heartbeatCallbacks;
private final RecordFactory recordFactory =
RecordFactoryProvider.getRecordFactory(null);
@@ -95,6 +99,7 @@ public abstract class RMCommunicator ext
this.applicationId = context.getApplicationID();
this.applicationAttemptId = context.getApplicationAttemptId();
this.stopped = new AtomicBoolean(false);
+ this.heartbeatCallbacks = new ConcurrentLinkedQueue<Runnable>();
}
@Override
@@ -136,14 +141,19 @@ public abstract class RMCommunicator ext
protected void register() {
//Register
- InetSocketAddress serviceAddr = clientService.getBindAddress();
+ InetSocketAddress serviceAddr = null;
+ if (clientService != null ) {
+ serviceAddr = clientService.getBindAddress();
+ }
try {
RegisterApplicationMasterRequest request =
recordFactory.newRecordInstance(RegisterApplicationMasterRequest.class);
request.setApplicationAttemptId(applicationAttemptId);
- request.setHost(serviceAddr.getHostName());
- request.setRpcPort(serviceAddr.getPort());
- request.setTrackingUrl(serviceAddr.getHostName() + ":" + clientService.getHttpPort());
+ if (serviceAddr != null) {
+ request.setHost(serviceAddr.getHostName());
+ request.setRpcPort(serviceAddr.getPort());
+ request.setTrackingUrl(serviceAddr.getHostName() + ":" + clientService.getHttpPort());
+ }
RegisterApplicationMasterResponse response =
scheduler.registerApplicationMaster(request);
minContainerCapability = response.getMinimumResourceCapability();
@@ -236,8 +246,12 @@ public abstract class RMCommunicator ext
return;
} catch (Exception e) {
LOG.error("ERROR IN CONTACTING RM. ", e);
+ continue;
// TODO: for other exceptions
}
+
+ lastHeartbeatTime = context.getClock().getTime();
+ executeHeartbeatCallbacks();
} catch (InterruptedException e) {
if (!stopped.get()) {
LOG.warn("Allocated thread interrupted. Returning.");
@@ -295,6 +309,23 @@ public abstract class RMCommunicator ext
protected abstract void heartbeat() throws Exception;
+ private void executeHeartbeatCallbacks() {
+ Runnable callback = null;
+ while ((callback = heartbeatCallbacks.poll()) != null) {
+ callback.run();
+ }
+ }
+
+ @Override
+ public long getLastHeartbeatTime() {
+ return lastHeartbeatTime;
+ }
+
+ @Override
+ public void runOnNextHeartbeat(Runnable callback) {
+ heartbeatCallbacks.add(callback);
+ }
+
public void setShouldUnregister(boolean shouldUnregister) {
this.shouldUnregister = shouldUnregister;
LOG.info("RMCommunicator notified that shouldUnregistered is: "
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapred/TestTaskAttemptListenerImpl.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapred/TestTaskAttemptListenerImpl.java?rev=1432796&r1=1432795&r2=1432796&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapred/TestTaskAttemptListenerImpl.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapred/TestTaskAttemptListenerImpl.java Mon Jan 14 03:44:35 2013
@@ -24,6 +24,8 @@ import static org.junit.Assert.assertNul
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
@@ -41,7 +43,9 @@ import org.apache.hadoop.mapreduce.v2.ap
import org.apache.hadoop.mapreduce.v2.app.AppContext;
import org.apache.hadoop.mapreduce.v2.app.TaskHeartbeatHandler;
import org.apache.hadoop.mapreduce.v2.app.job.Job;
+import org.apache.hadoop.mapreduce.v2.app.rm.RMHeartbeatHandler;
import org.apache.hadoop.mapreduce.v2.util.MRBuilderUtils;
+import org.apache.hadoop.yarn.SystemClock;
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.junit.Test;
@@ -51,8 +55,9 @@ public class TestTaskAttemptListenerImpl
public MockTaskAttemptListenerImpl(AppContext context,
JobTokenSecretManager jobTokenSecretManager,
+ RMHeartbeatHandler rmHeartbeatHandler,
TaskHeartbeatHandler hbHandler) {
- super(context, jobTokenSecretManager);
+ super(context, jobTokenSecretManager, rmHeartbeatHandler);
this.taskHeartbeatHandler = hbHandler;
}
@@ -76,9 +81,12 @@ public class TestTaskAttemptListenerImpl
public void testGetTask() throws IOException {
AppContext appCtx = mock(AppContext.class);
JobTokenSecretManager secret = mock(JobTokenSecretManager.class);
+ RMHeartbeatHandler rmHeartbeatHandler =
+ mock(RMHeartbeatHandler.class);
TaskHeartbeatHandler hbHandler = mock(TaskHeartbeatHandler.class);
MockTaskAttemptListenerImpl listener =
- new MockTaskAttemptListenerImpl(appCtx, secret, hbHandler);
+ new MockTaskAttemptListenerImpl(appCtx, secret,
+ rmHeartbeatHandler, hbHandler);
Configuration conf = new Configuration();
listener.init(conf);
listener.start();
@@ -152,9 +160,11 @@ public class TestTaskAttemptListenerImpl
AppContext appCtx = mock(AppContext.class);
when(appCtx.getJob(any(JobId.class))).thenReturn(mockJob);
JobTokenSecretManager secret = mock(JobTokenSecretManager.class);
+ RMHeartbeatHandler rmHeartbeatHandler =
+ mock(RMHeartbeatHandler.class);
final TaskHeartbeatHandler hbHandler = mock(TaskHeartbeatHandler.class);
TaskAttemptListenerImpl listener =
- new TaskAttemptListenerImpl(appCtx, secret) {
+ new TaskAttemptListenerImpl(appCtx, secret, rmHeartbeatHandler) {
@Override
protected void registerHeartbeatHandler(Configuration conf) {
taskHeartbeatHandler = hbHandler;
@@ -191,4 +201,46 @@ public class TestTaskAttemptListenerImpl
return tce;
}
+ @Test
+ public void testCommitWindow() throws IOException {
+ SystemClock clock = new SystemClock();
+
+ org.apache.hadoop.mapreduce.v2.app.job.Task mockTask =
+ mock(org.apache.hadoop.mapreduce.v2.app.job.Task.class);
+ when(mockTask.canCommit(any(TaskAttemptId.class))).thenReturn(true);
+ Job mockJob = mock(Job.class);
+ when(mockJob.getTask(any(TaskId.class))).thenReturn(mockTask);
+ AppContext appCtx = mock(AppContext.class);
+ when(appCtx.getJob(any(JobId.class))).thenReturn(mockJob);
+ when(appCtx.getClock()).thenReturn(clock);
+ JobTokenSecretManager secret = mock(JobTokenSecretManager.class);
+ RMHeartbeatHandler rmHeartbeatHandler =
+ mock(RMHeartbeatHandler.class);
+ final TaskHeartbeatHandler hbHandler = mock(TaskHeartbeatHandler.class);
+ TaskAttemptListenerImpl listener =
+ new TaskAttemptListenerImpl(appCtx, secret, rmHeartbeatHandler) {
+ @Override
+ protected void registerHeartbeatHandler(Configuration conf) {
+ taskHeartbeatHandler = hbHandler;
+ }
+ };
+
+ Configuration conf = new Configuration();
+ listener.init(conf);
+ listener.start();
+
+ // verify commit not allowed when RM heartbeat has not occurred recently
+ TaskAttemptID tid = new TaskAttemptID("12345", 1, TaskType.REDUCE, 1, 0);
+ boolean canCommit = listener.canCommit(tid);
+ assertFalse(canCommit);
+ verify(mockTask, never()).canCommit(any(TaskAttemptId.class));
+
+ // verify commit allowed when RM heartbeat is recent
+ when(rmHeartbeatHandler.getLastHeartbeatTime()).thenReturn(clock.getTime());
+ canCommit = listener.canCommit(tid);
+ assertTrue(canCommit);
+ verify(mockTask, times(1)).canCommit(any(TaskAttemptId.class));
+
+ listener.stop();
+ }
}
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/jobhistory/TestJobHistoryEventHandler.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/jobhistory/TestJobHistoryEventHandler.java?rev=1432796&r1=1432795&r2=1432796&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/jobhistory/TestJobHistoryEventHandler.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/jobhistory/TestJobHistoryEventHandler.java Mon Jan 14 03:44:35 2013
@@ -18,14 +18,9 @@
package org.apache.hadoop.mapreduce.jobhistory;
-import static junit.framework.Assert.assertFalse;
-import static junit.framework.Assert.assertTrue;
-import static org.mockito.Matchers.any;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.spy;
-import static org.mockito.Mockito.times;
-import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
+import static junit.framework.Assert.*;
+import static org.mockito.Matchers.*;
+import static org.mockito.Mockito.*;
import java.io.File;
import java.io.IOException;
@@ -51,7 +46,6 @@ import org.apache.hadoop.yarn.api.record
import org.apache.hadoop.yarn.util.BuilderUtils;
import org.junit.Test;
import org.mockito.Mockito;
-import org.mockito.verification.VerificationMode;
public class TestJobHistoryEventHandler {
@@ -260,13 +254,15 @@ public class TestJobHistoryEventHandler
}
}
- private AppContext mockAppContext(JobId jobId) {
+ private AppContext mockAppContext(ApplicationId appId) {
+ JobId jobId = TypeConverter.toYarn(TypeConverter.fromYarn(appId));
AppContext mockContext = mock(AppContext.class);
Job mockJob = mock(Job.class);
when(mockJob.getTotalMaps()).thenReturn(10);
when(mockJob.getTotalReduces()).thenReturn(10);
when(mockJob.getName()).thenReturn("mockjob");
when(mockContext.getJob(jobId)).thenReturn(mockJob);
+ when(mockContext.getApplicationID()).thenReturn(appId);
return mockContext;
}
@@ -279,7 +275,7 @@ public class TestJobHistoryEventHandler
ContainerId containerId = BuilderUtils.newContainerId(appAttemptId, 1);
TaskID taskID = TaskID.forName("task_200707121733_0003_m_000005");
JobId jobId = MRBuilderUtils.newJobId(appId, 1);
- AppContext mockAppContext = mockAppContext(jobId);
+ AppContext mockAppContext = mockAppContext(appId);
}
private JobHistoryEvent getEventToEnqueue(JobId jobId) {
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MRApp.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MRApp.java?rev=1432796&r1=1432795&r2=1432796&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MRApp.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MRApp.java Mon Jan 14 03:44:35 2013
@@ -30,11 +30,15 @@ import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileContext;
+import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapred.WrappedJvmID;
+import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.JobID;
+import org.apache.hadoop.mapreduce.JobStatus.State;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.OutputCommitter;
+import org.apache.hadoop.mapreduce.TaskAttemptContext;
import org.apache.hadoop.mapreduce.TypeConverter;
import org.apache.hadoop.mapreduce.jobhistory.JobHistoryEvent;
import org.apache.hadoop.mapreduce.jobhistory.NormalizedResourceEvent;
@@ -49,6 +53,8 @@ import org.apache.hadoop.mapreduce.v2.ap
import org.apache.hadoop.mapreduce.v2.api.records.TaskReport;
import org.apache.hadoop.mapreduce.v2.api.records.TaskState;
import org.apache.hadoop.mapreduce.v2.app.client.ClientService;
+import org.apache.hadoop.mapreduce.v2.app.commit.CommitterEvent;
+import org.apache.hadoop.mapreduce.v2.app.commit.CommitterEventHandler;
import org.apache.hadoop.mapreduce.v2.app.job.Job;
import org.apache.hadoop.mapreduce.v2.app.job.JobStateInternal;
import org.apache.hadoop.mapreduce.v2.app.job.Task;
@@ -69,8 +75,7 @@ import org.apache.hadoop.mapreduce.v2.ap
import org.apache.hadoop.mapreduce.v2.app.launcher.ContainerLauncherEvent;
import org.apache.hadoop.mapreduce.v2.app.rm.ContainerAllocator;
import org.apache.hadoop.mapreduce.v2.app.rm.ContainerAllocatorEvent;
-import org.apache.hadoop.mapreduce.v2.app.taskclean.TaskCleaner;
-import org.apache.hadoop.mapreduce.v2.app.taskclean.TaskCleanupEvent;
+import org.apache.hadoop.mapreduce.v2.app.rm.RMHeartbeatHandler;
import org.apache.hadoop.mapreduce.v2.util.MRApps;
import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
import org.apache.hadoop.net.NetUtils;
@@ -204,6 +209,16 @@ public class MRApp extends MRAppMaster {
@Override
public void init(Configuration conf) {
+ try {
+ //Create the staging directory if it does not exist
+ String user = UserGroupInformation.getCurrentUser().getShortUserName();
+ Path stagingDir = MRApps.getStagingAreaDir(conf, user);
+ FileSystem fs = getFileSystem(conf);
+ fs.mkdirs(stagingDir);
+ } catch (Exception e) {
+ throw new YarnException("Error creating staging dir", e);
+ }
+
super.init(conf);
if (this.clusterInfo != null) {
getContext().getClusterInfo().setMinContainerCapability(
@@ -212,9 +227,9 @@ public class MRApp extends MRAppMaster {
this.clusterInfo.getMaxContainerCapability());
} else {
getContext().getClusterInfo().setMinContainerCapability(
- BuilderUtils.newResource(1024));
+ BuilderUtils.newResource(1024, 1));
getContext().getClusterInfo().setMaxContainerCapability(
- BuilderUtils.newResource(10240));
+ BuilderUtils.newResource(10240, 1));
}
}
@@ -384,7 +399,8 @@ public class MRApp extends MRAppMaster {
}
@Override
- protected Job createJob(Configuration conf) {
+ protected Job createJob(Configuration conf, JobStateInternal forcedState,
+ String diagnostic) {
UserGroupInformation currentUser = null;
try {
currentUser = UserGroupInformation.getCurrentUser();
@@ -394,8 +410,8 @@ public class MRApp extends MRAppMaster {
Job newJob = new TestJob(getJobId(), getAttemptID(), conf,
getDispatcher().getEventHandler(),
getTaskAttemptListener(), getContext().getClock(),
- getCommitter(), isNewApiCommitter(),
- currentUser.getUserName(), getContext());
+ isNewApiCommitter(), currentUser.getUserName(), getContext(),
+ forcedState, diagnostic);
((AppContext) getContext()).getAllJobs().put(newJob.getID(), newJob);
getDispatcher().register(JobFinishEvent.Type.class,
@@ -487,7 +503,8 @@ public class MRApp extends MRAppMaster {
return new MRAppContainerAllocator();
}
- protected class MRAppContainerAllocator implements ContainerAllocator {
+ protected class MRAppContainerAllocator
+ implements ContainerAllocator, RMHeartbeatHandler {
private int containerCount;
@Override
@@ -512,19 +529,70 @@ public class MRApp extends MRAppMaster {
new TaskAttemptContainerAssignedEvent(event.getAttemptID(),
container, null));
}
+
+ @Override
+ public long getLastHeartbeatTime() {
+ return getContext().getClock().getTime();
+ }
+
+ @Override
+ public void runOnNextHeartbeat(Runnable callback) {
+ callback.run();
+ }
}
@Override
- protected TaskCleaner createTaskCleaner(AppContext context) {
- return new TaskCleaner() {
+ protected EventHandler<CommitterEvent> createCommitterEventHandler(
+ AppContext context, final OutputCommitter committer) {
+ // create an output committer with the task methods stubbed out
+ OutputCommitter stubbedCommitter = new OutputCommitter() {
@Override
- public void handle(TaskCleanupEvent event) {
- //send the cleanup done event
- getContext().getEventHandler().handle(
- new TaskAttemptEvent(event.getAttemptID(),
- TaskAttemptEventType.TA_CLEANUP_DONE));
+ public void setupJob(JobContext jobContext) throws IOException {
+ committer.setupJob(jobContext);
+ }
+ @SuppressWarnings("deprecation")
+ @Override
+ public void cleanupJob(JobContext jobContext) throws IOException {
+ committer.cleanupJob(jobContext);
+ }
+ @Override
+ public void commitJob(JobContext jobContext) throws IOException {
+ committer.commitJob(jobContext);
+ }
+ @Override
+ public void abortJob(JobContext jobContext, State state)
+ throws IOException {
+ committer.abortJob(jobContext, state);
+ }
+ @Override
+ public boolean isRecoverySupported() {
+ return committer.isRecoverySupported();
+ }
+ @Override
+ public void setupTask(TaskAttemptContext taskContext)
+ throws IOException {
+ }
+ @Override
+ public boolean needsTaskCommit(TaskAttemptContext taskContext)
+ throws IOException {
+ return false;
+ }
+ @Override
+ public void commitTask(TaskAttemptContext taskContext)
+ throws IOException {
+ }
+ @Override
+ public void abortTask(TaskAttemptContext taskContext)
+ throws IOException {
+ }
+ @Override
+ public void recoverTask(TaskAttemptContext taskContext)
+ throws IOException {
}
};
+
+ return new CommitterEventHandler(context, stubbedCommitter,
+ getRMHeartbeatHandler());
}
@Override
@@ -576,14 +644,14 @@ public class MRApp extends MRAppMaster {
public TestJob(JobId jobId, ApplicationAttemptId applicationAttemptId,
Configuration conf, EventHandler eventHandler,
TaskAttemptListener taskAttemptListener, Clock clock,
- OutputCommitter committer, boolean newApiCommitter, String user,
- AppContext appContext) {
+ boolean newApiCommitter, String user, AppContext appContext,
+ JobStateInternal forcedState, String diagnostic) {
super(jobId, getApplicationAttemptId(applicationId, getStartCount()),
conf, eventHandler, taskAttemptListener,
new JobTokenSecretManager(), new Credentials(), clock,
- getCompletedTaskFromPreviousRun(), metrics, committer,
+ getCompletedTaskFromPreviousRun(), metrics,
newApiCommitter, user, System.currentTimeMillis(), getAllAMInfos(),
- appContext);
+ appContext, forcedState, diagnostic);
// This "this leak" is okay because the retained pointer is in an
// instance variable.
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MRAppBenchmark.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MRAppBenchmark.java?rev=1432796&r1=1432795&r2=1432796&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MRAppBenchmark.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/MRAppBenchmark.java Mon Jan 14 03:44:35 2013
@@ -207,9 +207,9 @@ public class MRAppBenchmark {
RegisterApplicationMasterResponse response =
Records.newRecord(RegisterApplicationMasterResponse.class);
response.setMinimumResourceCapability(BuilderUtils
- .newResource(1024));
+ .newResource(1024, 1));
response.setMaximumResourceCapability(BuilderUtils
- .newResource(10240));
+ .newResource(10240, 1));
return response;
}
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestFail.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestFail.java?rev=1432796&r1=1432795&r2=1432796&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestFail.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestFail.java Mon Jan 14 03:44:35 2013
@@ -252,7 +252,7 @@ public class TestFail {
//task time out is reduced
//when attempt times out, heartbeat handler will send the lost event
//leading to Attempt failure
- return new TaskAttemptListenerImpl(getContext(), null) {
+ return new TaskAttemptListenerImpl(getContext(), null, null) {
@Override
public void startRpcServer(){};
@Override
Modified: hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestMRApp.java
URL: http://svn.apache.org/viewvc/hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestMRApp.java?rev=1432796&r1=1432795&r2=1432796&view=diff
==============================================================================
--- hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestMRApp.java (original)
+++ hadoop/common/branches/HDFS-2802/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestMRApp.java Mon Jan 14 03:44:35 2013
@@ -37,6 +37,7 @@ import org.apache.hadoop.mapreduce.v2.ap
import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptState;
import org.apache.hadoop.mapreduce.v2.api.records.TaskState;
import org.apache.hadoop.mapreduce.v2.app.job.Job;
+import org.apache.hadoop.mapreduce.v2.app.job.JobStateInternal;
import org.apache.hadoop.mapreduce.v2.app.job.Task;
import org.apache.hadoop.mapreduce.v2.app.job.TaskAttempt;
import org.apache.hadoop.mapreduce.v2.app.job.event.JobUpdatedNodesEvent;
@@ -370,8 +371,9 @@ public class TestMRApp {
}
@Override
- protected Job createJob(Configuration conf) {
- spiedJob = spy((JobImpl) super.createJob(conf));
+ protected Job createJob(Configuration conf, JobStateInternal forcedState,
+ String diagnostic) {
+ spiedJob = spy((JobImpl) super.createJob(conf, forcedState, diagnostic));
((AppContext) getContext()).getAllJobs().put(spiedJob.getID(), spiedJob);
return spiedJob;
}