Repository: tez Updated Branches: refs/heads/master 7f936f614 -> dce57c83a
TEZ-3909. DAG can hang if vertex with no tasks is killed (Jason Lowe via jeagles) Project: http://git-wip-us.apache.org/repos/asf/tez/repo Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/dce57c83 Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/dce57c83 Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/dce57c83 Branch: refs/heads/master Commit: dce57c83ab942e5698b7107bfa176e5efa35e269 Parents: 7f936f6 Author: Jonathan Eagles <[email protected]> Authored: Tue Apr 3 12:45:20 2018 -0500 Committer: Jonathan Eagles <[email protected]> Committed: Tue Apr 3 12:45:20 2018 -0500 ---------------------------------------------------------------------- .../apache/tez/dag/app/dag/impl/VertexImpl.java | 8 +++-- .../tez/dag/app/dag/impl/TestVertexImpl.java | 32 ++++++++++++++++++++ 2 files changed, 38 insertions(+), 2 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/tez/blob/dce57c83/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 d727e39..ad26173 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 @@ -514,6 +514,11 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex, EventHandl EnumSet.of(VertexState.TERMINATING, VertexState.KILLED, VertexState.FAILED, VertexState.ERROR), VertexEventType.V_TASK_COMPLETED, new TaskCompletedTransition()) + .addTransition + (VertexState.TERMINATING, + EnumSet.of(VertexState.TERMINATING, VertexState.KILLED, VertexState.FAILED, VertexState.ERROR), + VertexEventType.V_COMPLETED, + new VertexNoTasksCompletedTransition()) .addTransition( VertexState.TERMINATING, VertexState.ERROR, VertexEventType.V_INTERNAL_ERROR, @@ -534,8 +539,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex, EventHandl VertexEventType.V_ROUTE_EVENT, VertexEventType.V_SOURCE_TASK_ATTEMPT_COMPLETED, VertexEventType.V_TASK_ATTEMPT_COMPLETED, - VertexEventType.V_TASK_RESCHEDULED, - VertexEventType.V_COMPLETED)) + VertexEventType.V_TASK_RESCHEDULED)) // Transitions from SUCCEEDED state .addTransition( http://git-wip-us.apache.org/repos/asf/tez/blob/dce57c83/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 d382974..5722406 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 @@ -5769,6 +5769,38 @@ public class TestVertexImpl { } } + @Test(timeout = 10000) + public void testVertexNoTasksTerminated() { + TezVertexID vId = null; + try { + TezDAGID invalidDagId = TezDAGID.getInstance( + dagId.getApplicationId(), 1000); + vId = TezVertexID.getInstance(invalidDagId, 1); + VertexPlan vPlan = invalidDagPlan.getVertex(0); + EventHandler mockEventHandler = mock(EventHandler.class); + VertexImpl v = new VertexImpl(vId, vPlan, vPlan.getName(), conf, + mockEventHandler, taskCommunicatorManagerInterface, + clock, thh, true, appContext, vertexLocationHint, null, taskSpecificLaunchCmdOption, + updateTracker, new Configuration(false)); + v.setInputVertices(new HashMap()); + vertexIdMap.put(vId, v); + vertices.put(v.getName(), v); + v.handle(new VertexEvent(vId, VertexEventType.V_INIT)); + Assert.assertEquals(VertexState.INITED, v.getState()); + v.handle(new VertexEvent(vId, VertexEventType.V_START)); + Assert.assertEquals(VertexState.RUNNING, v.getState()); + v.handle(new VertexEventTermination(vId, VertexTerminationCause.OTHER_VERTEX_FAILURE)); + Assert.assertEquals(VertexState.TERMINATING, v.getState()); + v.handle(new VertexEvent(vId, VertexEventType.V_COMPLETED)); + Assert.assertEquals(VertexState.KILLED, v.getState()); + Assert.assertTrue(1.0f == v.getCompletedTaskProgress()); + } finally { + if (vId != null) { + vertexIdMap.remove(vId); + } + } + } + @SuppressWarnings("rawtypes") private static class VertexImplWithRunningInputInitializer extends VertexImpl {
