You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by bi...@apache.org on 2014/10/16 00:18:55 UTC
git commit: TEZ-1665. DAGScheduler should provide a priority range
instead of an exact priority (bikas)jps
Repository: tez
Updated Branches:
refs/heads/master 7be5830a9 -> 0a31652b9
TEZ-1665. DAGScheduler should provide a priority range instead of an exact priority (bikas)jps
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/0a31652b
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/0a31652b
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/0a31652b
Branch: refs/heads/master
Commit: 0a31652b9b68f48e936618e2c2edf9a402314da1
Parents: 7be5830
Author: Bikas Saha <bi...@apache.org>
Authored: Wed Oct 15 15:16:52 2014 -0700
Committer: Bikas Saha <bi...@apache.org>
Committed: Wed Oct 15 15:16:52 2014 -0700
----------------------------------------------------------------------
CHANGES.txt | 2 +
.../org/apache/tez/dag/app/dag/TaskAttempt.java | 2 -
.../app/dag/event/TaskAttemptEventSchedule.java | 19 +-
.../tez/dag/app/dag/impl/DAGSchedulerMRR.java | 334 -------------------
.../app/dag/impl/DAGSchedulerNaturalOrder.java | 19 +-
.../tez/dag/app/dag/impl/TaskAttemptImpl.java | 15 +-
.../app/rm/AMSchedulerEventTALaunchRequest.java | 7 +-
.../dag/app/rm/TaskSchedulerEventHandler.java | 5 +-
.../tez/dag/app/dag/impl/TestDAGImpl.java | 54 ---
.../tez/dag/app/dag/impl/TestDAGScheduler.java | 137 +-------
.../tez/dag/app/dag/impl/TestTaskAttempt.java | 177 ++++------
.../tez/dag/app/rm/TestContainerReuse.java | 2 +-
12 files changed, 106 insertions(+), 667 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/0a31652b/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index b70d234..e533c3b 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -28,6 +28,8 @@ ALL CHANGES:
TEZ-1083. Enable IFile RLE for DefaultSorter.
TEZ-1637. Improved shuffle error handling across NM restarts.
TEZ-1479. Disambiguate (refactor) between ShuffleInputEventHandlers and Fetchers.
+ TEZ-1665. DAGScheduler should provide a priority range instead of an exact
+ priority
Release 0.5.1: 2014-10-02
http://git-wip-us.apache.org/repos/asf/tez/blob/0a31652b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/TaskAttempt.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/TaskAttempt.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/TaskAttempt.java
index 0563037..f30fc5c 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/TaskAttempt.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/TaskAttempt.java
@@ -124,8 +124,6 @@ public interface TaskAttempt {
public Task getTask();
- public boolean getIsRescheduled();
-
TaskAttemptState restoreFromEvent(HistoryEvent event);
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a31652b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/TaskAttemptEventSchedule.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/TaskAttemptEventSchedule.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/TaskAttemptEventSchedule.java
index a7d4c9a..5c02db9 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/TaskAttemptEventSchedule.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/TaskAttemptEventSchedule.java
@@ -17,22 +17,25 @@
package org.apache.tez.dag.app.dag.event;
-import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.tez.dag.records.TezTaskAttemptID;
-
-
public class TaskAttemptEventSchedule extends TaskAttemptEvent {
- private final Priority priority;
+ private final int priLowLimit;
+ private final int priHighLimit;
- public TaskAttemptEventSchedule(TezTaskAttemptID id, Priority priority) {
+ public TaskAttemptEventSchedule(TezTaskAttemptID id, int lowLimit, int highLimit) {
super(id, TaskAttemptEventType.TA_SCHEDULE);
- this.priority = priority;
+ this.priHighLimit = highLimit;
+ this.priLowLimit = lowLimit;
}
- public Priority getPriority() {
- return this.priority;
+ public int getPriorityLowLimit() {
+ return priLowLimit;
+ }
+
+ public int getPriorityHighLimit() {
+ return priHighLimit;
}
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a31652b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGSchedulerMRR.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGSchedulerMRR.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGSchedulerMRR.java
deleted file mode 100644
index 914a562..0000000
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGSchedulerMRR.java
+++ /dev/null
@@ -1,334 +0,0 @@
-/**
-* Licensed to the Apache Software Foundation (ASF) under one
-* or more contributor license agreements. See the NOTICE file
-* distributed with this work for additional information
-* regarding copyright ownership. The ASF licenses this file
-* to you under the Apache License, Version 2.0 (the
-* "License"); you may not use this file except in compliance
-* with the License. You may obtain a copy of the License at
-*
-* http://www.apache.org/licenses/LICENSE-2.0
-*
-* Unless required by applicable law or agreed to in writing, software
-* distributed under the License is distributed on an "AS IS" BASIS,
-* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-* See the License for the specific language governing permissions and
-* limitations under the License.
-*/
-
-package org.apache.tez.dag.app.dag.impl;
-
-import java.util.HashSet;
-import java.util.LinkedList;
-import java.util.List;
-import java.util.Set;
-
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.yarn.api.records.Priority;
-import org.apache.hadoop.yarn.api.records.Resource;
-import org.apache.hadoop.yarn.event.EventHandler;
-import org.apache.tez.dag.api.TezUncheckedException;
-import org.apache.tez.dag.app.dag.DAG;
-import org.apache.tez.dag.app.dag.DAGScheduler;
-import org.apache.tez.dag.app.dag.TaskAttempt;
-import org.apache.tez.dag.app.dag.Vertex;
-import org.apache.tez.dag.app.dag.event.DAGEventSchedulerUpdate;
-import org.apache.tez.dag.app.dag.event.DAGEventSchedulerUpdateTAAssigned;
-import org.apache.tez.dag.app.dag.event.TaskAttemptEventSchedule;
-import org.apache.tez.dag.app.rm.TaskSchedulerEventHandler;
-import org.apache.tez.dag.records.TezTaskID;
-
-@SuppressWarnings("rawtypes")
-public class DAGSchedulerMRR implements DAGScheduler {
-
- private static final Log LOG = LogFactory.getLog(DAGSchedulerMRR.class);
-
- private final DAG dag;
- private final TaskSchedulerEventHandler taskScheduler;
- private final EventHandler handler;
-
- private final float minReservedShuffleResource;
-
- private Vertex currentPartitioner = null;
- private Vertex currentShuffler = null;
- private int currentShufflerDepth = 0;
-
- int numShuffleTasksScheduled = 0;
- List<TaskAttempt> pendingShuffleTasks = new LinkedList<TaskAttempt>();
- Set<TezTaskID> unassignedShuffleTasks = new HashSet<TezTaskID>();
- Resource realShufflerResource = null;
-
- Set<TezTaskID> unassignedPartitionTasks = new HashSet<TezTaskID>();
- Resource realPartitionerResource = null;
-
- public DAGSchedulerMRR(DAG dag, EventHandler dispatcher,
- TaskSchedulerEventHandler taskScheduler, float minReservedShuffleResource) {
- this.dag = dag;
- this.handler = dispatcher;
- this.taskScheduler = taskScheduler;
- this.minReservedShuffleResource = minReservedShuffleResource;
- }
-
- @Override
- public void vertexCompleted(Vertex vertex) {
- if(currentPartitioner != null) {
- if(vertex != currentPartitioner) {
- String message = vertex.getVertexId() + " finished. Expecting"
- + " current partitioner " + currentPartitioner.getVertexId()
- + " to finish.";
- LOG.fatal(message);
- throw new TezUncheckedException(message);
- }
- LOG.info("Current partitioner " + currentPartitioner.getVertexId()
- + " is completed. "
- + (currentShuffler!=null ?
- currentShuffler.getVertexId() + " is new partitioner":
- "No current shuffler to replace the partitioner"));
- currentPartitioner = currentShuffler;
- assert unassignedPartitionTasks.isEmpty();
- unassignedPartitionTasks.addAll(unassignedShuffleTasks);
- unassignedShuffleTasks.clear();
- realPartitionerResource = realShufflerResource;
- realShufflerResource = null;
- currentShuffler = null;
- // schedule all pending shuffle tasks
- schedulePendingShuffles(pendingShuffleTasks.size());
- assert pendingShuffleTasks.isEmpty();
- numShuffleTasksScheduled = 0;
- }
-
- }
-
- @Override
- public void scheduleTask(DAGEventSchedulerUpdate event) {
- TaskAttempt attempt = event.getAttempt();
- Vertex vertex = dag.getVertex(attempt.getVertexID());
- int vertexDistanceFromRoot = vertex.getDistanceFromRoot();
-
- LOG.info("Schedule task: " + attempt.getID());
-
- if(currentPartitioner == null) {
- // no partitioner. so set it.
- currentPartitioner = vertex;
- currentShufflerDepth = vertexDistanceFromRoot;
- assert realPartitionerResource == null;
- Resource partitionerResource = currentPartitioner.getTaskResource();
- realPartitionerResource = Resource.newInstance(
- partitionerResource.getMemory(),
- partitionerResource.getVirtualCores());
- LOG.info(vertex.getVertexId() + " is new partitioner at depth "
- + vertexDistanceFromRoot);
- } else if (currentShuffler == null &&
- vertexDistanceFromRoot > currentShufflerDepth) {
- // vertex not a partitioner. no shuffler set. has more depth than current
- // shuffler. this must be the new shuffler.
- currentShuffler = vertex;
- currentShufflerDepth = vertexDistanceFromRoot;
- assert realShufflerResource == null;
- Resource shufflerResource = currentShuffler.getTaskResource();
- realShufflerResource = Resource.newInstance(
- shufflerResource.getMemory(),
- shufflerResource.getVirtualCores());
- LOG.info(vertex.getVertexId() + " is new shuffler at depth "
- + currentShufflerDepth);
- }
-
- if(currentShuffler == vertex) {
- pendingShuffleTasks.add(attempt);
- unassignedShuffleTasks.add(attempt.getTaskID());
- schedulePendingShuffles(getNumShufflesToSchedule());
- return;
- }
-
- if(currentPartitioner == vertex) {
- unassignedPartitionTasks.add(attempt.getTaskID());
- }
-
- // sanity check
- // task should be a partitioner, a shuffler or a retry of an ancestor
- if(currentPartitioner != vertex && currentShuffler != vertex &&
- vertexDistanceFromRoot >= currentPartitioner.getDistanceFromRoot()) {
- String message = vertex.getVertexId() + " is neither the "
- + " current partitioner: " + currentPartitioner.getVertexId()
- + " nor the current shuffler: " + currentShuffler.getVertexId();
- LOG.fatal(message);
- throw new TezUncheckedException(message);
- }
-
- scheduleTaskAttempt(attempt);
- }
-
- @Override
- public void taskScheduled(DAGEventSchedulerUpdateTAAssigned event) {
- TaskAttempt attempt = event.getAttempt();
- Vertex vertex = dag.getVertex(attempt.getVertexID());
- LOG.info("Task assigned: " + attempt.getID() + " Vertex: Total:"
- + vertex.getTotalTasks() + " succeeded: " + vertex.getSucceededTasks()
- + " Resource: " + event.getContainer().getResource().getMemory());
-
- if (currentPartitioner == vertex) {
- unassignedPartitionTasks.remove(attempt.getTaskID());
- Resource resource = event.getContainer().getResource();
- if(resource.getMemory() > realPartitionerResource.getMemory()) {
- realPartitionerResource.setMemory(resource.getMemory());
- }
- } else if (currentShuffler == vertex) {
- unassignedShuffleTasks.remove(attempt.getTaskID());
- Resource resource = event.getContainer().getResource();
- if(resource.getMemory() > realShufflerResource.getMemory()) {
- realShufflerResource.setMemory(resource.getMemory());
- }
- }
- schedulePendingShuffles(getNumShufflesToSchedule());
- }
-
- @Override
- public void taskSucceeded(DAGEventSchedulerUpdate event) {
- TaskAttempt attempt = event.getAttempt();
- Vertex vertex = dag.getVertex(attempt.getVertexID());
- LOG.info("Task succeeded: " + attempt.getID() + " Vertex: Total:" + vertex.getTotalTasks() +
- " succeeded: " + vertex.getSucceededTasks());
-
- // resources now available. try to schedule pending shuffles
- schedulePendingShuffles(getNumShufflesToSchedule());
- }
-
- int getNumShufflesToSchedule() {
- assert currentPartitioner != null;
-
- if(pendingShuffleTasks.isEmpty()) {
- return 0;
- }
-
- if(unassignedPartitionTasks.isEmpty()) {
- LOG.info("All partitioners assigned. Scheduling all shufflers.");
- return pendingShuffleTasks.size();
- }
-
- assert currentShuffler != null;
-
- // get total resource limit
- Resource totalResources = taskScheduler.getTotalResources();
- Resource freeResources = taskScheduler.getAvailableResources();
- int totalMem = totalResources.getMemory();
- int freeMem = freeResources.getMemory();
- int partitionerTaskMem = realPartitionerResource.getMemory();
- int shufflerTaskMem = realShufflerResource.getMemory();
- int shufflerMemAssigned = shufflerTaskMem * numShuffleTasksScheduled;
-
- // get resources needed by partitioner
- int numPartitioners = currentPartitioner.getTotalTasks();
- int numPartionersSucceeded = currentPartitioner.getSucceededTasks();
- int numPartionersLeft = numPartitioners - numPartionersSucceeded;
- int partitionerMemNeeded = numPartionersLeft * partitionerTaskMem;
-
- // find leftover resources for shuffler
- int shufflerMemLeft = totalMem - partitionerMemNeeded;
-
- int maxShufflerMem = (int) (totalMem *
- (Math.min(minReservedShuffleResource,
- numPartionersSucceeded/(float)numPartitioners)));
-
- if(shufflerMemLeft < maxShufflerMem) {
- shufflerMemLeft = maxShufflerMem;
- }
-
- shufflerMemLeft -= shufflerMemAssigned;
-
- LOG.info("TotalMem: " + totalMem +
- " Headroom: " + freeMem +
- " PartitionerTaskMem: " + partitionerTaskMem +
- " ShufflerTaskMem: " + shufflerTaskMem +
- " MaxShuffleMem: " + maxShufflerMem +
- " PartitionerMemNeeded:" + partitionerMemNeeded +
- " ShufflerMemAssigned: " + shufflerMemAssigned +
- " ShufflerMemLeft: " + shufflerMemLeft +
- " Pending shufflers: " + pendingShuffleTasks.size());
-
- if(shufflerMemLeft < 0) {
- // not enough resource to schedule a shuffler
- return 0;
- }
-
- if(shufflerTaskMem == 0) {
- return pendingShuffleTasks.size();
- }
-
- int shufflersToSchedule = shufflerMemLeft / shufflerTaskMem;
- shufflerMemAssigned += shufflerTaskMem * shufflersToSchedule;
-
- if(totalMem - shufflerMemAssigned < partitionerTaskMem) {
- // safety check when reduce ramp up limit is aggressively high
- LOG.info("Not scheduling more shufflers as it starves partitioners");
- return 0;
- }
-
- return shufflersToSchedule;
- }
-
- void schedulePendingShuffles(int scheduleCount) {
- while(!pendingShuffleTasks.isEmpty() && scheduleCount>0) {
- --scheduleCount;
- TaskAttempt shuffleAttempt = pendingShuffleTasks.remove(0);
- scheduleTaskAttempt(shuffleAttempt);
- if(!shuffleAttempt.getIsRescheduled()) {
- // dont double count same shuffle task
- numShuffleTasksScheduled++;
- }
- }
- }
-
- void scheduleTaskAttempt(TaskAttempt attempt) {
- boolean reOrderPriority = false;
- Vertex vertex = dag.getVertex(attempt.getVertexID());
- int vertexDistanceFromRoot = vertex.getDistanceFromRoot();
-
- // natural priority. Handles failures and retries.
- int priority = (vertexDistanceFromRoot + 1) * 3;
-
- if(currentShuffler == vertex) {
- assert currentPartitioner != null;
- // assign higher priority only if its needed. If all partitioners are done
- // then no need to do so.
- // TODO fix with assigned instead of succeeded
- if (!unassignedPartitionTasks.isEmpty()) {
- // current shuffler vertex to be scheduled while current partitioner is
- // still running. This needs to be higher priority or else it wont get
- // allocated. This higher priority will be lower than the priority of a
- // partitioner task that is a retry. so retries are safe.
- // assign special priority
- reOrderPriority = true;
- }
- }
-
- if(reOrderPriority) {
- // special priority for current reducers while current partitioners are
- // still running. Schedule at priority one higher than natural priority
- // of previous vertex.
- priority -= 4; // this == (partitionerDepth+1)*3 - 1
- } else {
- if(attempt.getIsRescheduled()) {
- // higher priority for retries of failed attempts. Only makes sense in
- // case the task is faulty and we want to retry before other tasks in
- // the same vertex to fail fast. But looks like this may happen also for
- // other cases like retry because outputs were unavailable.
- priority -= 2;
- }
- }
-
- LOG.info("Scheduling " + attempt.getID() +
- " with depth " + vertexDistanceFromRoot +
- " at priority " + priority);
-
- TaskAttemptEventSchedule attemptEvent = new TaskAttemptEventSchedule(
- attempt.getID(), Priority.newInstance(priority));
-
- sendEvent(attemptEvent);
- }
-
- @SuppressWarnings("unchecked")
- void sendEvent(TaskAttemptEventSchedule event) {
- handler.handle(event);
- }
-}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a31652b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGSchedulerNaturalOrder.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGSchedulerNaturalOrder.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGSchedulerNaturalOrder.java
index fc52344..253d410 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGSchedulerNaturalOrder.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGSchedulerNaturalOrder.java
@@ -20,7 +20,6 @@ package org.apache.tez.dag.app.dag.impl;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.tez.dag.app.dag.DAG;
import org.apache.tez.dag.app.dag.DAGScheduler;
@@ -55,20 +54,16 @@ public class DAGSchedulerNaturalOrder implements DAGScheduler {
int vertexDistanceFromRoot = vertex.getDistanceFromRoot();
// natural priority. Handles failures and retries.
- int priority = (vertexDistanceFromRoot + 1) * 2;
-
- if(attempt.getIsRescheduled()) {
- // higher priority for retries of failed attempts. Only makes sense in
- // case the task is faulty and we want to retry before other tasks in
- // the same vertex to fail fast. But looks like this may happen also for
- // other cases like retry because outputs were unavailable.
- priority--;
- }
+ int priorityLowLimit = (vertexDistanceFromRoot + 1) * 3;
+ int priorityHighLimit = priorityLowLimit - 2;
- LOG.info("Scheduling " + attempt.getID() + " at priority " + priority);
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Scheduling " + attempt.getID() + " between priorityLow: " + priorityLowLimit
+ + " and priorityHigh: " + priorityHighLimit);
+ }
TaskAttemptEventSchedule attemptEvent = new TaskAttemptEventSchedule(
- attempt.getID(), Priority.newInstance(priority));
+ attempt.getID(), priorityLowLimit, priorityHighLimit);
sendEvent(attemptEvent);
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a31652b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
index b8430cc..56dd303 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
@@ -778,11 +778,6 @@ public class TaskAttemptImpl implements TaskAttempt,
}
@Override
- public boolean getIsRescheduled() {
- return isRescheduled;
- }
-
- @Override
public TaskAttemptState restoreFromEvent(HistoryEvent historyEvent) {
switch (historyEvent.getEventType()) {
case TASK_ATTEMPT_STARTED:
@@ -1087,11 +1082,19 @@ public class TaskAttemptImpl implements TaskAttempt,
LOG.debug("Asking for container launch with taskAttemptContext: "
+ remoteTaskSpec);
}
+
// Send out a launch request to the scheduler.
+ int priority;
+ if (ta.isRescheduled) {
+ // higher priority for rescheduled attempts
+ priority = scheduleEvent.getPriorityHighLimit();
+ } else {
+ priority = (scheduleEvent.getPriorityHighLimit() + scheduleEvent.getPriorityLowLimit()) / 2;
+ }
AMSchedulerEventTALaunchRequest launchRequestEvent = new AMSchedulerEventTALaunchRequest(
ta.attemptId, ta.taskResource, remoteTaskSpec, ta, locationHint,
- scheduleEvent.getPriority(), ta.containerContext);
+ priority, ta.containerContext);
ta.sendEvent(launchRequestEvent);
}
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a31652b/tez-dag/src/main/java/org/apache/tez/dag/app/rm/AMSchedulerEventTALaunchRequest.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/rm/AMSchedulerEventTALaunchRequest.java b/tez-dag/src/main/java/org/apache/tez/dag/app/rm/AMSchedulerEventTALaunchRequest.java
index d7ebed9..5c4d43c 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/rm/AMSchedulerEventTALaunchRequest.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/rm/AMSchedulerEventTALaunchRequest.java
@@ -17,7 +17,6 @@
package org.apache.tez.dag.app.rm;
-import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.tez.dag.api.TaskLocationHint;
import org.apache.tez.dag.app.ContainerContext;
@@ -31,7 +30,7 @@ public class AMSchedulerEventTALaunchRequest extends AMSchedulerEvent {
//.... Maybe have the Container talk to the TaskAttempt to pull in the remote task.
private final TezTaskAttemptID attemptId;
- private final Priority priority;
+ private final int priority;
private final Resource capability;
private final TaskLocationHint locationHint;
private final ContainerContext containerContext;
@@ -42,7 +41,7 @@ public class AMSchedulerEventTALaunchRequest extends AMSchedulerEvent {
public AMSchedulerEventTALaunchRequest(TezTaskAttemptID attemptId,
Resource capability,
TaskSpec remoteTaskSpec, TaskAttempt ta,
- TaskLocationHint locationHint, Priority priority,
+ TaskLocationHint locationHint, int priority,
ContainerContext containerContext) {
super(AMSchedulerEventType.S_TA_LAUNCH_REQUEST);
this.attemptId = attemptId;
@@ -66,7 +65,7 @@ public class AMSchedulerEventTALaunchRequest extends AMSchedulerEvent {
return locationHint;
}
- public Priority getPriority() {
+ public int getPriority() {
return priority;
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a31652b/tez-dag/src/main/java/org/apache/tez/dag/app/rm/TaskSchedulerEventHandler.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/rm/TaskSchedulerEventHandler.java b/tez-dag/src/main/java/org/apache/tez/dag/app/rm/TaskSchedulerEventHandler.java
index b1b551a..508472a 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/rm/TaskSchedulerEventHandler.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/rm/TaskSchedulerEventHandler.java
@@ -36,6 +36,7 @@ import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.ContainerStatus;
import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
import org.apache.hadoop.yarn.api.records.NodeReport;
+import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.event.Event;
import org.apache.hadoop.yarn.event.EventHandler;
@@ -265,7 +266,7 @@ public class TaskSchedulerEventHandler extends AbstractService
taskScheduler.allocateTask(taskAttempt,
event.getCapability(),
locationHint.getAffinitizedContainer(),
- event.getPriority(),
+ Priority.newInstance(event.getPriority()),
event.getContainerContext(),
event);
return;
@@ -282,7 +283,7 @@ public class TaskSchedulerEventHandler extends AbstractService
event.getCapability(),
hosts,
racks,
- event.getPriority(),
+ Priority.newInstance(event.getPriority()),
event.getContainerContext(),
event);
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a31652b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGImpl.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGImpl.java
index 2d7cf65..9430526 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGImpl.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGImpl.java
@@ -1212,58 +1212,4 @@ public class TestDAGImpl {
Assert.assertEquals(1, dagFinishEventHandler.dagFinishEvents);
}
- @Test(timeout = 5000)
- public void testDiagnosticUpdates() {
- // FIXME need to implement
- }
-
- @Test(timeout = 5000)
- public void testCounterUpdates() {
- // FIXME need to implement
- }
-
- @Ignore
- @SuppressWarnings("unchecked")
- @Test(timeout = 5000)
- public void testOutofBandFailureForMRRScheduler() {
- initDAG(mrrDag);
- dispatcher.await();
- Assert.assertTrue(mrrDag.dagScheduler instanceof DAGSchedulerMRR);
- startDAG(mrrDag);
- dispatcher.await();
-
- TaskAttempt attempt = mock(TaskAttempt.class);
- doReturn(
- mrrDag.getVertex("vertex1").getVertexId()).when(attempt).getVertexID();
-
- DAGEventSchedulerUpdate scheduleEvent =
- new DAGEventSchedulerUpdate(UpdateType.TA_SCHEDULE, attempt);
- dispatcher.getEventHandler().handle(scheduleEvent);
- dispatcher.await();
-
- dispatcher.getEventHandler().handle(new DAGEventVertexCompleted(
- mrrDag.getVertex("vertex2").getVertexId(), VertexState.FAILED));
- dispatcher.await();
-
- Assert.assertEquals(DAGState.FAILED, mrrDag.getState());
- }
-
- @SuppressWarnings("unchecked")
- @Test(timeout = 5000)
- public void testDAGKillInternalError() {
- initDAG(dag);
- startDAG(dag);
- dispatcher.await();
-
- dispatcher.getEventHandler().handle(new DAGEvent(dagId,
- DAGEventType.INTERNAL_ERROR));
-
- dispatcher.await();
- Assert.assertEquals(DAGState.ERROR, dag.getState());
- Assert.assertEquals(0, dag.getSuccessfulVertices());
- Assert.assertEquals(dag.getVertex(TezVertexID.getInstance(dagId, 2)).getTerminationCause(),
- VertexTerminationCause.INTERNAL_ERROR);
- Assert.assertEquals(1, dagFinishEventHandler.dagFinishEvents);
- }
-
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a31652b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGScheduler.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGScheduler.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGScheduler.java
index 391dc0c..913f5fa 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGScheduler.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGScheduler.java
@@ -25,11 +25,8 @@ import org.apache.tez.dag.app.dag.TaskAttempt;
import org.apache.tez.dag.app.dag.Vertex;
import org.apache.tez.dag.app.dag.event.DAGEventSchedulerUpdate;
import org.apache.tez.dag.app.dag.event.TaskAttemptEventSchedule;
-import org.apache.tez.dag.app.rm.TaskSchedulerEventHandler;
-import org.apache.tez.dag.records.TezDAGID;
import org.apache.tez.dag.records.TezVertexID;
import org.junit.Assert;
-import org.junit.Ignore;
import org.junit.Test;
import static org.mockito.Mockito.*;
@@ -47,7 +44,7 @@ public class TestDAGScheduler {
MockEventHandler mockEventHandler = new MockEventHandler();
- @Test(timeout=10000)
+ @Test(timeout=5000)
public void testDAGSchedulerNaturalOrder() {
DAG mockDag = mock(DAG.class);
Vertex mockVertex = mock(Vertex.class);
@@ -55,7 +52,6 @@ public class TestDAGScheduler {
when(mockDag.getVertex((TezVertexID) any())).thenReturn(mockVertex);
when(mockVertex.getDistanceFromRoot()).thenReturn(0).thenReturn(1)
.thenReturn(2);
- when(mockAttempt.getIsRescheduled()).thenReturn(false);
DAGEventSchedulerUpdate event = new DAGEventSchedulerUpdate(
DAGEventSchedulerUpdate.UpdateType.TA_SCHEDULE, mockAttempt);
@@ -63,133 +59,14 @@ public class TestDAGScheduler {
DAGScheduler scheduler = new DAGSchedulerNaturalOrder(mockDag,
mockEventHandler);
scheduler.scheduleTask(event);
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 2);
+ Assert.assertEquals(1, mockEventHandler.event.getPriorityHighLimit());
+ Assert.assertEquals(3, mockEventHandler.event.getPriorityLowLimit());
scheduler.scheduleTask(event);
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 4);
+ Assert.assertEquals(4, mockEventHandler.event.getPriorityHighLimit());
+ Assert.assertEquals(6, mockEventHandler.event.getPriorityLowLimit());
scheduler.scheduleTask(event);
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 6);
-
- when(mockAttempt.getIsRescheduled()).thenReturn(true);
- scheduler.scheduleTask(event);
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 5);
+ Assert.assertEquals(7, mockEventHandler.event.getPriorityHighLimit());
+ Assert.assertEquals(9, mockEventHandler.event.getPriorityLowLimit());
}
- @Ignore
- @Test(timeout=10000)
- public void testDAGSchedulerMRR() {
- DAG mockDag = mock(DAG.class);
- TezDAGID dagId = TezDAGID.getInstance("1", 1, 1);
-
- TaskSchedulerEventHandler mockTaskScheduler =
- mock(TaskSchedulerEventHandler.class);
-
- Vertex mockVertex1 = mock(Vertex.class);
- TezVertexID mockVertexId1 = TezVertexID.getInstance(dagId, 1);
- when(mockVertex1.getVertexId()).thenReturn(mockVertexId1);
- when(mockVertex1.getDistanceFromRoot()).thenReturn(0);
- TaskAttempt mockAttempt1 = mock(TaskAttempt.class);
- when(mockAttempt1.getVertexID()).thenReturn(mockVertexId1);
- when(mockAttempt1.getIsRescheduled()).thenReturn(false);
- when(mockDag.getVertex(mockVertexId1)).thenReturn(mockVertex1);
-
- Vertex mockVertex2 = mock(Vertex.class);
- TezVertexID mockVertexId2 = TezVertexID.getInstance(dagId, 2);
- when(mockVertex2.getVertexId()).thenReturn(mockVertexId2);
- when(mockVertex2.getDistanceFromRoot()).thenReturn(1);
- TaskAttempt mockAttempt2 = mock(TaskAttempt.class);
- when(mockAttempt2.getVertexID()).thenReturn(mockVertexId2);
- when(mockAttempt2.getIsRescheduled()).thenReturn(false);
- when(mockDag.getVertex(mockVertexId2)).thenReturn(mockVertex2);
- TaskAttempt mockAttempt2f = mock(TaskAttempt.class);
- when(mockAttempt2f.getVertexID()).thenReturn(mockVertexId2);
- when(mockAttempt2f.getIsRescheduled()).thenReturn(true);
-
- Vertex mockVertex3 = mock(Vertex.class);
- TezVertexID mockVertexId3 = TezVertexID.getInstance(dagId, 3);
- when(mockVertex3.getVertexId()).thenReturn(mockVertexId3);
- when(mockVertex3.getDistanceFromRoot()).thenReturn(2);
- TaskAttempt mockAttempt3 = mock(TaskAttempt.class);
- when(mockAttempt3.getVertexID()).thenReturn(mockVertexId3);
- when(mockAttempt3.getIsRescheduled()).thenReturn(false);
- when(mockDag.getVertex(mockVertexId3)).thenReturn(mockVertex3);
-
- DAGEventSchedulerUpdate mockEvent1 = mock(DAGEventSchedulerUpdate.class);
- when(mockEvent1.getAttempt()).thenReturn(mockAttempt1);
- DAGEventSchedulerUpdate mockEvent2 = mock(DAGEventSchedulerUpdate.class);
- when(mockEvent2.getAttempt()).thenReturn(mockAttempt2);
- DAGEventSchedulerUpdate mockEvent2f = mock(DAGEventSchedulerUpdate.class);
- when(mockEvent2f.getAttempt()).thenReturn(mockAttempt2f);
- DAGEventSchedulerUpdate mockEvent3 = mock(DAGEventSchedulerUpdate.class);
- when(mockEvent3.getAttempt()).thenReturn(mockAttempt3);
- DAGScheduler scheduler = new DAGSchedulerMRR(mockDag, mockEventHandler,
- mockTaskScheduler, 0.5f);
-
- // M starts. M completes. R1 starts. R1 completes. R2 starts. R2 completes
- scheduler.scheduleTask(mockEvent1); // M starts
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 3);
- scheduler.scheduleTask(mockEvent1); // M runs another
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 3);
- scheduler.vertexCompleted(mockVertex1); // M completes
- scheduler.scheduleTask(mockEvent2); // R1 starts
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 6);
- scheduler.scheduleTask(mockEvent2); // R1 runs another
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 6);
- scheduler.scheduleTask(mockEvent2f); // R1 runs retry. Retry priority
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 4);
- scheduler.vertexCompleted(mockVertex2); // R1 completes
- scheduler.scheduleTask(mockEvent3); // R2 starts
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 9);
- scheduler.scheduleTask(mockEvent3); // R2 runs another
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 9);
- scheduler.vertexCompleted(mockVertex3); // R2 completes
-
- // M starts. R1 starts. M completes. R2 starts. R1 completes. R2 completes
- scheduler.scheduleTask(mockEvent1); // M starts
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 3);
- scheduler.scheduleTask(mockEvent2); // R1 starts. Reordered priority
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 2);
- scheduler.scheduleTask(mockEvent1); // M runs another
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 3);
- scheduler.scheduleTask(mockEvent2); // R1 runs another. Reordered priority
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 2);
- scheduler.scheduleTask(mockEvent2f); // R1 runs retry. Reordered priority
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 2);
- scheduler.vertexCompleted(mockVertex1); // M completes
- scheduler.scheduleTask(mockEvent3); // R2 starts. Reordered priority
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 5);
- scheduler.scheduleTask(mockEvent2); // R1 runs another. Normal priority
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 6);
- scheduler.scheduleTask(mockEvent2f); // R1 runs retry. Retry priority
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 4);
- scheduler.scheduleTask(mockEvent3); // R2 runs another. Reordered priority
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 5);
- scheduler.vertexCompleted(mockVertex2); // R1 completes
- scheduler.vertexCompleted(mockVertex3); // R2 completes
-
- // M starts. M completes. R1 starts. R2 starts. R1 completes. R2 completes
- scheduler.scheduleTask(mockEvent1); // M starts
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 3);
- scheduler.vertexCompleted(mockVertex1); // M completes
- scheduler.scheduleTask(mockEvent2); // R1 starts
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 6);
- scheduler.scheduleTask(mockEvent3); // R2 starts. Reordered priority
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 5);
- scheduler.scheduleTask(mockEvent2); // R1 runs another
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 6);
- scheduler.vertexCompleted(mockVertex2); // R1 completes
- scheduler.vertexCompleted(mockVertex3); // R2 completes
-
- // M starts. R1 starts. M completes. R1 completes. R2 starts. R2 completes
- scheduler.scheduleTask(mockEvent1); // M starts
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 3);
- scheduler.scheduleTask(mockEvent2); // R1 starts. Reordered priority
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 2);
- scheduler.vertexCompleted(mockVertex1); // M completes
- scheduler.scheduleTask(mockEvent2); // R1 starts. Normal priority
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 6);
- scheduler.vertexCompleted(mockVertex2); // R1 completes
- scheduler.scheduleTask(mockEvent3); // R2 starts
- Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 9);
- scheduler.vertexCompleted(mockVertex3); // R2 completes
- }
}
http://git-wip-us.apache.org/repos/asf/tez/blob/0a31652b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestTaskAttempt.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestTaskAttempt.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestTaskAttempt.java
index 8e134f2..c98e3de 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestTaskAttempt.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestTaskAttempt.java
@@ -50,7 +50,6 @@ import org.apache.hadoop.yarn.api.records.Container;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.NodeId;
-import org.apache.hadoop.yarn.api.records.Priority;
import org.apache.hadoop.yarn.api.records.Resource;
import org.apache.hadoop.yarn.event.Event;
import org.apache.hadoop.yarn.event.EventHandler;
@@ -91,6 +90,7 @@ import org.apache.tez.runtime.api.events.InputReadErrorEvent;
import org.apache.tez.runtime.api.impl.EventMetaData;
import org.apache.tez.runtime.api.impl.TaskSpec;
import org.apache.tez.runtime.api.impl.TezEvent;
+import org.junit.Assert;
import org.junit.Test;
import org.mockito.ArgumentCaptor;
@@ -104,25 +104,8 @@ public class TestTaskAttempt {
}
}
- // @Test
- // // Verifies # tasks, attempts and diagnostics for a failing job.
- // // TODO Move to TestTask - to verify # retries
- // public void testMRAppHistoryForMap() throws Exception {
- // MRApp app = new FailingAttemptsMRApp(1, 0);
- // testMRAppHistory(app);
- // }
- //
- // @Test
- // // Verifies # tasks, attempts and diagnostics for a failing job.
- // // Move to TestTask - to verify # retries
- // public void testMRAppHistoryForReduce() throws Exception {
- // MRApp app = new FailingAttemptsMRApp(0, 1);
- // testMRAppHistory(app);
- // }
-
@Test(timeout = 5000)
public void testLocalityRequest() {
-
TaskAttemptImpl.ScheduleTaskattemptTransition sta =
new TaskAttemptImpl.ScheduleTaskattemptTransition();
@@ -159,7 +142,61 @@ public class TestTaskAttempt {
assertEquals(host, true, taImpl.taskHosts.contains(host));
}
}
+
+ @Test(timeout = 5000)
+ public void testPriority() {
+ TaskAttemptImpl.ScheduleTaskattemptTransition sta =
+ new TaskAttemptImpl.ScheduleTaskattemptTransition();
+ EventHandler eventHandler = mock(EventHandler.class);
+ TezTaskID taskID = TezTaskID.getInstance(
+ TezVertexID.getInstance(TezDAGID.getInstance("1", 1, 1), 1), 1);
+ TaskAttemptImpl taImpl = new MockTaskAttemptImpl(taskID, 1, eventHandler,
+ mock(TaskAttemptListener.class), new Configuration(), new SystemClock(),
+ mock(TaskHeartbeatHandler.class), mock(AppContext.class),
+ null, false, Resource.newInstance(1024, 1), createFakeContainerContext(), false);
+
+ TaskAttemptImpl taImplReScheduled = new MockTaskAttemptImpl(taskID, 1, eventHandler,
+ mock(TaskAttemptListener.class), new Configuration(), new SystemClock(),
+ mock(TaskHeartbeatHandler.class), mock(AppContext.class),
+ null, true, Resource.newInstance(1024, 1), createFakeContainerContext(), false);
+
+ ArgumentCaptor<Event> arg = ArgumentCaptor.forClass(Event.class);
+
+ TaskAttemptEventSchedule sEvent = mock(TaskAttemptEventSchedule.class);
+ when(sEvent.getPriorityLowLimit()).thenReturn(3);
+ when(sEvent.getPriorityHighLimit()).thenReturn(1);
+ sta.transition(taImpl, sEvent);
+ verify(eventHandler, times(1)).handle(arg.capture());
+ AMSchedulerEventTALaunchRequest launchEvent = (AMSchedulerEventTALaunchRequest)arg.getValue();
+ Assert.assertEquals(2, launchEvent.getPriority());
+ sta.transition(taImplReScheduled, sEvent);
+ verify(eventHandler, times(2)).handle(arg.capture());
+ launchEvent = (AMSchedulerEventTALaunchRequest)arg.getValue();
+ Assert.assertEquals(1, launchEvent.getPriority());
+
+ when(sEvent.getPriorityLowLimit()).thenReturn(6);
+ when(sEvent.getPriorityHighLimit()).thenReturn(4);
+ sta.transition(taImpl, sEvent);
+ verify(eventHandler, times(3)).handle(arg.capture());
+ launchEvent = (AMSchedulerEventTALaunchRequest)arg.getValue();
+ Assert.assertEquals(5, launchEvent.getPriority());
+ sta.transition(taImplReScheduled, sEvent);
+ verify(eventHandler, times(4)).handle(arg.capture());
+ launchEvent = (AMSchedulerEventTALaunchRequest)arg.getValue();
+ Assert.assertEquals(4, launchEvent.getPriority());
+
+ when(sEvent.getPriorityLowLimit()).thenReturn(5);
+ when(sEvent.getPriorityHighLimit()).thenReturn(5);
+ sta.transition(taImpl, sEvent);
+ verify(eventHandler, times(5)).handle(arg.capture());
+ launchEvent = (AMSchedulerEventTALaunchRequest)arg.getValue();
+ Assert.assertEquals(5, launchEvent.getPriority());
+ sta.transition(taImplReScheduled, sEvent);
+ verify(eventHandler, times(6)).handle(arg.capture());
+ launchEvent = (AMSchedulerEventTALaunchRequest)arg.getValue();
+ Assert.assertEquals(5, launchEvent.getPriority());
+ }
@Test(timeout = 5000)
// Tests that an attempt is made to resolve the localized hosts to racks.
@@ -207,93 +244,6 @@ public class TestTaskAttempt {
assertEquals(0, expected.size());
}
- // @Test(timeout = 5000)
- // // Verifies accounting of slot_milli counters. Time spent in running tasks.
- // // TODO Fix this test to work without MRApp.
- // public void testSlotMillisCounterUpdate() throws Exception {
- // verifySlotMillis(2048, 2048, 1024);
- // verifySlotMillis(2048, 1024, 1024);
- // verifySlotMillis(10240, 1024, 2048);
- // }
-
- // public void verifySlotMillis(int mapMemMb, int reduceMemMb,
- // int minContainerSize) throws Exception {
- // Clock actualClock = new SystemClock();
- // ControlledClock clock = new ControlledClock(actualClock);
- // clock.setTime(10);
- // MRApp app =
- // new MRApp(1, 1, false, "testSlotMillisCounterUpdate", true, clock);
- // Configuration conf = new Configuration();
- // conf.setInt(MRJobConfig.MAP_MEMORY_MB, mapMemMb);
- // conf.setInt(MRJobConfig.REDUCE_MEMORY_MB, reduceMemMb);
- // app.setClusterInfo(new ClusterInfo(Resource.newInstance(minContainerSize, 1),
- // Resource.newInstance(10240,1)));
- //
- // Job job = app.submit(conf);
- // app.waitForState(job, JobState.RUNNING);
- // Map<TaskId, Task> tasks = job.getTasks();
- // Assert.assertEquals("Num tasks is not correct", 2, tasks.size());
- // Iterator<Task> taskIter = tasks.values().iterator();
- // Task mTask = taskIter.next();
- // app.waitForState(mTask, TaskState.RUNNING);
- // Task rTask = taskIter.next();
- // app.waitForState(rTask, TaskState.RUNNING);
- // Map<TaskAttemptId, TaskAttempt> mAttempts = mTask.getAttempts();
- // Assert.assertEquals("Num attempts is not correct", 1, mAttempts.size());
- // Map<TaskAttemptId, TaskAttempt> rAttempts = rTask.getAttempts();
- // Assert.assertEquals("Num attempts is not correct", 1, rAttempts.size());
- // TaskAttempt mta = mAttempts.values().iterator().next();
- // TaskAttempt rta = rAttempts.values().iterator().next();
- // app.waitForState(mta, TaskAttemptState.RUNNING);
- // app.waitForState(rta, TaskAttemptState.RUNNING);
- //
- // clock.setTime(11);
- // app.getContext()
- // .getEventHandler()
- // .handle(new TaskAttemptEvent(mta.getID(), TaskAttemptEventType.TA_DONE));
- // app.getContext()
- // .getEventHandler()
- // .handle(new TaskAttemptEvent(rta.getID(), TaskAttemptEventType.TA_DONE));
- // app.waitForState(job, JobState.SUCCEEDED);
- // Assert.assertEquals(mta.getFinishTime(), 11);
- // Assert.assertEquals(mta.getLaunchTime(), 10);
- // Assert.assertEquals(rta.getFinishTime(), 11);
- // Assert.assertEquals(rta.getLaunchTime(), 10);
- // Assert.assertEquals((int) Math.ceil((float) mapMemMb / minContainerSize),
- // job.getAllCounters().findCounter(JobCounter.SLOTS_MILLIS_MAPS)
- // .getValue());
- // Assert.assertEquals(
- // (int) Math.ceil((float) reduceMemMb / minContainerSize), job
- // .getAllCounters().findCounter(JobCounter.SLOTS_MILLIS_REDUCES)
- // .getValue());
- // }
- //
-
- // private void testMRAppHistory(MRApp app) throws Exception {
- // Configuration conf = new Configuration();
- // Job job = app.submit(conf);
- // app.waitForState(job, JobState.FAILED);
- // Map<TaskId, Task> tasks = job.getTasks();
- //
- // Assert.assertEquals("Num tasks is not correct", 1, tasks.size());
- // Task task = tasks.values().iterator().next();
- // Assert.assertEquals("Task state not correct", TaskState.FAILED, task
- // .getReport().getTaskState());
- // Map<TaskAttemptId, TaskAttempt> attempts = tasks.values().iterator().next()
- // .getAttempts();
- // Assert.assertEquals("Num attempts is not correct", 4, attempts.size());
- //
- // Iterator<TaskAttempt> it = attempts.values().iterator();
- // TaskAttemptReport report = it.next().getReport();
- // Assert.assertEquals("Attempt state not correct", TaskAttemptState.FAILED,
- // report.getTaskAttemptState());
- // Assert.assertEquals("Diagnostic Information is not Correct",
- // "Test Diagnostic Event", report.getDiagnosticInfo());
- // report = it.next().getReport();
- // Assert.assertEquals("Attempt state not correct", TaskAttemptState.FAILED,
- // report.getTaskAttemptState());
- // }
-
@Test(timeout = 5000)
// Ensure the dag does not go into an error state if a attempt kill is
// received while STARTING
@@ -333,8 +283,7 @@ public class TestTaskAttempt {
when(container.getId()).thenReturn(contId);
when(container.getNodeId()).thenReturn(nid);
- taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, Priority
- .newInstance(3)));
+ taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, 0, 0));
// At state STARTING.
taImpl.handle(new TaskAttemptEventKillRequest(taskAttemptID, null));
// At some KILLING state.
@@ -392,7 +341,7 @@ public class TestTaskAttempt {
ArgumentCaptor<Event> arg = ArgumentCaptor.forClass(Event.class);
- taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, null));
+ taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, 0, 0));
// At state STARTING.
taImpl.handle(new TaskAttemptEventStartedRemotely(taskAttemptID, contId,
null));
@@ -484,7 +433,7 @@ public class TestTaskAttempt {
mock(TaskHeartbeatHandler.class), appCtx, locationHint, false,
resource, createFakeContainerContext(), false);
- taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, null));
+ taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, 0, 0));
// At state STARTING.
taImpl.handle(new TaskAttemptEventStartedRemotely(taskAttemptID, contId,
null));
@@ -548,7 +497,7 @@ public class TestTaskAttempt {
ArgumentCaptor<Event> arg = ArgumentCaptor.forClass(Event.class);
- taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, null));
+ taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, 0, 0));
// At state STARTING.
taImpl.handle(new TaskAttemptEventStartedRemotely(taskAttemptID, contId,
null));
@@ -638,7 +587,7 @@ public class TestTaskAttempt {
ArgumentCaptor<Event> arg = ArgumentCaptor.forClass(Event.class);
- taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, null));
+ taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, 0, 0));
// At state STARTING.
taImpl.handle(new TaskAttemptEventStartedRemotely(taskAttemptID, contId,
null));
@@ -727,7 +676,7 @@ public class TestTaskAttempt {
ArgumentCaptor<Event> arg = ArgumentCaptor.forClass(Event.class);
- taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, null));
+ taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, 0, 0));
// At state STARTING.
taImpl.handle(new TaskAttemptEventStartedRemotely(taskAttemptID, contId,
null));
@@ -824,7 +773,7 @@ public class TestTaskAttempt {
ArgumentCaptor<Event> arg = ArgumentCaptor.forClass(Event.class);
- taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, null));
+ taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, 0, 0));
// At state STARTING.
taImpl.handle(new TaskAttemptEventStartedRemotely(taskAttemptID, contId,
null));
@@ -915,7 +864,7 @@ public class TestTaskAttempt {
mock(TaskHeartbeatHandler.class), appCtx, locationHint, false,
resource, createFakeContainerContext(), false);
- taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, null));
+ taImpl.handle(new TaskAttemptEventSchedule(taskAttemptID, 0, 0));
// At state STARTING.
taImpl.handle(new TaskAttemptEventStartedRemotely(taskAttemptID, contId,
null));
http://git-wip-us.apache.org/repos/asf/tez/blob/0a31652b/tez-dag/src/test/java/org/apache/tez/dag/app/rm/TestContainerReuse.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/rm/TestContainerReuse.java b/tez-dag/src/test/java/org/apache/tez/dag/app/rm/TestContainerReuse.java
index 1118ee9..20d152e 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/rm/TestContainerReuse.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/rm/TestContainerReuse.java
@@ -1114,7 +1114,7 @@ public class TestContainerReuse {
InputDescriptor.create("inputClassName"), 1)),
Collections.singletonList(new OutputSpec("vertexName",
OutputDescriptor.create("outputClassName"), 1)), null), ta, locationHint,
- priority, containerContext);
+ priority.getPriority(), containerContext);
return lr;
}