hadoop-common-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rohithsharm...@apache.org
Subject hadoop git commit: MAPREDUCE-6485. Create a new task attempt with failed map task priority if in-progress attempts are unassigned. (Xianyin Xin via rohithsharmaks)
Date Fri, 02 Oct 2015 15:05:45 GMT
Repository: hadoop
Updated Branches:
  refs/heads/trunk fd026f535 -> 439f43ad3


MAPREDUCE-6485. Create a new task attempt with failed map task priority if in-progress attempts
are unassigned. (Xianyin Xin via rohithsharmaks)


Project: http://git-wip-us.apache.org/repos/asf/hadoop/repo
Commit: http://git-wip-us.apache.org/repos/asf/hadoop/commit/439f43ad
Tree: http://git-wip-us.apache.org/repos/asf/hadoop/tree/439f43ad
Diff: http://git-wip-us.apache.org/repos/asf/hadoop/diff/439f43ad

Branch: refs/heads/trunk
Commit: 439f43ad3defbac907eda2d139a793f153544430
Parents: fd026f5
Author: rohithsharmaks <rohithsharmaks@apache.org>
Authored: Fri Oct 2 20:34:34 2015 +0530
Committer: rohithsharmaks <rohithsharmaks@apache.org>
Committed: Fri Oct 2 20:34:34 2015 +0530

----------------------------------------------------------------------
 hadoop-mapreduce-project/CHANGES.txt            |  3 +
 .../v2/app/job/impl/TaskAttemptImpl.java        |  7 +-
 .../mapreduce/v2/app/job/impl/TaskImpl.java     | 20 ++++-
 .../mapreduce/v2/app/job/impl/TestTaskImpl.java | 94 +++++++++++++++++++-
 4 files changed, 117 insertions(+), 7 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/hadoop/blob/439f43ad/hadoop-mapreduce-project/CHANGES.txt
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/CHANGES.txt b/hadoop-mapreduce-project/CHANGES.txt
index e05688d..781cb76 100644
--- a/hadoop-mapreduce-project/CHANGES.txt
+++ b/hadoop-mapreduce-project/CHANGES.txt
@@ -588,6 +588,9 @@ Release 2.8.0 - UNRELEASED
    MAPREDUCE-6494. Permission issue when running archive-logs tool as
    different users (rkanter)
 
+   MAPREDUCE-6485. Create a new task attempt with failed map task priority
+   if in-progress attempts are unassigned. (Xianyin Xin via rohithsharmaks)
+
 Release 2.7.2 - UNRELEASED
 
   INCOMPATIBLE CHANGES

http://git-wip-us.apache.org/repos/asf/hadoop/blob/439f43ad/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java
old mode 100644
new mode 100755
index db4f585..a532141
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskAttemptImpl.java
@@ -1333,7 +1333,7 @@ public abstract class TaskAttemptImpl implements
     return attemptState;
   }
 
