Repository: tez Updated Branches: refs/heads/master aa6a84c2c -> 41173aa0a
TEZ-2409. Allow different edges to have different routing plugins (bikas) Project: http://git-wip-us.apache.org/repos/asf/tez/repo Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/41173aa0 Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/41173aa0 Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/41173aa0 Branch: refs/heads/master Commit: 41173aa0a0cc60924902ce9e78a452e2069a2a31 Parents: aa6a84c Author: Bikas Saha <[email protected]> Authored: Thu May 21 23:26:25 2015 -0700 Committer: Bikas Saha <[email protected]> Committed: Thu May 21 23:26:25 2015 -0700 ---------------------------------------------------------------------- CHANGES.txt | 1 + .../tez/dag/app/TaskAttemptEventInfo.java | 8 +- .../dag/app/TaskAttemptListenerImpTezDag.java | 3 +- .../java/org/apache/tez/dag/app/dag/Task.java | 3 +- .../java/org/apache/tez/dag/app/dag/Vertex.java | 2 +- .../org/apache/tez/dag/app/dag/impl/Edge.java | 8 +- .../apache/tez/dag/app/dag/impl/TaskImpl.java | 9 +- .../apache/tez/dag/app/dag/impl/VertexImpl.java | 93 +++++++---------- .../apache/tez/dag/app/MockDAGAppMaster.java | 9 +- .../tez/dag/app/TestMockDAGAppMaster.java | 23 ++--- .../app/TestTaskAttemptListenerImplTezDag.java | 4 +- .../tez/dag/app/dag/impl/TestDAGImpl.java | 100 ++++++++++++++----- .../tez/dag/app/dag/impl/TestVertexImpl.java | 52 ++++++---- .../org/apache/tez/runtime/RuntimeTask.java | 10 ++ .../runtime/api/impl/TezHeartbeatRequest.java | 13 ++- .../runtime/api/impl/TezHeartbeatResponse.java | 12 +++ .../apache/tez/runtime/task/TaskReporter.java | 6 +- 17 files changed, 221 insertions(+), 135 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/CHANGES.txt ---------------------------------------------------------------------- diff --git a/CHANGES.txt b/CHANGES.txt index 0f248b9..995bb6c 100644 --- a/CHANGES.txt +++ b/CHANGES.txt @@ -29,6 +29,7 @@ ALL CHANGES: TEZ-2455. Tez UI: Dag view caching, error handling and minor layout changes TEZ-2453. Tez UI: show the dagInfo is the application has set the same. TEZ-2447. Tez UI: Generic changes based on feedbacks. + TEZ-2409. Allow different edges to have different routing plugins Release 0.7.0: 2015-05-18 http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptEventInfo.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptEventInfo.java b/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptEventInfo.java index 49ff044..d6b03e4 100644 --- a/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptEventInfo.java +++ b/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptEventInfo.java @@ -23,10 +23,12 @@ import org.apache.tez.runtime.api.impl.TezEvent; public class TaskAttemptEventInfo { private final int nextFromEventId; + private final int nextPreRoutedFromEventId; private final List<TezEvent> events; - public TaskAttemptEventInfo(int nextFromEventId, List<TezEvent> events) { + public TaskAttemptEventInfo(int nextFromEventId, List<TezEvent> events, int nextPreRoutedEventId) { this.nextFromEventId = nextFromEventId; + this.nextPreRoutedFromEventId = nextPreRoutedEventId; this.events = events; } @@ -34,6 +36,10 @@ public class TaskAttemptEventInfo { return nextFromEventId; } + public int getNextPreRoutedFromEventId() { + return nextPreRoutedFromEventId; + } + public List<TezEvent> getEvents() { return events; } http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptListenerImpTezDag.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptListenerImpTezDag.java b/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptListenerImpTezDag.java index 970489d..2bf7de3 100644 --- a/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptListenerImpTezDag.java +++ b/tez-dag/src/main/java/org/apache/tez/dag/app/TaskAttemptListenerImpTezDag.java @@ -448,9 +448,10 @@ public class TaskAttemptListenerImpTezDag extends AbstractService implements .getCurrentDAG() .getVertex(taskAttemptID.getTaskID().getVertexID()) .getTaskAttemptTezEvents(taskAttemptID, request.getStartIndex(), - request.getMaxEvents()); + request.getPreRoutedStartIndex(), request.getMaxEvents()); response.setEvents(eventInfo.getEvents()); response.setNextFromEventId(eventInfo.getNextFromEventId()); + response.setNextPreRoutedEventId(eventInfo.getNextPreRoutedFromEventId()); } containerInfo.lastRequestId = requestId; containerInfo.lastReponse = response; http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-dag/src/main/java/org/apache/tez/dag/app/dag/Task.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/Task.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/Task.java index 177ee8a..47b56f2 100644 --- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/Task.java +++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/Task.java @@ -18,6 +18,7 @@ package org.apache.tez.dag.app.dag; +import java.util.ArrayList; import java.util.List; import java.util.Map; @@ -59,7 +60,7 @@ public interface Task { public Vertex getVertex(); - public List<TezEvent> getTaskAttemptTezEvents(TezTaskAttemptID attemptID, + public ArrayList<TezEvent> getTaskAttemptTezEvents(TezTaskAttemptID attemptID, int fromEventId, int maxEvents); public List<String> getDiagnostics(); http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-dag/src/main/java/org/apache/tez/dag/app/dag/Vertex.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/Vertex.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/Vertex.java index bb42392..bb3548d 100644 --- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/Vertex.java +++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/Vertex.java @@ -147,7 +147,7 @@ public interface Vertex extends Comparable<Vertex> { Resource getTaskResource(); public TaskAttemptEventInfo getTaskAttemptTezEvents(TezTaskAttemptID attemptID, - int fromEventId, int maxEvents); + int fromEventId, int nextPreRoutedFromEventId, int maxEvents); void handleSpeculatorEvent(SpeculatorEvent event); http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java index 78bab05..f9cbede 100644 --- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java +++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java @@ -197,7 +197,7 @@ public class Edge { setEdgeProperty(modifiedEdgeProperty); } - public synchronized boolean routingToBegin() throws AMUserCodeException { + public synchronized void routingToBegin() throws AMUserCodeException { if (edgeManagerContext.getDestinationVertexNumTasks() == 0) { routingNeeded = false; } else if (edgeManagerContext.getDestinationVertexNumTasks() < 0) { @@ -216,7 +216,11 @@ public class Edge { } } - LOG.info("Routing to begin for edge: " + getEdgeInfo() + ". EdgeProperty: " + edgeProperty); + LOG.info("Routing to begin for edge: " + getEdgeInfo() + ". EdgeProperty: " + edgeProperty + + " onDemandRouting: " + hasOnDemandRouting()); + } + + public synchronized boolean hasOnDemandRouting() { return onDemandRouting; } http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java index de5ab2a..b2eb81e 100644 --- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java +++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java @@ -138,7 +138,7 @@ public class TaskImpl implements Task, EventHandler<TaskEvent> { final StateChangeNotifier stateChangeNotifier; private final List<TezEvent> tezEventsForTaskAttempts = new ArrayList<TezEvent>(); - private static final List<TezEvent> EMPTY_TASK_ATTEMPT_TEZ_EVENTS = + static final ArrayList<TezEvent> EMPTY_TASK_ATTEMPT_TEZ_EVENTS = new ArrayList(0); // track the status of TaskAttempt (true mean completed, false mean uncompleted) @@ -485,9 +485,9 @@ public class TaskImpl implements Task, EventHandler<TaskEvent> { } @Override - public List<TezEvent> getTaskAttemptTezEvents(TezTaskAttemptID attemptID, + public ArrayList<TezEvent> getTaskAttemptTezEvents(TezTaskAttemptID attemptID, int fromEventId, int maxEvents) { - List<TezEvent> events = EMPTY_TASK_ATTEMPT_TEZ_EVENTS; + ArrayList<TezEvent> events = EMPTY_TASK_ATTEMPT_TEZ_EVENTS; readLock.lock(); try { @@ -500,8 +500,7 @@ public class TaskImpl implements Task, EventHandler<TaskEvent> { int actualMax = Math.min(maxEvents, (tezEventsForTaskAttempts.size() - fromEventId)); int toEventId = actualMax + fromEventId; - events = Collections.unmodifiableList(new ArrayList<TezEvent>( - tezEventsForTaskAttempts.subList(fromEventId, toEventId))); + events = new ArrayList<TezEvent>(tezEventsForTaskAttempts.subList(fromEventId, toEventId)); LOG.info("TaskAttempt:" + attemptID + " sent events: (" + fromEventId + "-" + toEventId + ")"); // currently not modifying the events so that we dont have to create http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java index e20ee6e..e909c9f 100644 --- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java +++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java @@ -228,9 +228,6 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex, EventHandl private Configuration vertexConf; private final boolean isSpeculationEnabled; - - @VisibleForTesting - public boolean useOnDemandRouting = true; //fields initialized in init @@ -742,8 +739,6 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex, EventHandl private final Lock onDemandRouteEventsReadLock = onDemandRouteEventsReadWriteLock.readLock(); private final Lock onDemandRouteEventsWriteLock = onDemandRouteEventsReadWriteLock.writeLock(); - private static final List<TezEvent> EMPTY_TASK_ATTEMPT_TEZ_EVENTS = - new ArrayList(0); List<TezEvent> pendingRouteEvents = new LinkedList<TezEvent>(); List<TezTaskAttemptID> pendingReportedSrcCompletions = Lists.newLinkedList(); @@ -1419,26 +1414,18 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex, EventHandl } } - boolean setupEdgeRouting() throws AMUserCodeException { - boolean doOnDemand = useOnDemandRouting; + void setupEdgeRouting() throws AMUserCodeException { for (Edge e : sourceVertices.values()) { - boolean edgeDoingOnDemand = e.routingToBegin(); - if (doOnDemand && !edgeDoingOnDemand) { - doOnDemand = false; - LOG.info("Not using ondemand routing because of edge between " + e.getSourceVertexName() - + " and " + getLogIdentifier()); - } + e.routingToBegin(); } - return doOnDemand; } private void unsetTasksNotYetScheduled() throws AMUserCodeException { if (tasksNotYetScheduled) { - boolean doOnDemand = setupEdgeRouting(); + setupEdgeRouting(); // change state under lock writeLock.lock(); try { - useOnDemandRouting = doOnDemand; tasksNotYetScheduled = false; // only now can we be sure of the edge manager type. so until now // we will accumulate pending tasks in case legacy routing gets used. @@ -4053,20 +4040,22 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex, EventHandl @Override public TaskAttemptEventInfo getTaskAttemptTezEvents(TezTaskAttemptID attemptID, - int fromEventId, int maxEvents) { - if (!useOnDemandRouting) { - List<TezEvent> events = getTask(attemptID.getTaskID()).getTaskAttemptTezEvents(attemptID, fromEventId, maxEvents); - return new TaskAttemptEventInfo(fromEventId + events.size(), events); - } - + int fromEventId, int preRoutedFromEventId, int maxEvents) { + ArrayList<TezEvent> events = getTask(attemptID.getTaskID()).getTaskAttemptTezEvents( + attemptID, preRoutedFromEventId, maxEvents); + int nextPreRoutedFromEventId = preRoutedFromEventId + events.size(); + int nextFromEventId = fromEventId; onDemandRouteEventsReadLock.lock(); try { - List<TezEvent> events = EMPTY_TASK_ATTEMPT_TEZ_EVENTS; - int nextFromEventId = fromEventId; int currEventCount = onDemandRouteEvents.size(); try { if (currEventCount > fromEventId) { - events = Lists.newArrayListWithCapacity(maxEvents); + if (events != TaskImpl.EMPTY_TASK_ATTEMPT_TEZ_EVENTS) { + events.ensureCapacity(maxEvents); + } else { + events = Lists.newArrayListWithCapacity(maxEvents); + } + int numPreRoutedEvents = events.size(); int taskIndex = attemptID.getTaskID().getId(); Preconditions.checkState(taskIndex < tasks.size(), "Invalid task index for TA: " + attemptID + " vertex: " + getLogIdentifier()); @@ -4118,6 +4107,15 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex, EventHandl break; } } + int numEventsSent = events.size() - numPreRoutedEvents; + if (numEventsSent > 0) { + StringBuilder builder = new StringBuilder(); + builder.append("Sending ").append(attemptID).append(" numEvents: ").append(numEventsSent) + .append(" from: ").append(fromEventId).append(" to: ").append(nextFromEventId) + .append(" out of ").append(currEventCount).append(" on-demand events in vertex: ") + .append(getLogIdentifier()); + LOG.info(builder.toString()); + } } } catch (AMUserCodeException e) { String msg = "Exception in " + e.getSource() + ", vertex=" + getLogIdentifier(); @@ -4126,18 +4124,10 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex, EventHandl nextFromEventId = fromEventId; events.clear(); } - - if (events.size() > 0) { - StringBuilder builder = new StringBuilder(); - builder.append("Sending ").append(attemptID).append(" numEvents: ").append(events.size()) - .append(" from: ").append(fromEventId).append(" to: ").append(nextFromEventId) - .append(" out of ").append(currEventCount).append(" events in vertex: ").append(getLogIdentifier()); - LOG.info(builder.toString()); - } - return new TaskAttemptEventInfo(nextFromEventId, events); } finally { onDemandRouteEventsReadLock.unlock(); } + return new TaskAttemptEventInfo(nextFromEventId, events, nextPreRoutedFromEventId); } private void handleRoutedTezEvents(List<TezEvent> tezEvents, boolean recovered, boolean isPendingEvents) throws AMUserCodeException { @@ -4206,15 +4196,15 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex, EventHandl pendingTaskEvents.add(tezEvent); } else { // event not from this vertex. must have come from source vertex. - if (useOnDemandRouting) { - int srcTaskIndex = sourceMeta.getTaskAttemptID().getTaskID().getId(); - Vertex edgeVertex = getDAG().getVertex(sourceMeta.getTaskVertexName()); - Edge srcEdge = sourceVertices.get(edgeVertex); - if (srcEdge == null) { - throw new TezUncheckedException("Bad source vertex: " + - sourceMeta.getTaskVertexName() + " for destination vertex: " + - getLogIdentifier()); - } + int srcTaskIndex = sourceMeta.getTaskAttemptID().getTaskID().getId(); + Vertex edgeVertex = getDAG().getVertex(sourceMeta.getTaskVertexName()); + Edge srcEdge = sourceVertices.get(edgeVertex); + if (srcEdge == null) { + throw new TezUncheckedException("Bad source vertex: " + + sourceMeta.getTaskVertexName() + " for destination vertex: " + + getLogIdentifier()); + } + if (srcEdge.hasOnDemandRouting()) { onDemandRouteEventsWriteLock.lock(); try { onDemandRouteEvents.add(new EventInfo(tezEvent, srcEdge, srcTaskIndex)); @@ -4223,13 +4213,6 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex, EventHandl } } else { // send to tasks - Edge srcEdge = sourceVertices.get(getDAG().getVertex( - sourceMeta.getTaskVertexName())); - if (srcEdge == null) { - throw new TezUncheckedException("Bad source vertex: " - + sourceMeta.getTaskVertexName() + " for destination vertex: " - + getLogIdentifier()); - } srcEdge.sendTezEventToDestinationTasks(tezEvent); } } @@ -4246,13 +4229,9 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex, EventHandl // removed. pendingTaskEvents.add(tezEvent); } else { - if (useOnDemandRouting) { - onDemandRouteEvents.add(new EventInfo(tezEvent, null, -1)); - } else { - InputDataInformationEvent riEvent = (InputDataInformationEvent) tezEvent.getEvent(); - Task targetTask = getTask(riEvent.getTargetIndex()); - targetTask.registerTezEvent(tezEvent); - } + InputDataInformationEvent riEvent = (InputDataInformationEvent) tezEvent.getEvent(); + Task targetTask = getTask(riEvent.getTargetIndex()); + targetTask.registerTezEvent(tezEvent); } } break; http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-dag/src/test/java/org/apache/tez/dag/app/MockDAGAppMaster.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/MockDAGAppMaster.java b/tez-dag/src/test/java/org/apache/tez/dag/app/MockDAGAppMaster.java index 5cd487c..08f6ff6 100644 --- a/tez-dag/src/test/java/org/apache/tez/dag/app/MockDAGAppMaster.java +++ b/tez-dag/src/test/java/org/apache/tez/dag/app/MockDAGAppMaster.java @@ -162,6 +162,7 @@ public class MockDAGAppMaster extends DAGAppMaster { ContainerLaunchContext launchContext; int numUpdates = 0; int nextFromEventId = 0; + int nextPreRoutedFromEventId = 0; boolean completed; String cIdStr; AtomicBoolean remove = new AtomicBoolean(false); @@ -184,6 +185,7 @@ public class MockDAGAppMaster extends DAGAppMaster { launchContext = null; numUpdates = 0; nextFromEventId = 0; + nextPreRoutedFromEventId = 0; cIdStr = null; remove.set(false); } @@ -338,7 +340,8 @@ public class MockDAGAppMaster extends DAGAppMaster { if (response.shouldDie()) { cData.remove(); } else { - cData.nextFromEventId += response.getNextFromEventId(); + cData.nextFromEventId = response.getNextFromEventId(); + cData.nextPreRoutedFromEventId = response.getNextPreRoutedEventId(); if (!response.getEvents().isEmpty()) { long stopTime = System.nanoTime(); long stopCpuTime = threadMxBean.getCurrentThreadCpuTime(); @@ -421,7 +424,7 @@ public class MockDAGAppMaster extends DAGAppMaster { events.add(new TezEvent(new TaskStatusUpdateEvent(counters, progress, stats), new EventMetaData( EventProducerConsumerType.SYSTEM, cData.vName, "", cData.taId))); TezHeartbeatRequest request = new TezHeartbeatRequest(cData.numUpdates, events, - cData.cIdStr, cData.taId, cData.nextFromEventId, 50000); + cData.nextPreRoutedFromEventId, cData.cIdStr, cData.taId, cData.nextFromEventId, 50000); doHeartbeat(request, cData); } else if (version != null && cData.taId.getId() <= version.intValue()) { preemptContainer(cData); @@ -432,7 +435,7 @@ public class MockDAGAppMaster extends DAGAppMaster { new TaskAttemptCompletedEvent(), new EventMetaData( EventProducerConsumerType.SYSTEM, cData.vName, "", cData.taId))); TezHeartbeatRequest request = new TezHeartbeatRequest(++cData.numUpdates, events, - cData.cIdStr, cData.taId, cData.nextFromEventId, 10000); + cData.nextPreRoutedFromEventId, cData.cIdStr, cData.taId, cData.nextFromEventId, 10000); doHeartbeat(request, cData); cData.clear(); } http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-dag/src/test/java/org/apache/tez/dag/app/TestMockDAGAppMaster.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/TestMockDAGAppMaster.java b/tez-dag/src/test/java/org/apache/tez/dag/app/TestMockDAGAppMaster.java index 1e7faf9..4137d42 100644 --- a/tez-dag/src/test/java/org/apache/tez/dag/app/TestMockDAGAppMaster.java +++ b/tez-dag/src/test/java/org/apache/tez/dag/app/TestMockDAGAppMaster.java @@ -253,8 +253,8 @@ public class TestMockDAGAppMaster { Assert.assertEquals(DAGStatus.State.SUCCEEDED, dagClient.getDAGStatus(null).getState()); VertexImpl vImpl = (VertexImpl) dagImpl.getVertex(vB.getName()); TaskImpl tImpl = (TaskImpl) vImpl.getTask(1); - TezTaskAttemptID taId = TezTaskAttemptID.getInstance(tImpl.getTaskId(), 1); - List<TezEvent> tEvents = vImpl.getTaskAttemptTezEvents(taId, 0, 1000).getEvents(); + TezTaskAttemptID taId = TezTaskAttemptID.getInstance(tImpl.getTaskId(), 0); + List<TezEvent> tEvents = vImpl.getTaskAttemptTezEvents(taId, 0, 0, 1000).getEvents(); Assert.assertEquals(2, tEvents.size()); // 2 from vA Assert.assertEquals(vA.getName(), tEvents.get(0).getDestinationInfo().getEdgeVertexName()); Assert.assertEquals(0, ((DataMovementEvent)tEvents.get(0).getEvent()).getSourceIndex()); @@ -267,8 +267,8 @@ public class TestMockDAGAppMaster { (targetIndex1 == 0 && targetIndex2 == 1) || (targetIndex1 == 1 && targetIndex2 == 0)); vImpl = (VertexImpl) dagImpl.getVertex(vC.getName()); tImpl = (TaskImpl) vImpl.getTask(1); - taId = TezTaskAttemptID.getInstance(tImpl.getTaskId(), 1); - tEvents = vImpl.getTaskAttemptTezEvents(taId, 0, 1000).getEvents(); + taId = TezTaskAttemptID.getInstance(tImpl.getTaskId(), 0); + tEvents = vImpl.getTaskAttemptTezEvents(taId, 0, 0, 1000).getEvents(); Assert.assertEquals(2, tEvents.size()); // 2 from vA Assert.assertEquals(vA.getName(), tEvents.get(0).getDestinationInfo().getEdgeVertexName()); Assert.assertEquals(1, ((DataMovementEvent)tEvents.get(0).getEvent()).getSourceIndex()); @@ -281,8 +281,8 @@ public class TestMockDAGAppMaster { (targetIndex1 == 0 && targetIndex2 == 1) || (targetIndex1 == 1 && targetIndex2 == 0)); vImpl = (VertexImpl) dagImpl.getVertex(vD.getName()); tImpl = (TaskImpl) vImpl.getTask(1); - taId = TezTaskAttemptID.getInstance(tImpl.getTaskId(), 1); - tEvents = vImpl.getTaskAttemptTezEvents(taId, 0, 1000).getEvents(); + taId = TezTaskAttemptID.getInstance(tImpl.getTaskId(), 0); + tEvents = vImpl.getTaskAttemptTezEvents(taId, 0, 0, 1000).getEvents(); Assert.assertEquals(1, tEvents.size()); // 1 from vA Assert.assertEquals(vA.getName(), tEvents.get(0).getDestinationInfo().getEdgeVertexName()); Assert.assertEquals(0, ((DataMovementEvent)tEvents.get(0).getEvent()).getTargetIndex()); @@ -390,22 +390,19 @@ public class TestMockDAGAppMaster { Assert.assertEquals(DAGStatus.State.SUCCEEDED, dagClient.getDAGStatus(null).getState()); // vC uses on demand routing and its task does not provide events VertexImpl vImpl = (VertexImpl) dagImpl.getVertex(vC.getName()); - Assert.assertEquals(true, vImpl.useOnDemandRouting); TaskImpl tImpl = (TaskImpl) vImpl.getTask(0); TezTaskAttemptID taId = TezTaskAttemptID.getInstance(tImpl.getTaskId(), 0); Assert.assertEquals(0, tImpl.getTaskAttemptTezEvents(taId, 0, 1000).size()); - // vD is mixed more and does not use on demand routing and its task provides events + // vD is mixed mode and only 1 out of 2 edges does legacy routing with task providing events vImpl = (VertexImpl) dagImpl.getVertex(vD.getName()); - Assert.assertEquals(false, vImpl.useOnDemandRouting); tImpl = (TaskImpl) vImpl.getTask(0); taId = TezTaskAttemptID.getInstance(tImpl.getTaskId(), 0); - Assert.assertEquals(2, tImpl.getTaskAttemptTezEvents(taId, 0, 1000).size()); + Assert.assertEquals(1, tImpl.getTaskAttemptTezEvents(taId, 0, 1000).size()); // vE has single legacy edge and does not use on demand routing and its task provides events - vImpl = (VertexImpl) dagImpl.getVertex(vD.getName()); - Assert.assertEquals(false, vImpl.useOnDemandRouting); + vImpl = (VertexImpl) dagImpl.getVertex(vE.getName()); tImpl = (TaskImpl) vImpl.getTask(0); taId = TezTaskAttemptID.getInstance(tImpl.getTaskId(), 0); - Assert.assertEquals(2, tImpl.getTaskAttemptTezEvents(taId, 0, 1000).size()); + Assert.assertEquals(1, tImpl.getTaskAttemptTezEvents(taId, 0, 1000).size()); tezClient.stop(); } http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-dag/src/test/java/org/apache/tez/dag/app/TestTaskAttemptListenerImplTezDag.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/TestTaskAttemptListenerImplTezDag.java b/tez-dag/src/test/java/org/apache/tez/dag/app/TestTaskAttemptListenerImplTezDag.java index db8eff1..ac816f4 100644 --- a/tez-dag/src/test/java/org/apache/tez/dag/app/TestTaskAttemptListenerImplTezDag.java +++ b/tez-dag/src/test/java/org/apache/tez/dag/app/TestTaskAttemptListenerImplTezDag.java @@ -243,8 +243,8 @@ public class TestTaskAttemptListenerImplTezDag { doReturn(vertex).when(dag).getVertex(vertexID); doReturn("test_vertex").when(vertex).getName(); - TaskAttemptEventInfo eventInfo = new TaskAttemptEventInfo(nextFromEventId, sendEvents); - doReturn(eventInfo).when(vertex).getTaskAttemptTezEvents(taskAttemptID, fromEventId, maxEvents); + TaskAttemptEventInfo eventInfo = new TaskAttemptEventInfo(nextFromEventId, sendEvents, 0); + doReturn(eventInfo).when(vertex).getTaskAttemptTezEvents(taskAttemptID, fromEventId, 0, maxEvents); taskAttemptListener.registerRunningContainer(containerId); taskAttemptListener.registerTaskAttempt(amContainerTask, containerId); http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/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 fff95b5..4787247 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 @@ -51,6 +51,7 @@ import org.apache.hadoop.yarn.util.Clock; import org.apache.hadoop.yarn.util.SystemClock; import org.apache.tez.common.MockDNSToSwitchMapping; import org.apache.tez.dag.api.DataSinkDescriptor; +import org.apache.tez.dag.api.EdgeManagerPlugin; import org.apache.tez.dag.api.EdgeManagerPluginContext; import org.apache.tez.dag.api.EdgeManagerPluginDescriptor; import org.apache.tez.dag.api.EdgeManagerPluginOnDemand; @@ -115,7 +116,6 @@ import org.apache.tez.dag.app.dag.event.VertexEventRouteEvent; import org.apache.tez.dag.app.dag.event.VertexEventTaskCompleted; import org.apache.tez.dag.app.dag.event.VertexEventTaskReschedule; import org.apache.tez.dag.app.dag.event.VertexEventType; -import org.apache.tez.dag.app.dag.impl.TestDAGImpl.CustomizedEdgeManager.ExceptionLocation; import org.apache.tez.dag.app.dag.impl.TestVertexImpl.CountingOutputCommitter; import org.apache.tez.dag.app.rm.AMSchedulerEvent; import org.apache.tez.dag.app.rm.AMSchedulerEventType; @@ -670,8 +670,8 @@ public class TestDAGImpl { } // v1 -> v2 - private DAGPlan createDAGWithCustomEdge(ExceptionLocation exLocation) { - LOG.info("Setting up dag plan"); + private DAGPlan createDAGWithCustomEdge(ExceptionLocation exLocation, boolean useLegacy) { + LOG.info("Setting up custome edge dag plan " + exLocation + " " + useLegacy); DAGPlan dag = DAGPlan.newBuilder() .setName("testverteximpl") .addVertex( @@ -723,7 +723,8 @@ public class TestDAGImpl { .addEdge( EdgePlan.newBuilder() .setEdgeManager(TezEntityDescriptorProto.newBuilder() - .setClassName(CustomizedEdgeManager.class.getName()) + .setClassName(useLegacy ? CustomizedEdgeManagerLegacy.class.getName() : + CustomizedEdgeManager.class.getName()) .setTezUserPayload(DAGProtos.TezUserPayloadProto.newBuilder() .setUserPayload(ByteString.copyFromUtf8(exLocation.name()))) ) @@ -866,8 +867,12 @@ public class TestDAGImpl { } private void setupDAGWithCustomEdge(ExceptionLocation exLocation) { + setupDAGWithCustomEdge(exLocation, false); + } + + private void setupDAGWithCustomEdge(ExceptionLocation exLocation, boolean useLegacy) { dagWithCustomEdgeId = TezDAGID.getInstance(appAttemptId.getApplicationId(), 4); - dagPlanWithCustomEdge = createDAGWithCustomEdge(exLocation); + dagPlanWithCustomEdge = createDAGWithCustomEdge(exLocation, useLegacy); dagWithCustomEdgeAppContext = mock(AppContext.class); doReturn(aclManager).when(dagWithCustomEdgeAppContext).getAMACLManager(); dagWithCustomEdge = new DAGImpl(dagWithCustomEdgeId, conf, dagPlanWithCustomEdge, @@ -1018,7 +1023,7 @@ public class TestDAGImpl { new EventMetaData(EventProducerConsumerType.INPUT, "vertex1", "vertex2", ta1.getID())); dispatcher.getEventHandler().handle(new VertexEventRouteEvent(v2.getVertexId(), Lists.newArrayList(tezEvent))); dispatcher.await(); - v2.getTaskAttemptTezEvents(ta1.getID(), 0, 1000); + v2.getTaskAttemptTezEvents(ta1.getID(), 0, 0, 1000); dispatcher.await(); Assert.assertEquals(VertexState.FAILED, v2.getState()); @@ -1031,7 +1036,7 @@ public class TestDAGImpl { @Test(timeout = 5000) public void testEdgeManager_RouteDataMovementEventToDestinationWithLegacyRouting() { // Remove after legacy routing is removed - setupDAGWithCustomEdge(ExceptionLocation.RouteDataMovementEventToDestination); + setupDAGWithCustomEdge(ExceptionLocation.RouteDataMovementEventToDestination, true); dispatcher.getEventHandler().handle( new DAGEvent(dagWithCustomEdge.getID(), DAGEventType.DAG_INIT)); dispatcher.getEventHandler().handle(new DAGEventStartDag(dagWithCustomEdge.getID(), @@ -1041,8 +1046,7 @@ public class TestDAGImpl { VertexImpl v1 = (VertexImpl)dagWithCustomEdge.getVertex("vertex1"); VertexImpl v2 = (VertexImpl)dagWithCustomEdge.getVertex("vertex2"); - v1.useOnDemandRouting = false; - v2.useOnDemandRouting = false; + dispatcher.await(); Task t1= v2.getTask(0); TaskAttemptImpl ta1= (TaskAttemptImpl)t1.getAttempt(TezTaskAttemptID.getInstance(t1.getTaskId(), 0)); @@ -1063,7 +1067,7 @@ public class TestDAGImpl { @Test(timeout = 5000) public void testEdgeManager_RouteInputSourceTaskFailedEventToDestinationLegacyRouting() { // Remove after legacy routing is removed - setupDAGWithCustomEdge(ExceptionLocation.RouteInputSourceTaskFailedEventToDestination); + setupDAGWithCustomEdge(ExceptionLocation.RouteInputSourceTaskFailedEventToDestination, true); dispatcher.getEventHandler().handle( new DAGEvent(dagWithCustomEdge.getID(), DAGEventType.DAG_INIT)); dispatcher.getEventHandler().handle(new DAGEventStartDag(dagWithCustomEdge.getID(), @@ -1073,8 +1077,6 @@ public class TestDAGImpl { VertexImpl v1 = (VertexImpl)dagWithCustomEdge.getVertex("vertex1"); VertexImpl v2 = (VertexImpl)dagWithCustomEdge.getVertex("vertex2"); - v1.useOnDemandRouting = false; - v2.useOnDemandRouting = false; dispatcher.await(); Task t1= v2.getTask(0); @@ -1084,7 +1086,7 @@ public class TestDAGImpl { new EventMetaData(EventProducerConsumerType.INPUT,"vertex1", "vertex2", ta1.getID())); dispatcher.getEventHandler().handle(new VertexEventRouteEvent(v2.getVertexId(), Lists.newArrayList(tezEvent))); dispatcher.await(); - v2.getTaskAttemptTezEvents(ta1.getID(), 0, 1000); + v2.getTaskAttemptTezEvents(ta1.getID(), 0, 0, 1000); dispatcher.await(); Assert.assertEquals(VertexState.FAILED, v2.getState()); @@ -1802,18 +1804,17 @@ public class TestDAGImpl { TezConfiguration.TEZ_AM_TASK_MAX_FAILED_ATTEMPTS_DEFAULT)); } - public static class CustomizedEdgeManager extends EdgeManagerPluginOnDemand { - - public static enum ExceptionLocation { - Initialize, - GetNumDestinationTaskPhysicalInputs, - GetNumSourceTaskPhysicalOutputs, - RouteDataMovementEventToDestination, - RouteInputSourceTaskFailedEventToDestination, - GetNumDestinationConsumerTasks, - RouteInputErrorEventToSource - } + public static enum ExceptionLocation { + Initialize, + GetNumDestinationTaskPhysicalInputs, + GetNumSourceTaskPhysicalOutputs, + RouteDataMovementEventToDestination, + RouteInputSourceTaskFailedEventToDestination, + GetNumDestinationConsumerTasks, + RouteInputErrorEventToSource + } + public static class CustomizedEdgeManagerLegacy extends EdgeManagerPlugin { private ExceptionLocation exLocation; public static EdgeManagerPluginDescriptor getUserPayload(ExceptionLocation exLocation) { @@ -1821,7 +1822,7 @@ public class TestDAGImpl { .setUserPayload(UserPayload.create(ByteBuffer.wrap(exLocation.name().getBytes()))); } - public CustomizedEdgeManager(EdgeManagerPluginContext context) { + public CustomizedEdgeManagerLegacy(EdgeManagerPluginContext context) { super(context); this.exLocation = ExceptionLocation.valueOf( new String(context.getUserPayload().deepCopyAsArray())); @@ -1890,6 +1891,55 @@ public class TestDAGImpl { } return 0; } + } + + public static class CustomizedEdgeManager extends EdgeManagerPluginOnDemand { + private ExceptionLocation exLocation; + + public static EdgeManagerPluginDescriptor getUserPayload(ExceptionLocation exLocation) { + return EdgeManagerPluginDescriptor.create(CustomizedEdgeManager.class.getName()) + .setUserPayload(UserPayload.create(ByteBuffer.wrap(exLocation.name().getBytes()))); + } + + public CustomizedEdgeManager(EdgeManagerPluginContext context) { + super(context); + this.exLocation = ExceptionLocation.valueOf( + new String(context.getUserPayload().deepCopyAsArray())); + } + + @Override + public void initialize() throws Exception { + if (exLocation == ExceptionLocation.Initialize) { + throw new Exception(exLocation.name()); + } + } + + @Override + public int getNumDestinationTaskPhysicalInputs(int destinationTaskIndex) + throws Exception { + if (exLocation == ExceptionLocation.GetNumDestinationTaskPhysicalInputs) { + throw new Exception(exLocation.name()); + } + return 0; + } + + @Override + public int getNumSourceTaskPhysicalOutputs(int sourceTaskIndex) + throws Exception { + if (exLocation == ExceptionLocation.GetNumSourceTaskPhysicalOutputs) { + throw new Exception(exLocation.name()); + } + return 0; + } + + @Override + public int getNumDestinationConsumerTasks(int sourceTaskIndex) + throws Exception { + if (exLocation == ExceptionLocation.GetNumDestinationConsumerTasks) { + throw new Exception(exLocation.name()); + } + return 0; + } @Override public int routeInputErrorEventToSource(int destinationTaskIndex, http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java ---------------------------------------------------------------------- diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java index 6c94465..e569949 100644 --- a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java +++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java @@ -2493,7 +2493,13 @@ public class TestVertexImpl { // verify all events have been put in pending. // this is not necessary after legacy routing has been removed Assert.assertEquals(5, v4.pendingTaskEvents.size()); - v4.scheduleTasks(new LinkedList<VertexManagerPluginContext.TaskWithLocationHint>()); + List<TaskWithLocationHint> taskList = new LinkedList<VertexManagerPluginContext.TaskWithLocationHint>(); + // scheduling start to trigger edge routing to begin + for (int i=0; i<v4.getTotalTasks(); ++i) { + taskList.add(new TaskWithLocationHint(i, null)); + } + v4.scheduleTasks(taskList); + dispatcher.await(); // verify all events have been moved to taskEvents Assert.assertEquals(5, v4.getOnDemandRouteEvents().size()); for (int i=5; i<11; ++i) { @@ -2511,7 +2517,7 @@ public class TestVertexImpl { e.edgeManager = mockPlugin; // source task id will not match. all events will return null when(mockPlugin.routeDataMovementEventToDestination(1, 0, 0)).thenReturn(mockRoute); - eventInfo = v4.getTaskAttemptTezEvents(v4TaId, 0, 1); + eventInfo = v4.getTaskAttemptTezEvents(v4TaId, 0, 0, 1); Assert.assertEquals(11, eventInfo.getNextFromEventId()); // all events traversed Assert.assertEquals(0, eventInfo.getEvents().size()); // no events @@ -2522,24 +2528,24 @@ public class TestVertexImpl { mockPlugin.routeDataMovementEventToDestination(anyInt(), anyInt(), anyInt())).thenReturn(mockRoute); for (int i=0; i<11; ++i) { - eventInfo = v4.getTaskAttemptTezEvents(v4TaId, fromEventId, 1); + eventInfo = v4.getTaskAttemptTezEvents(v4TaId, fromEventId, 0, 1); fromEventId = eventInfo.getNextFromEventId(); Assert.assertEquals((i+1), fromEventId); Assert.assertEquals(1, eventInfo.getEvents().size()); } - eventInfo = v4.getTaskAttemptTezEvents(v4TaId, fromEventId, 1); + eventInfo = v4.getTaskAttemptTezEvents(v4TaId, fromEventId, 0, 1); Assert.assertEquals(11, eventInfo.getNextFromEventId()); // all events traversed Assert.assertEquals(0, eventInfo.getEvents().size()); // no events // change max events to larger value. max events does not evenly divide total events fromEventId = 0; for (int i=1; i<=2; ++i) { - eventInfo = v4.getTaskAttemptTezEvents(v4TaId, fromEventId, 5); + eventInfo = v4.getTaskAttemptTezEvents(v4TaId, fromEventId, 0, 5); fromEventId = eventInfo.getNextFromEventId(); Assert.assertEquals((i*5), fromEventId); Assert.assertEquals(5, eventInfo.getEvents().size()); } - eventInfo = v4.getTaskAttemptTezEvents(v4TaId, fromEventId, 5); + eventInfo = v4.getTaskAttemptTezEvents(v4TaId, fromEventId, 0, 5); Assert.assertEquals(11, eventInfo.getNextFromEventId()); // all events traversed Assert.assertEquals(1, eventInfo.getEvents().size()); // remainder events @@ -2551,12 +2557,12 @@ public class TestVertexImpl { fromEventId = 0; int lastFromEventId = 0; for (int i=1; i<=4; ++i) { - eventInfo = v4.getTaskAttemptTezEvents(v4TaId, fromEventId, 5); + eventInfo = v4.getTaskAttemptTezEvents(v4TaId, fromEventId, 0, 5); fromEventId = eventInfo.getNextFromEventId(); Assert.assertEquals((i%2 > 0 ? (lastFromEventId+=2) : (lastFromEventId+=3)), fromEventId); Assert.assertEquals(5, eventInfo.getEvents().size()); } - eventInfo = v4.getTaskAttemptTezEvents(v4TaId, fromEventId, 5); + eventInfo = v4.getTaskAttemptTezEvents(v4TaId, fromEventId, 0, 5); Assert.assertEquals(11, eventInfo.getNextFromEventId()); // all events traversed Assert.assertEquals(2, eventInfo.getEvents().size()); // remainder events } @@ -2754,15 +2760,12 @@ public class TestVertexImpl { } @Test(timeout = 5000) - public void testVertexPendingTaskEventsLegacyRouting() { + public void testVertexPendingTaskEvents() { // Remove after bulk routing API is removed initAllVertices(VertexState.INITED); VertexImpl v3 = vertices.get("vertex3"); VertexImpl v2 = vertices.get("vertex2"); VertexImpl v1 = vertices.get("vertex1"); - v1.useOnDemandRouting = false; - v2.useOnDemandRouting = false; - v3.useOnDemandRouting = false; startVertex(v1); @@ -4733,14 +4736,19 @@ public class TestVertexImpl { Assert.assertEquals(1, inputSpecs.get(0).getPhysicalEdgeCount()); } - // fake scheduling start to trigger edge routing to begin - v1.scheduleTasks(new LinkedList<VertexManagerPluginContext.TaskWithLocationHint>()); + List<TaskWithLocationHint> taskList = new LinkedList<VertexManagerPluginContext.TaskWithLocationHint>(); + // scheduling start to trigger edge routing to begin + for (int i=0; i<v1.getTotalTasks(); ++i) { + taskList.add(new TaskWithLocationHint(i, null)); + } + v1.scheduleTasks(taskList); + dispatcher.await(); // check all tasks get their events for (int i=0; i<v1.getTotalTasks(); ++i) { Assert.assertEquals( 1, v1.getTaskAttemptTezEvents(TezTaskAttemptID.getInstance(v1.getTask(i).getTaskId(), 0), - 0, 100).getEvents().size()); + 0, 0, 100).getEvents().size()); } VertexImplWithControlledInitializerManager v2 = (VertexImplWithControlledInitializerManager) vertices.get("vertex2"); @@ -4775,14 +4783,20 @@ public class TestVertexImpl { } Assert.assertEquals(true, initializerManager2.hasShutDown); - // fake scheduling start to trigger edge routing to begin - v2.scheduleTasks(new LinkedList<VertexManagerPluginContext.TaskWithLocationHint>()); + // scheduling start to trigger edge routing to begin + taskList = new LinkedList<VertexManagerPluginContext.TaskWithLocationHint>(); + // scheduling start to trigger edge routing to begin + for (int i=0; i<v2.getTotalTasks(); ++i) { + taskList.add(new TaskWithLocationHint(i, null)); + } + v2.scheduleTasks(taskList); + dispatcher.await(); // check all tasks get their events for (int i=0; i<v2.getTotalTasks(); ++i) { Assert.assertEquals( ((i==0) ? 2 : 1), v2.getTaskAttemptTezEvents(TezTaskAttemptID.getInstance(v2.getTask(i).getTaskId(), 0), - 0, 100).getEvents().size()); + 0, 0, 100).getEvents().size()); } for (int i = 0; i < 10; i++) { List<InputSpec> inputSpecs = v1.getInputSpecList(i); @@ -4801,7 +4815,6 @@ public class TestVertexImpl { VertexImplWithControlledInitializerManager v1 = (VertexImplWithControlledInitializerManager) vertices .get("vertex1"); - v1.useOnDemandRouting = false; dispatcher.getEventHandler().handle( new VertexEvent(v1.getVertexId(), VertexEventType.V_INIT)); dispatcher.await(); @@ -4828,7 +4841,6 @@ public class TestVertexImpl { VertexImplWithControlledInitializerManager v2 = (VertexImplWithControlledInitializerManager) vertices.get("vertex2"); Assert.assertEquals(VertexState.INITIALIZING, v2.getState()); - v2.useOnDemandRouting = false; // non-task events don't get buffered List<TezEvent> events = Lists.newLinkedList(); http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-runtime-internals/src/main/java/org/apache/tez/runtime/RuntimeTask.java ---------------------------------------------------------------------- diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/RuntimeTask.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/RuntimeTask.java index 921095c..17d7053 100644 --- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/RuntimeTask.java +++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/RuntimeTask.java @@ -50,6 +50,7 @@ public abstract class RuntimeTask { protected final TezUmbilical tezUmbilical; protected final AtomicInteger eventCounter; protected final AtomicInteger nextFromEventId; + protected final AtomicInteger nextPreRoutedEventId; private final AtomicBoolean taskDone; private final TaskCounterUpdater counterUpdater; private final TaskStatistics statistics; @@ -62,6 +63,7 @@ public abstract class RuntimeTask { this.tezCounters = new TezCounters(); this.eventCounter = new AtomicInteger(0); this.nextFromEventId = new AtomicInteger(0); + this.nextPreRoutedEventId = new AtomicInteger(0); this.progress = 0.0f; this.taskDone = new AtomicBoolean(false); this.statistics = new TaskStatistics(); @@ -137,9 +139,17 @@ public abstract class RuntimeTask { return nextFromEventId.get(); } + public int getNextPreRoutedEventId() { + return nextPreRoutedEventId.get(); + } + public void setNextFromEventId(int nextFromEventId) { this.nextFromEventId.set(nextFromEventId); } + + public void setNextPreRoutedEventId(int nextPreRoutedEventId) { + this.nextPreRoutedEventId.set(nextPreRoutedEventId); + } public boolean isTaskDone() { return taskDone.get(); http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezHeartbeatRequest.java ---------------------------------------------------------------------- diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezHeartbeatRequest.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezHeartbeatRequest.java index 3baef93..7ed89f8 100644 --- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezHeartbeatRequest.java +++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezHeartbeatRequest.java @@ -36,6 +36,7 @@ public class TezHeartbeatRequest implements Writable { private List<TezEvent> events; private TezTaskAttemptID currentTaskAttemptID; private int startIndex; + private int preRoutedStartIndex; private int maxEvents; private long requestId; @@ -43,12 +44,13 @@ public class TezHeartbeatRequest implements Writable { } public TezHeartbeatRequest(long requestId, List<TezEvent> events, - String containerIdentifier, TezTaskAttemptID taskAttemptID, - int startIndex, int maxEvents) { + int preRoutedStartIndex, String containerIdentifier, + TezTaskAttemptID taskAttemptID, int startIndex, int maxEvents) { this.containerIdentifier = containerIdentifier; this.requestId = requestId; this.events = Collections.unmodifiableList(events); this.startIndex = startIndex; + this.preRoutedStartIndex = preRoutedStartIndex; this.maxEvents = maxEvents; this.currentTaskAttemptID = taskAttemptID; } @@ -65,6 +67,10 @@ public class TezHeartbeatRequest implements Writable { return startIndex; } + public int getPreRoutedStartIndex() { + return preRoutedStartIndex; + } + public int getMaxEvents() { return maxEvents; } @@ -95,6 +101,7 @@ public class TezHeartbeatRequest implements Writable { out.writeBoolean(false); } out.writeInt(startIndex); + out.writeInt(preRoutedStartIndex); out.writeInt(maxEvents); out.writeLong(requestId); Text.writeString(out, containerIdentifier); @@ -117,6 +124,7 @@ public class TezHeartbeatRequest implements Writable { currentTaskAttemptID = null; } startIndex = in.readInt(); + preRoutedStartIndex = in.readInt(); maxEvents = in.readInt(); requestId = in.readLong(); containerIdentifier = Text.readString(in); @@ -128,6 +136,7 @@ public class TezHeartbeatRequest implements Writable { + " containerId=" + containerIdentifier + ", requestId=" + requestId + ", startIndex=" + startIndex + + ", preRoutedStartIndex=" + preRoutedStartIndex + ", maxEventsToGet=" + maxEvents + ", taskAttemptId=" + currentTaskAttemptID + ", eventCount=" + (events != null ? events.size() : 0) http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezHeartbeatResponse.java ---------------------------------------------------------------------- diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezHeartbeatResponse.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezHeartbeatResponse.java index cecc706..0aa4db4 100644 --- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezHeartbeatResponse.java +++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/api/impl/TezHeartbeatResponse.java @@ -33,6 +33,7 @@ public class TezHeartbeatResponse implements Writable { private boolean shouldDie = false; private List<TezEvent> events; private int nextFromEventId; + private int nextPreRoutedEventId; public TezHeartbeatResponse() { } @@ -57,6 +58,10 @@ public class TezHeartbeatResponse implements Writable { return nextFromEventId; } + public int getNextPreRoutedEventId() { + return nextPreRoutedEventId; + } + public void setEvents(List<TezEvent> events) { this.events = Collections.unmodifiableList(events); } @@ -73,11 +78,16 @@ public class TezHeartbeatResponse implements Writable { this.nextFromEventId = nextFromEventId; } + public void setNextPreRoutedEventId(int nextPreRoutedEventId) { + this.nextPreRoutedEventId = nextPreRoutedEventId; + } + @Override public void write(DataOutput out) throws IOException { out.writeLong(lastRequestId); out.writeBoolean(shouldDie); out.writeInt(nextFromEventId); + out.writeInt(nextPreRoutedEventId); if(events != null) { out.writeBoolean(true); out.writeInt(events.size()); @@ -94,6 +104,7 @@ public class TezHeartbeatResponse implements Writable { lastRequestId = in.readLong(); shouldDie = in.readBoolean(); nextFromEventId = in.readInt(); + nextPreRoutedEventId = in.readInt(); if(in.readBoolean()) { int eventCount = in.readInt(); events = new ArrayList<TezEvent>(eventCount); @@ -111,6 +122,7 @@ public class TezHeartbeatResponse implements Writable { + " lastRequestId=" + lastRequestId + ", shouldDie=" + shouldDie + ", nextFromEventId=" + nextFromEventId + + ", nextPreRoutedEventId=" + nextPreRoutedEventId + ", eventCount=" + (events != null ? events.size() : 0) + " }"; } http://git-wip-us.apache.org/repos/asf/tez/blob/41173aa0/tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TaskReporter.java ---------------------------------------------------------------------- diff --git a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TaskReporter.java b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TaskReporter.java index 8b9db16..d9a7786 100644 --- a/tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TaskReporter.java +++ b/tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TaskReporter.java @@ -249,8 +249,9 @@ public class TaskReporter { long requestId = requestCounter.incrementAndGet(); int fromEventId = task.getNextFromEventId(); - TezHeartbeatRequest request = new TezHeartbeatRequest(requestId, events, containerIdStr, - task.getTaskAttemptID(), fromEventId, maxEventsToGet); + int fromPreRoutedEventId = task.getNextPreRoutedEventId(); + TezHeartbeatRequest request = new TezHeartbeatRequest(requestId, events, fromPreRoutedEventId, + containerIdStr, task.getTaskAttemptID(), fromEventId, maxEventsToGet); if (LOG.isDebugEnabled()) { LOG.debug("Sending heartbeat to AM, request=" + request); } @@ -282,6 +283,7 @@ public class TaskReporter { } } else { task.setNextFromEventId(response.getNextFromEventId()); + task.setNextPreRoutedEventId(response.getNextPreRoutedEventId()); if (response.getEvents() != null && !response.getEvents().isEmpty()) { LOG.info("Routing events from heartbeat response to task" + ", currentTaskAttemptId=" + task.getTaskAttemptID() + ", eventCount=" + response.getEvents().size()
