You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by ss...@apache.org on 2013/06/04 07:33:34 UTC
[32/50] [abbrv] git commit: TEZ-27. Create tests for DAG Scheduler
(bikas)
TEZ-27. Create tests for DAG Scheduler (bikas)
Project: http://git-wip-us.apache.org/repos/asf/incubator-tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-tez/commit/82ee45e8
Tree: http://git-wip-us.apache.org/repos/asf/incubator-tez/tree/82ee45e8
Diff: http://git-wip-us.apache.org/repos/asf/incubator-tez/diff/82ee45e8
Branch: refs/heads/master
Commit: 82ee45e8d3d7f0168b49c534384217305ab81a6d
Parents: 3bb954b
Author: Bikas Saha <bi...@apache.org>
Authored: Tue May 28 19:20:30 2013 -0700
Committer: Bikas Saha <bi...@apache.org>
Committed: Tue May 28 19:20:30 2013 -0700
----------------------------------------------------------------------
.../org/apache/tez/dag/app/dag/TaskAttempt.java | 5 +
.../dag/app/dag/event/DAGEventSchedulerUpdate.java | 2 +-
.../org/apache/tez/dag/app/dag/impl/DAGImpl.java | 30 ++-
.../tez/dag/app/dag/impl/DAGSchedulerMRR.java | 87 +++----
.../dag/app/dag/impl/DAGSchedulerNaturalOrder.java | 2 +-
.../tez/dag/app/dag/impl/TaskAttemptImpl.java | 18 ++-
.../tez/dag/app/rm/TaskSchedulerEventHandler.java | 2 +-
.../tez/dag/app/dag/impl/TestDAGScheduler.java | 188 +++++++++++++++
8 files changed, 277 insertions(+), 57 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/82ee45e8/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 712f968..2628f0c 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
@@ -27,13 +27,18 @@ import org.apache.hadoop.yarn.api.records.NodeId;
import org.apache.tez.common.counters.TezCounters;
import org.apache.tez.dag.api.oldrecords.TaskAttemptReport;
import org.apache.tez.dag.api.oldrecords.TaskAttemptState;
+import org.apache.tez.dag.records.TezDAGID;
import org.apache.tez.dag.records.TezTaskAttemptID;
+import org.apache.tez.dag.records.TezVertexID;
/**
* Read only view of TaskAttempt.
*/
public interface TaskAttempt {
TezTaskAttemptID getID();
+ TezVertexID getVertexID();
+ TezDAGID getDAGID();
+
TaskAttemptReport getReport();
List<String> getDiagnostics();
TezCounters getCounters();
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/82ee45e8/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/DAGEventSchedulerUpdate.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/DAGEventSchedulerUpdate.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/DAGEventSchedulerUpdate.java
index ff11e99..99b872e 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/DAGEventSchedulerUpdate.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/DAGEventSchedulerUpdate.java
@@ -30,7 +30,7 @@ public class DAGEventSchedulerUpdate extends DAGEvent {
private final UpdateType updateType;
public DAGEventSchedulerUpdate(UpdateType updateType, TaskAttempt attempt) {
- super(attempt.getID().getTaskID().getVertexID().getDAGId(),
+ super(attempt.getDAGID(),
DAGEventType.DAG_SCHEDULER_UPDATE);
this.attempt = attempt;
this.updateType = updateType;
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/82ee45e8/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java
index cc6699d..3681c85 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java
@@ -869,9 +869,8 @@ public class DAGImpl implements org.apache.tez.dag.app.dag.DAG,
for (Vertex v : dag.vertices.values()) {
parseVertexEdges(dag, edgePlans, v);
}
-
- dag.dagScheduler = new DAGSchedulerNaturalOrder(dag, dag.eventHandler);
- //dag.dagScheduler = new DAGSchedulerMRR(dag, dag.eventHandler);
+
+ assignDAGScheduler(dag);
// TODO Metrics
//dag.metrics.endPreparingJob(dag);
@@ -887,6 +886,31 @@ public class DAGImpl implements org.apache.tez.dag.app.dag.DAG,
return dag.finished(DAGState.FAILED);
}
}
+
+ private void assignDAGScheduler(DAGImpl dag) {
+ boolean isMRR = true;
+ for(Vertex vertex : dag.vertices.values()) {
+ Map<Vertex, EdgeProperty> outVertices = vertex.getOutputVertices();
+ if(outVertices == null || outVertices.isEmpty()) {
+ continue;
+ }
+ if(outVertices.size() > 1 ||
+ outVertices.values().iterator().next().getConnectionPattern() !=
+ EdgeProperty.ConnectionPattern.BIPARTITE) {
+ // more than 1 output OR single output is not bipartite
+ isMRR = false;
+ break;
+ }
+ }
+
+ if(isMRR) {
+ LOG.info("Using MRR dag scheduler");
+ dag.dagScheduler = new DAGSchedulerMRR(dag, dag.eventHandler);
+ } else {
+ LOG.info("Using Natural order dag scheduler");
+ dag.dagScheduler = new DAGSchedulerNaturalOrder(dag, dag.eventHandler);
+ }
+ }
private VertexImpl createVertex(DAGImpl dag, String vertexName, int vId) {
TezVertexID vertexId = TezBuilderUtils.newVertexID(dag.getID(), vId);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/82ee45e8/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
index 12fdb10..d404155 100644
--- 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
@@ -48,7 +48,7 @@ public class DAGSchedulerMRR implements DAGScheduler {
@Override
public void vertexCompleted(Vertex vertex) {
- if(currentPartitioner!= null) {
+ if(currentPartitioner != null) {
if(vertex != currentPartitioner) {
String message = vertex.getVertexId() + " finished. Expecting "
+ currentPartitioner + " to finish.";
@@ -57,68 +57,59 @@ public class DAGSchedulerMRR implements DAGScheduler {
}
LOG.info("Current partitioner " + currentPartitioner.getVertexId()
+ " is completed. "
- + (currentShuffler!=null?currentShuffler.getVertexId():"null")
- + " is new partitioner");
+ + (currentShuffler!=null ?
+ currentShuffler.getVertexId() + " is new partitioner":
+ "No current shuffler to replace the partitioner"));
currentPartitioner = currentShuffler;
- currentShuffler = null;
- } else {
- if(vertex != currentShuffler) {
- String message = vertex.getVertexId() + " finished. Expecting "
- + currentShuffler.getVertexId() + " to finish";
- LOG.fatal(message);
- throw new TezException(message);
- }
+ currentShuffler = null;
}
}
@Override
public void scheduleTask(DAGEventSchedulerUpdate event) {
TaskAttempt attempt = event.getAttempt();
- Vertex vertex = dag.getVertex(attempt.getID().getTaskID().getVertexID());
+ Vertex vertex = dag.getVertex(attempt.getVertexID());
int vertexDistanceFromRoot = vertex.getDistanceFromRoot();
- if(vertexDistanceFromRoot == 0) {
+ boolean reOrderPriority = false;
+
+ if(currentPartitioner == null) {
+ // no partitioner. so set it.
currentPartitioner = vertex;
- LOG.info(vertex.getVertexId() + " is first partitioner");
+ currentShufflerDepth = vertexDistanceFromRoot;
+ 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;
+ LOG.info(vertex.getVertexId() + " is new shuffler at depth "
+ + currentShufflerDepth);
}
- if(vertexDistanceFromRoot > currentShufflerDepth) {
- if(currentShuffler == null) {
- currentShuffler = vertex;
- currentShufflerDepth = vertexDistanceFromRoot;
- LOG.info(currentShuffler.getVertexId() + " is new shuffler at depth " +
- currentShufflerDepth);
- } else {
- if(currentShufflerDepth+1 == vertexDistanceFromRoot &&
- currentPartitioner == null
- ) {
- currentPartitioner = currentShuffler;
- currentShuffler = vertex;
- currentShufflerDepth = vertexDistanceFromRoot;
- LOG.info("Shuffler " + currentPartitioner.getVertexId() +
- " becomes partitioner as new shuffler " +
- currentShuffler.getVertexId() + " has started at depth " +
- currentShufflerDepth);
- } else {
- String message = vertex.getVertexId()
- + " has scheduled tasks at depth " + vertexDistanceFromRoot
- + " greater than depth " + currentShufflerDepth
- + " of current shuffler " + currentShuffler.getVertexId()
- + ". Unexpected.";
- LOG.fatal(message);
- throw new TezException(message);
- }
- }
+
+ if(currentShuffler == vertex) {
+ // current shuffler vertex. assign special priority
+ reOrderPriority = true;
}
+
+ // sanity check
+ if(currentPartitioner != vertex && currentShuffler != vertex) {
+ String message = vertex.getVertexId() + " is neither the "
+ + " current partitioner: " + currentPartitioner.getVertexId()
+ + " nor the current shuffler: " + currentShuffler.getVertexId();
+ LOG.fatal(message);
+ throw new TezException(message);
+ }
// natural priority. Handles failures and retries.
int priority = (vertexDistanceFromRoot + 1) * 3;
- if(currentShuffler == vertex) {
- if(currentPartitioner != null) {
- // 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
- }
+ 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
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/82ee45e8/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 607e5af..9e8729e 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
@@ -50,7 +50,7 @@ public class DAGSchedulerNaturalOrder implements DAGScheduler {
@Override
public void scheduleTask(DAGEventSchedulerUpdate event) {
TaskAttempt attempt = event.getAttempt();
- Vertex vertex = dag.getVertex(attempt.getID().getTaskID().getVertexID());
+ Vertex vertex = dag.getVertex(attempt.getVertexID());
int vertexDistanceFromRoot = vertex.getDistanceFromRoot();
// natural priority. Handles failures and retries.
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/82ee45e8/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 5d4242d..5225ecf 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
@@ -90,8 +90,10 @@ import org.apache.tez.dag.app.rm.AMSchedulerEventTALaunchRequest;
import org.apache.tez.dag.history.DAGHistoryEvent;
import org.apache.tez.dag.history.events.TaskAttemptFinishedEvent;
import org.apache.tez.dag.history.events.TaskAttemptStartedEvent;
+import org.apache.tez.dag.records.TezDAGID;
import org.apache.tez.dag.records.TezTaskAttemptID;
import org.apache.tez.dag.records.TezTaskID;
+import org.apache.tez.dag.records.TezVertexID;
import org.apache.tez.dag.utils.TezBuilderUtils;
import org.apache.tez.engine.common.security.JobTokenIdentifier;
@@ -290,6 +292,16 @@ public class TaskAttemptImpl implements TaskAttempt,
return attemptId;
}
+ @Override
+ public TezVertexID getVertexID() {
+ return attemptId.getTaskID().getVertexID();
+ }
+
+ @Override
+ public TezDAGID getDAGID() {
+ return getVertexID().getDAGId();
+ }
+
TezTaskContext createRemoteTask() {
Vertex vertex = getTask().getVertex();
DAG dag = vertex.getDAG();
@@ -601,7 +613,7 @@ public class TaskAttemptImpl implements TaskAttempt,
TaskAttemptImpl ta) {
DAGEventCounterUpdate jce =
new DAGEventCounterUpdate(
- ta.getID().getTaskID().getVertexID().getDAGId()
+ ta.getDAGID()
);
jce.addCounterUpdate(DAGCounter.TOTAL_LAUNCHED_TASKS, 1);
return jce;
@@ -611,7 +623,7 @@ public class TaskAttemptImpl implements TaskAttempt,
TaskAttemptImpl ta) {
DAGEventCounterUpdate jce =
new DAGEventCounterUpdate(
- ta.getID().getTaskID().getVertexID().getDAGId()
+ ta.getDAGID()
);
long slotMillis = computeSlotMillis(ta);
@@ -624,7 +636,7 @@ public class TaskAttemptImpl implements TaskAttempt,
TaskAttemptStateInternal taState) {
DAGEventCounterUpdate jce =
new DAGEventCounterUpdate(
- taskAttempt.getID().getTaskID().getVertexID().getDAGId());
+ taskAttempt.getDAGID());
long slotMillisIncrement = computeSlotMillis(taskAttempt);
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/82ee45e8/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 c61cbcc..8b98c68 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
@@ -461,7 +461,7 @@ public class TaskSchedulerEventHandler extends AbstractService
sendEvent(new AMContainerEventLaunchRequest(
containerId,
- taskAttempt.getID().getTaskID().getVertexID(),
+ taskAttempt.getVertexID(),
event.getJobToken(),
// TODO getConf from AMSchedulerEventTALaunchRequest
event.getCredentials(), false, event.getConf(),
http://git-wip-us.apache.org/repos/asf/incubator-tez/blob/82ee45e8/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
new file mode 100644
index 0000000..988a116
--- /dev/null
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestDAGScheduler.java
@@ -0,0 +1,188 @@
+/**
+ * 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 org.apache.hadoop.yarn.event.EventHandler;
+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.TaskAttemptEventSchedule;
+import org.apache.tez.dag.records.TezDAGID;
+import org.apache.tez.dag.records.TezVertexID;
+import org.junit.Assert;
+import org.junit.Test;
+
+import static org.mockito.Mockito.*;
+
+public class TestDAGScheduler {
+
+ class MockEventHandler implements EventHandler<TaskAttemptEventSchedule> {
+ TaskAttemptEventSchedule event;
+ @Override
+ public void handle(TaskAttemptEventSchedule event) {
+ this.event = event;
+ }
+
+ }
+
+ MockEventHandler mockEventHandler = new MockEventHandler();
+
+ //@Test(timeout=10000)
+ public void testDAGSchedulerNaturalOrder() {
+ DAG mockDag = mock(DAG.class);
+ Vertex mockVertex = mock(Vertex.class);
+ TaskAttempt mockAttempt = mock(TaskAttempt.class);
+ 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);
+
+ DAGScheduler scheduler = new DAGSchedulerNaturalOrder(mockDag,
+ mockEventHandler);
+ scheduler.scheduleTask(event);
+ Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 2);
+ scheduler.scheduleTask(event);
+ Assert.assertTrue(mockEventHandler.event.getPriority().getPriority() == 4);
+ 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);
+ }
+
+ @Test//(timeout=10000)
+ public void testDAGSchedulerMRR() {
+ DAG mockDag = mock(DAG.class);
+ TezDAGID dagId = new TezDAGID("1", 1, 1);
+
+ Vertex mockVertex1 = mock(Vertex.class);
+ TezVertexID mockVertexId1 = new TezVertexID(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 = new TezVertexID(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 = new TezVertexID(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);
+
+ // 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
+ }
+}