-  private static TaskAttemptState getExternalState(
+  protected static TaskAttemptState getExternalState(
       TaskAttemptStateInternal smState) {
     switch (smState) {
     case ASSIGNED:
@@ -1365,6 +1365,11 @@ public abstract class TaskAttemptImpl implements
     }
   }
 
+  // check whether the attempt is assigned if container is not null
+  boolean isContainerAssigned() {
+    return container != null;
+  }
+
   //always called in write lock
   private void setFinishTime() {
     //set the finish time only if launch time is set

http://git-wip-us.apache.org/repos/asf/hadoop/blob/439f43ad/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskImpl.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskImpl.java
old mode 100644
new mode 100755
index 5f5f300..a392837
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskImpl.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskImpl.java
@@ -1057,9 +1057,21 @@ public abstract class TaskImpl implements Task, EventHandler<TaskEvent>
{
             TaskAttemptCompletionEventStatus.FAILED);
         // we don't need a new event if we already have a spare
         task.inProgressAttempts.remove(taskAttemptId);
-        if (task.inProgressAttempts.size() == 0
-            && task.successfulAttempt == null) {
-          task.addAndScheduleAttempt(Avataar.VIRGIN);
+        if (task.successfulAttempt == null) {
+          boolean shouldAddNewAttempt = true;
+          if (task.inProgressAttempts.size() > 0) {
+            // if not all of the inProgressAttempts are hanging for resource
+            for (TaskAttemptId attemptId : task.inProgressAttempts) {
+              if (((TaskAttemptImpl) task.getAttempt(attemptId))
+                  .isContainerAssigned()) {
+                shouldAddNewAttempt = false;
+                break;
+              }
+            }
+          }
+          if (shouldAddNewAttempt) {
+            task.addAndScheduleAttempt(Avataar.VIRGIN);
+          }
         }
       } else {
         task.handleTaskAttemptCompletion(
@@ -1080,7 +1092,7 @@ public abstract class TaskImpl implements Task, EventHandler<TaskEvent>
{
             taskFailedEvent));
         } else {
           LOG.debug("Not generating HistoryFinish event since start event not" +
-          		" generated for task: " + task.getID());
+              " generated for task: " + task.getID());
         }
         task.eventHandler.handle(
             new JobTaskEvent(task.taskId, TaskState.FAILED));

http://git-wip-us.apache.org/repos/asf/hadoop/blob/439f43ad/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java
----------------------------------------------------------------------
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java
b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java
old mode 100644
new mode 100755
index ae8a797..af51817
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TestTaskImpl.java
@@ -50,6 +50,8 @@ import org.apache.hadoop.mapreduce.v2.app.AppContext;
 import org.apache.hadoop.mapreduce.v2.app.TaskAttemptListener;
 import org.apache.hadoop.mapreduce.v2.app.job.TaskAttempt;
 import org.apache.hadoop.mapreduce.v2.app.job.TaskStateInternal;
+import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEvent;
+import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEventType;
 import org.apache.hadoop.mapreduce.v2.app.job.event.TaskEvent;
 import org.apache.hadoop.mapreduce.v2.app.job.event.TaskEventType;
 import org.apache.hadoop.mapreduce.v2.app.job.event.TaskTAttemptEvent;
@@ -57,6 +59,7 @@ import org.apache.hadoop.mapreduce.v2.app.metrics.MRAppMetrics;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.event.Event;
 import org.apache.hadoop.yarn.event.EventHandler;
 import org.apache.hadoop.yarn.event.InlineDispatcher;
 import org.apache.hadoop.yarn.util.Clock;
@@ -140,6 +143,8 @@ public class TestTaskImpl {
 
     private float progress = 0;
     private TaskAttemptState state = TaskAttemptState.NEW;
+    boolean rescheduled = false;
+    boolean containerAssigned = false;
     private TaskType taskType;
     private Counters attemptCounters = TaskAttemptImpl.EMPTY_COUNTERS;
 
@@ -153,6 +158,15 @@ public class TestTaskImpl {
       this.taskType = taskType;
     }
 
+    public void assignContainer() {
+      containerAssigned = true;
+    }
+
+    @Override
+    boolean isContainerAssigned() {
+      return containerAssigned;
+    }
+
     public TaskAttemptId getAttemptId() {
       return getID();
     }
@@ -173,11 +187,20 @@ public class TestTaskImpl {
     public void setState(TaskAttemptState state) {
       this.state = state;
     }
-    
+
+    @Override
     public TaskAttemptState getState() {
       return state;
     }
 
+    public boolean getRescheduled() {
+      return this.rescheduled;
+    }
+
+    public void setRescheduled(boolean rescheduled) {
+      this.rescheduled = rescheduled;
+    }
+
     @Override
     public Counters getCounters() {
       return attemptCounters;
@@ -279,7 +302,9 @@ public class TestTaskImpl {
   private void launchTaskAttempt(TaskAttemptId attemptId) {
     mockTask.handle(new TaskTAttemptEvent(attemptId, 
         TaskEventType.T_ATTEMPT_LAUNCHED));
-    assertTaskRunningState();    
+    ((MockTaskAttemptImpl)(mockTask.getAttempt(attemptId)))
+        .assignContainer();
+    assertTaskRunningState();
   }
   
   private void commitTaskAttempt(TaskAttemptId attemptId) {
@@ -708,6 +733,71 @@ public class TestTaskImpl {
     assertEquals(TaskState.FAILED, mockTask.getState());
   }
 
+  private class PartialAttemptEventHandler implements EventHandler {
+
+    @Override
+    public void handle(Event event) {
+      if (event instanceof TaskAttemptEvent)
+        if (event.getType() == TaskAttemptEventType.TA_RESCHEDULE) {
+          TaskAttempt attempt = mockTask.getAttempt(((TaskAttemptEvent) event).getTaskAttemptID());
+          ((MockTaskAttemptImpl)attempt).setRescheduled(true);
+        }
+    }
+  }
+
+  @Test
+  public void testFailedTransitionWithHangingSpeculativeMap() {
+    mockTask = new MockTaskImpl(jobId, partition, new PartialAttemptEventHandler(),
+        remoteJobConfFile, conf, taskAttemptListener, jobToken,
+        credentials, clock, startCount, metrics, appContext, TaskType.MAP) {
+      @Override
+      protected int getMaxAttempts() {
+        return 4;
+      }
+    };
+
+    // start a new task, schedule and launch a new attempt
+    TaskId taskId = getNewTaskID();
+    scheduleTaskAttempt(taskId);
+    launchTaskAttempt(getLastAttempt().getAttemptId());
+
+    // add a speculative attempt(#2), but not launch it
+    mockTask.handle(new TaskTAttemptEvent(getLastAttempt().getAttemptId(),
+        TaskEventType.T_ADD_SPEC_ATTEMPT));
+
+    // have the first attempt(#1) fail, verify task still running since the
+    // max attempts is 4
+    MockTaskAttemptImpl taskAttempt = taskAttempts.get(0);
+    taskAttempt.setState(TaskAttemptState.FAILED);
+    mockTask.handle(new TaskTAttemptEvent(taskAttempt.getAttemptId(),
+        TaskEventType.T_ATTEMPT_FAILED));
+    assertEquals(TaskState.RUNNING, mockTask.getState());
+
+    // verify a new attempt(#3) added because the speculative attempt(#2)
+    // is hanging
+    assertEquals(3, taskAttempts.size());
+
+    // verify the speculative attempt(#2) is not a rescheduled attempt
+    assertEquals(false, taskAttempts.get(1).getRescheduled());
+
+    // verify the third attempt is a rescheduled attempt
+    assertEquals(true, taskAttempts.get(2).getRescheduled());
+
+    // now launch the latest attempt(#3) and set the internal state to running
+    launchTaskAttempt(getLastAttempt().getAttemptId());
+
+    // have the speculative attempt(#2) fail, verify task still since it
+    // hasn't reach the max attempts which is 4
+    MockTaskAttemptImpl taskAttempt1 = taskAttempts.get(1);
+    taskAttempt1.setState(TaskAttemptState.FAILED);
+    mockTask.handle(new TaskTAttemptEvent(taskAttempt1.getAttemptId(),
+        TaskEventType.T_ATTEMPT_FAILED));
+    assertEquals(TaskState.RUNNING, mockTask.getState());
+
+    // verify there's no new attempt added because of the running attempt(#3)
+    assertEquals(3, taskAttempts.size());
+  }
+
   @Test
   public void testCountersWithSpeculation() {
     mockTask = new MockTaskImpl(jobId, partition, dispatcher.getEventHandler(),


Mime
View raw message