Author: tomwhite
Date: Tue Sep 11 14:13:58 2012
New Revision: 1383425
URL: http://svn.apache.org/viewvc?rev=1383425&view=rev
Log:
Merge -r 1383421:1383422 from trunk to branch-2. Fixes: MAPREDUCE-4607. Race
condition in ReduceTask completion can result in Task being incorrectly failed.
Contributed by Bikas Saha.
Modified:
hadoop/common/branches/branch-2/hadoop-mapreduce-project/CHANGES.txt
hadoop/common/branches/branch-2/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java
hadoop/common/branches/branch-2/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskImpl.java
hadoop/common/branches/branch-2/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestMRApp.java
hadoop/common/branches/branch-2/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java
Modified: hadoop/common/branches/branch-2/hadoop-mapreduce-project/CHANGES.txt
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-mapreduce-project/CHANGES.txt?rev=1383425&r1=1383424&r2=1383425&view=diff
==============================================================================
--- hadoop/common/branches/branch-2/hadoop-mapreduce-project/CHANGES.txt
(original)
+++ hadoop/common/branches/branch-2/hadoop-mapreduce-project/CHANGES.txt Tue
Sep 11 14:13:58 2012
@@ -10,7 +10,10 @@ Release 2.0.3-alpha - Unreleased
OPTIMIZATIONS
- BUG FIXES
+ BUG FIXES
+
+ MAPREDUCE-4607. Race condition in ReduceTask completion can result in Task
+ being incorrectly failed. (Bikas Saha via tomwhite)
Release 2.0.2-alpha - 2012-09-07
Modified:
hadoop/common/branches/branch-2/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/branch-2/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java?rev=1383425&r1=1383424&r2=1383425&view=diff
==============================================================================
---
hadoop/common/branches/branch-2/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/branch-2/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java
Tue Sep 11 14:13:58 2012
@@ -71,6 +71,7 @@ import org.apache.hadoop.mapreduce.v2.ap
import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptReport;
import org.apache.hadoop.mapreduce.v2.api.records.TaskAttemptState;
import org.apache.hadoop.mapreduce.v2.api.records.TaskId;
+import org.apache.hadoop.mapreduce.v2.api.records.TaskState;
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;
@@ -86,6 +87,7 @@ import org.apache.hadoop.mapreduce.v2.ap
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEventType;
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptKillEvent;
import
org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptStatusUpdateEvent;
+import org.apache.hadoop.mapreduce.v2.app.job.event.TaskEvent;
import
org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptStatusUpdateEvent.TaskAttemptStatus;
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskEventType;
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskTAttemptEvent;
@@ -120,6 +122,7 @@ import org.apache.hadoop.yarn.event.Even
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.apache.hadoop.yarn.state.InvalidStateTransitonException;
+import org.apache.hadoop.yarn.state.MultipleArcTransition;
import org.apache.hadoop.yarn.state.SingleArcTransition;
import org.apache.hadoop.yarn.state.StateMachine;
import org.apache.hadoop.yarn.state.StateMachineFactory;
@@ -128,6 +131,8 @@ import org.apache.hadoop.yarn.util.Build
import org.apache.hadoop.yarn.util.ConverterUtils;
import org.apache.hadoop.yarn.util.RackResolver;
+import com.google.common.base.Preconditions;
+
/**
* Implementation of TaskAttempt interface.
*/
@@ -404,10 +409,10 @@ public abstract class TaskAttemptImpl im
TaskAttemptState.FAILED,
TaskAttemptEventType.TA_TOO_MANY_FETCH_FAILURE,
new TooManyFetchFailureTransition())
- .addTransition(
- TaskAttemptState.SUCCEEDED, TaskAttemptState.KILLED,
- TaskAttemptEventType.TA_KILL,
- new KilledAfterSuccessTransition())
+ .addTransition(TaskAttemptState.SUCCEEDED,
+ EnumSet.of(TaskAttemptState.SUCCEEDED, TaskAttemptState.KILLED),
+ TaskAttemptEventType.TA_KILL,
+ new KilledAfterSuccessTransition())
.addTransition(
TaskAttemptState.SUCCEEDED, TaskAttemptState.SUCCEEDED,
TaskAttemptEventType.TA_DIAGNOSTICS_UPDATE,
@@ -1483,6 +1488,9 @@ public abstract class TaskAttemptImpl im
@SuppressWarnings("unchecked")
@Override
public void transition(TaskAttemptImpl taskAttempt, TaskAttemptEvent
event) {
+ // too many fetch failure can only happen for map tasks
+ Preconditions
+ .checkArgument(taskAttempt.getID().getTaskId().getTaskType() ==
TaskType.MAP);
//add to diagnostic
taskAttempt.addDiagnosticInfo("Too Many fetch failures.Failing the
attempt");
//set the finish time
@@ -1506,15 +1514,30 @@ public abstract class TaskAttemptImpl im
}
private static class KilledAfterSuccessTransition implements
- SingleArcTransition<TaskAttemptImpl, TaskAttemptEvent> {
+ MultipleArcTransition<TaskAttemptImpl, TaskAttemptEvent,
TaskAttemptState> {
@SuppressWarnings("unchecked")
@Override
- public void transition(TaskAttemptImpl taskAttempt,
+ public TaskAttemptState transition(TaskAttemptImpl taskAttempt,
TaskAttemptEvent event) {
- TaskAttemptKillEvent msgEvent = (TaskAttemptKillEvent) event;
- //add to diagnostic
- taskAttempt.addDiagnosticInfo(msgEvent.getMessage());
+ if(taskAttempt.getID().getTaskId().getTaskType() == TaskType.REDUCE) {
+ // after a reduce task has succeeded, its outputs are in safe in HDFS.
+ // logically such a task should not be killed. we only come here when
+ // there is a race condition in the event queue. E.g. some logic sends
+ // a kill request to this attempt when the successful completion event
+ // for this task is already in the event queue. so the kill event will
+ // get executed immediately after the attempt is marked successful and
+ // result in this transition being exercised.
+ // ignore this for reduce tasks
+ LOG.info("Ignoring killed event for successful reduce task attempt" +
+ taskAttempt.getID().toString());
+ return TaskAttemptState.SUCCEEDED;
+ }
+ if(event instanceof TaskAttemptKillEvent) {
+ TaskAttemptKillEvent msgEvent = (TaskAttemptKillEvent) event;
+ //add to diagnostic
+ taskAttempt.addDiagnosticInfo(msgEvent.getMessage());
+ }
// not setting a finish time since it was set on success
assert (taskAttempt.getFinishTime() != 0);
@@ -1528,6 +1551,7 @@ public abstract class TaskAttemptImpl im
.getTaskId().getJobId(), tauce));
taskAttempt.eventHandler.handle(new TaskTAttemptEvent(
taskAttempt.attemptId, TaskEventType.T_ATTEMPT_KILLED));
+ return TaskAttemptState.KILLED;
}
}
Modified:
hadoop/common/branches/branch-2/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/branch-2/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskImpl.java?rev=1383425&r1=1383424&r2=1383425&view=diff
==============================================================================
---
hadoop/common/branches/branch-2/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/branch-2/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskImpl.java
Tue Sep 11 14:13:58 2012
@@ -191,12 +191,12 @@ public abstract class TaskImpl implement
TaskEventType.T_ADD_SPEC_ATTEMPT))
// Transitions from SUCCEEDED state
- .addTransition(TaskState.SUCCEEDED, //only possible for map tasks
+ .addTransition(TaskState.SUCCEEDED,
EnumSet.of(TaskState.SCHEDULED, TaskState.SUCCEEDED, TaskState.FAILED),
- TaskEventType.T_ATTEMPT_FAILED, new MapRetroactiveFailureTransition())
- .addTransition(TaskState.SUCCEEDED, //only possible for map tasks
+ TaskEventType.T_ATTEMPT_FAILED, new RetroactiveFailureTransition())
+ .addTransition(TaskState.SUCCEEDED,
EnumSet.of(TaskState.SCHEDULED, TaskState.SUCCEEDED),
- TaskEventType.T_ATTEMPT_KILLED, new MapRetroactiveKilledTransition())
+ TaskEventType.T_ATTEMPT_KILLED, new RetroactiveKilledTransition())
// Ignore-able transitions.
.addTransition(
TaskState.SUCCEEDED, TaskState.SUCCEEDED,
@@ -897,7 +897,7 @@ public abstract class TaskImpl implement
}
}
- private static class MapRetroactiveFailureTransition
+ private static class RetroactiveFailureTransition
extends AttemptFailedTransition {
@Override
@@ -911,8 +911,8 @@ public abstract class TaskImpl implement
return TaskState.SUCCEEDED;
}
}
-
- //verify that this occurs only for map task
+
+ // a successful REDUCE task should not be overridden
//TODO: consider moving it to MapTaskImpl
if (!TaskType.MAP.equals(task.getType())) {
LOG.error("Unexpected event for REDUCE task " + event.getType());
@@ -938,42 +938,46 @@ public abstract class TaskImpl implement
}
}
- private static class MapRetroactiveKilledTransition implements
+ private static class RetroactiveKilledTransition implements
MultipleArcTransition<TaskImpl, TaskEvent, TaskState> {
@Override
public TaskState transition(TaskImpl task, TaskEvent event) {
- // verify that this occurs only for map task
+ TaskAttemptId attemptId = null;
+ if (event instanceof TaskTAttemptEvent) {
+ TaskTAttemptEvent castEvent = (TaskTAttemptEvent) event;
+ attemptId = castEvent.getTaskAttemptID();
+ if (task.getState() == TaskState.SUCCEEDED &&
+ !attemptId.equals(task.successfulAttempt)) {
+ // don't allow a different task attempt to override a previous
+ // succeeded state
+ return TaskState.SUCCEEDED;
+ }
+ }
+
+ // a successful REDUCE task should not be overridden
// TODO: consider moving it to MapTaskImpl
if (!TaskType.MAP.equals(task.getType())) {
LOG.error("Unexpected event for REDUCE task " + event.getType());
task.internalError(event.getType());
}
- TaskTAttemptEvent attemptEvent = (TaskTAttemptEvent) event;
- TaskAttemptId attemptId = attemptEvent.getTaskAttemptID();
- if(task.successfulAttempt == attemptId) {
- // successful attempt is now killed. reschedule
- // tell the job about the rescheduling
- unSucceed(task);
- task.handleTaskAttemptCompletion(
- attemptId,
- TaskAttemptCompletionEventStatus.KILLED);
- task.eventHandler.handle(new JobMapTaskRescheduledEvent(task.taskId));
- // typically we are here because this map task was run on a bad node
and
- // we want to reschedule it on a different node.
- // Depending on whether there are previous failed attempts or not this
- // can SCHEDULE or RESCHEDULE the container allocate request. If this
- // SCHEDULE's then the dataLocal hosts of this taskAttempt will be used
- // from the map splitInfo. So the bad node might be sent as a location
- // to the RM. But the RM would ignore that just like it would ignore
- // currently pending container requests affinitized to bad nodes.
- task.addAndScheduleAttempt();
- return TaskState.SCHEDULED;
- } else {
- // nothing to do
- return TaskState.SUCCEEDED;
- }
+ // successful attempt is now killed. reschedule
+ // tell the job about the rescheduling
+ unSucceed(task);
+ task.handleTaskAttemptCompletion(attemptId,
+ TaskAttemptCompletionEventStatus.KILLED);
+ task.eventHandler.handle(new JobMapTaskRescheduledEvent(task.taskId));
+ // typically we are here because this map task was run on a bad node and
+ // we want to reschedule it on a different node.
+ // Depending on whether there are previous failed attempts or not this
+ // can SCHEDULE or RESCHEDULE the container allocate request. If this
+ // SCHEDULE's then the dataLocal hosts of this taskAttempt will be used
+ // from the map splitInfo. So the bad node might be sent as a location
+ // to the RM. But the RM would ignore that just like it would ignore
+ // currently pending container requests affinitized to bad nodes.
+ task.addAndScheduleAttempt();
+ return TaskState.SCHEDULED;
}
}
Modified:
hadoop/common/branches/branch-2/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/branch-2/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestMRApp.java?rev=1383425&r1=1383424&r2=1383425&view=diff
==============================================================================
---
hadoop/common/branches/branch-2/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/branch-2/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestMRApp.java
Tue Sep 11 14:13:58 2012
@@ -180,7 +180,7 @@ public class TestMRApp {
@Test
public void testUpdatedNodes() throws Exception {
int runCount = 0;
- MRApp app = new MRAppWithHistory(2, 1, false, this.getClass().getName(),
+ MRApp app = new MRAppWithHistory(2, 2, false, this.getClass().getName(),
true, ++runCount);
Configuration conf = new Configuration();
// after half of the map completion, reduce will start
@@ -189,7 +189,7 @@ public class TestMRApp {
conf.setBoolean(MRJobConfig.JOB_UBERTASK_ENABLE, false);
Job job = app.submit(conf);
app.waitForState(job, JobState.RUNNING);
- Assert.assertEquals("Num tasks not correct", 3, job.getTasks().size());
+ Assert.assertEquals("Num tasks not correct", 4, job.getTasks().size());
Iterator<Task> it = job.getTasks().values().iterator();
Task mapTask1 = it.next();
Task mapTask2 = it.next();
@@ -272,18 +272,19 @@ public class TestMRApp {
// rerun
// in rerun the 1st map will be recovered from previous run
- app = new MRAppWithHistory(2, 1, false, this.getClass().getName(), false,
+ app = new MRAppWithHistory(2, 2, false, this.getClass().getName(), false,
++runCount);
conf = new Configuration();
conf.setBoolean(MRJobConfig.MR_AM_JOB_RECOVERY_ENABLE, true);
conf.setBoolean(MRJobConfig.JOB_UBERTASK_ENABLE, false);
job = app.submit(conf);
app.waitForState(job, JobState.RUNNING);
- Assert.assertEquals("No of tasks not correct", 3, job.getTasks().size());
+ Assert.assertEquals("No of tasks not correct", 4, job.getTasks().size());
it = job.getTasks().values().iterator();
mapTask1 = it.next();
mapTask2 = it.next();
- Task reduceTask = it.next();
+ Task reduceTask1 = it.next();
+ Task reduceTask2 = it.next();
// map 1 will be recovered, no need to send done
app.waitForState(mapTask1, TaskState.SUCCEEDED);
@@ -306,19 +307,36 @@ public class TestMRApp {
Assert.assertEquals("Expecting 1 more completion events for success", 3,
events.length);
- app.waitForState(reduceTask, TaskState.RUNNING);
- TaskAttempt task3Attempt = reduceTask.getAttempts().values().iterator()
+ app.waitForState(reduceTask1, TaskState.RUNNING);
+ app.waitForState(reduceTask2, TaskState.RUNNING);
+
+ TaskAttempt task3Attempt = reduceTask1.getAttempts().values().iterator()
.next();
app.getContext()
.getEventHandler()
.handle(
new TaskAttemptEvent(task3Attempt.getID(),
TaskAttemptEventType.TA_DONE));
- app.waitForState(reduceTask, TaskState.SUCCEEDED);
+ app.waitForState(reduceTask1, TaskState.SUCCEEDED);
+ app.getContext()
+ .getEventHandler()
+ .handle(
+ new TaskAttemptEvent(task3Attempt.getID(),
+ TaskAttemptEventType.TA_KILL));
+ app.waitForState(reduceTask1, TaskState.SUCCEEDED);
+
+ TaskAttempt task4Attempt = reduceTask2.getAttempts().values().iterator()
+ .next();
+ app.getContext()
+ .getEventHandler()
+ .handle(
+ new TaskAttemptEvent(task4Attempt.getID(),
+ TaskAttemptEventType.TA_DONE));
+ app.waitForState(reduceTask2, TaskState.SUCCEEDED);
events = job.getTaskAttemptCompletionEvents(0, 100);
- Assert.assertEquals("Expecting 1 more completion events for success", 4,
- events.length);
+ Assert.assertEquals("Expecting 2 more completion events for reduce
success",
+ 5, events.length);
// job succeeds
app.waitForState(job, JobState.SUCCEEDED);
Modified:
hadoop/common/branches/branch-2/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java
URL:
http://svn.apache.org/viewvc/hadoop/common/branches/branch-2/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java?rev=1383425&r1=1383424&r2=1383425&view=diff
==============================================================================
---
hadoop/common/branches/branch-2/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java
(original)
+++
hadoop/common/branches/branch-2/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java
Tue Sep 11 14:13:58 2012
@@ -84,7 +84,6 @@ public class TestTaskImpl {
private ApplicationId appId;
private TaskSplitMetaInfo taskSplitMetaInfo;
private String[] dataLocations = new String[0];
- private final TaskType taskType = TaskType.MAP;
private AppContext appContext;
private int startCount = 0;
@@ -97,6 +96,7 @@ public class TestTaskImpl {
private class MockTaskImpl extends TaskImpl {
private int taskAttemptCounter = 0;
+ TaskType taskType;
public MockTaskImpl(JobId jobId, int partition,
EventHandler eventHandler, Path remoteJobConfFile, JobConf conf,
@@ -104,11 +104,12 @@ public class TestTaskImpl {
Token<JobTokenIdentifier> jobToken,
Credentials credentials, Clock clock,
Map<TaskId, TaskInfo> completedTasksFromPreviousRun, int startCount,
- MRAppMetrics metrics, AppContext appContext) {
+ MRAppMetrics metrics, AppContext appContext, TaskType taskType) {
super(jobId, taskType , partition, eventHandler,
remoteJobConfFile, conf, taskAttemptListener, committer,
jobToken, credentials, clock,
completedTasksFromPreviousRun, startCount, metrics, appContext);
+ this.taskType = taskType;
}
@Override
@@ -120,7 +121,7 @@ public class TestTaskImpl {
protected TaskAttemptImpl createAttempt() {
MockTaskAttemptImpl attempt = new MockTaskAttemptImpl(getID(),
++taskAttemptCounter,
eventHandler, taskAttemptListener, remoteJobConfFile, partition,
- conf, committer, jobToken, credentials, clock, appContext);
+ conf, committer, jobToken, credentials, clock, appContext, taskType);
taskAttempts.add(attempt);
return attempt;
}
@@ -142,18 +143,20 @@ public class TestTaskImpl {
private float progress = 0;
private TaskAttemptState state = TaskAttemptState.NEW;
private TaskAttemptId attemptId;
+ private TaskType taskType;
public MockTaskAttemptImpl(TaskId taskId, int id, EventHandler
eventHandler,
TaskAttemptListener taskAttemptListener, Path jobFile, int partition,
JobConf conf, OutputCommitter committer,
Token<JobTokenIdentifier> jobToken,
Credentials credentials, Clock clock,
- AppContext appContext) {
+ AppContext appContext, TaskType taskType) {
super(taskId, id, eventHandler, taskAttemptListener, jobFile, partition,
conf,
dataLocations, committer, jobToken, credentials, clock, appContext);
attemptId = Records.newRecord(TaskAttemptId.class);
attemptId.setId(id);
attemptId.setTaskId(taskId);
+ this.taskType = taskType;
}
public TaskAttemptId getAttemptId() {
@@ -162,7 +165,7 @@ public class TestTaskImpl {
@Override
protected Task createRemoteTask() {
- return new MockTask();
+ return new MockTask(taskType);
}
public float getProgress() {
@@ -185,6 +188,11 @@ public class TestTaskImpl {
private class MockTask extends Task {
+ private TaskType taskType;
+ MockTask(TaskType taskType) {
+ this.taskType = taskType;
+ }
+
@Override
public void run(JobConf job, TaskUmbilicalProtocol umbilical)
throws IOException, ClassNotFoundException, InterruptedException {
@@ -193,7 +201,7 @@ public class TestTaskImpl {
@Override
public boolean isMapTask() {
- return true;
+ return (taskType == TaskType.MAP);
}
}
@@ -227,14 +235,15 @@ public class TestTaskImpl {
taskSplitMetaInfo = mock(TaskSplitMetaInfo.class);
when(taskSplitMetaInfo.getLocations()).thenReturn(dataLocations);
- taskAttempts = new ArrayList<MockTaskAttemptImpl>();
-
- mockTask = new MockTaskImpl(jobId, partition, dispatcher.getEventHandler(),
+ taskAttempts = new ArrayList<MockTaskAttemptImpl>();
+ }
+
+ private MockTaskImpl createMockTask(TaskType taskType) {
+ return new MockTaskImpl(jobId, partition, dispatcher.getEventHandler(),
remoteJobConfFile, conf, taskAttemptListener, committer, jobToken,
credentials, clock,
completedTasksFromPreviousRun, startCount,
- metrics, appContext);
-
+ metrics, appContext, taskType);
}
@After
@@ -342,6 +351,7 @@ public class TestTaskImpl {
@Test
public void testInit() {
LOG.info("--- START: testInit ---");
+ mockTask = createMockTask(TaskType.MAP);
assertTaskNewState();
assert(taskAttempts.size() == 0);
}
@@ -352,6 +362,7 @@ public class TestTaskImpl {
*/
public void testScheduleTask() {
LOG.info("--- START: testScheduleTask ---");
+ mockTask = createMockTask(TaskType.MAP);
TaskId taskId = getNewTaskID();
scheduleTaskAttempt(taskId);
}
@@ -362,6 +373,7 @@ public class TestTaskImpl {
*/
public void testKillScheduledTask() {
LOG.info("--- START: testKillScheduledTask ---");
+ mockTask = createMockTask(TaskType.MAP);
TaskId taskId = getNewTaskID();
scheduleTaskAttempt(taskId);
killTask(taskId);
@@ -374,6 +386,7 @@ public class TestTaskImpl {
*/
public void testKillScheduledTaskAttempt() {
LOG.info("--- START: testKillScheduledTaskAttempt ---");
+ mockTask = createMockTask(TaskType.MAP);
TaskId taskId = getNewTaskID();
scheduleTaskAttempt(taskId);
killScheduledTaskAttempt(getLastAttempt().getAttemptId());
@@ -386,6 +399,7 @@ public class TestTaskImpl {
*/
public void testLaunchTaskAttempt() {
LOG.info("--- START: testLaunchTaskAttempt ---");
+ mockTask = createMockTask(TaskType.MAP);
TaskId taskId = getNewTaskID();
scheduleTaskAttempt(taskId);
launchTaskAttempt(getLastAttempt().getAttemptId());
@@ -398,6 +412,7 @@ public class TestTaskImpl {
*/
public void testKillRunningTaskAttempt() {
LOG.info("--- START: testKillRunningTaskAttempt ---");
+ mockTask = createMockTask(TaskType.MAP);
TaskId taskId = getNewTaskID();
scheduleTaskAttempt(taskId);
launchTaskAttempt(getLastAttempt().getAttemptId());
@@ -407,6 +422,7 @@ public class TestTaskImpl {
@Test
public void testTaskProgress() {
LOG.info("--- START: testTaskProgress ---");
+ mockTask = createMockTask(TaskType.MAP);
// launch task
TaskId taskId = getNewTaskID();
@@ -444,6 +460,7 @@ public class TestTaskImpl {
@Test
public void testFailureDuringTaskAttemptCommit() {
+ mockTask = createMockTask(TaskType.MAP);
TaskId taskId = getNewTaskID();
scheduleTaskAttempt(taskId);
launchTaskAttempt(getLastAttempt().getAttemptId());
@@ -469,8 +486,7 @@ public class TestTaskImpl {
assertTaskSucceededState();
}
- @Test
- public void testSpeculativeTaskAttemptSucceedsEvenIfFirstFails() {
+ private void runSpeculativeTaskAttemptSucceedsEvenIfFirstFails(TaskEventType
failEvent) {
TaskId taskId = getNewTaskID();
scheduleTaskAttempt(taskId);
launchTaskAttempt(getLastAttempt().getAttemptId());
@@ -489,11 +505,34 @@ public class TestTaskImpl {
// Now fail the first task attempt, after the second has succeeded
mockTask.handle(new TaskTAttemptEvent(taskAttempts.get(0).getAttemptId(),
- TaskEventType.T_ATTEMPT_FAILED));
+ failEvent));
// The task should still be in the succeeded state
assertTaskSucceededState();
-
+ }
+
+ @Test
+ public void testMapSpeculativeTaskAttemptSucceedsEvenIfFirstFails() {
+ mockTask = createMockTask(TaskType.MAP);
+
runSpeculativeTaskAttemptSucceedsEvenIfFirstFails(TaskEventType.T_ATTEMPT_FAILED);
+ }
+
+ @Test
+ public void testReduceSpeculativeTaskAttemptSucceedsEvenIfFirstFails() {
+ mockTask = createMockTask(TaskType.REDUCE);
+
runSpeculativeTaskAttemptSucceedsEvenIfFirstFails(TaskEventType.T_ATTEMPT_FAILED);
+ }
+
+ @Test
+ public void testMapSpeculativeTaskAttemptSucceedsEvenIfFirstIsKilled() {
+ mockTask = createMockTask(TaskType.MAP);
+
runSpeculativeTaskAttemptSucceedsEvenIfFirstFails(TaskEventType.T_ATTEMPT_KILLED);
+ }
+
+ @Test
+ public void testReduceSpeculativeTaskAttemptSucceedsEvenIfFirstIsKilled() {
+ mockTask = createMockTask(TaskType.REDUCE);
+
runSpeculativeTaskAttemptSucceedsEvenIfFirstFails(TaskEventType.T_ATTEMPT_KILLED);
}
}
