You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@tez.apache.org by zj...@apache.org on 2014/10/23 01:48:08 UTC
[1/2] TEZ-1267. Exception handling for VertexManager (zjffdu)
Repository: tez
Updated Branches:
refs/heads/branch-0.5 87e473ee8 -> eee882e94
http://git-wip-us.apache.org/repos/asf/tez/blob/eee882e9/tez-tests/src/test/java/org/apache/tez/test/TestExceptionPropagation.java
----------------------------------------------------------------------
diff --git a/tez-tests/src/test/java/org/apache/tez/test/TestExceptionPropagation.java b/tez-tests/src/test/java/org/apache/tez/test/TestExceptionPropagation.java
index 38d5998..44b557b 100644
--- a/tez-tests/src/test/java/org/apache/tez/test/TestExceptionPropagation.java
+++ b/tez-tests/src/test/java/org/apache/tez/test/TestExceptionPropagation.java
@@ -41,18 +41,29 @@ import org.apache.hadoop.yarn.api.records.YarnApplicationState;
import org.apache.hadoop.yarn.client.api.YarnClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.tez.client.TezClient;
+import org.apache.tez.common.TezUtils;
import org.apache.tez.dag.api.DAG;
import org.apache.tez.dag.api.DataSinkDescriptor;
import org.apache.tez.dag.api.DataSourceDescriptor;
+import org.apache.tez.dag.api.Edge;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
import org.apache.tez.dag.api.InputDescriptor;
import org.apache.tez.dag.api.InputInitializerDescriptor;
import org.apache.tez.dag.api.OutputDescriptor;
import org.apache.tez.dag.api.ProcessorDescriptor;
import org.apache.tez.dag.api.TezConfiguration;
+import org.apache.tez.dag.api.TezUncheckedException;
import org.apache.tez.dag.api.UserPayload;
import org.apache.tez.dag.api.Vertex;
+import org.apache.tez.dag.api.VertexManagerPluginContext;
+import org.apache.tez.dag.api.VertexManagerPluginDescriptor;
+import org.apache.tez.dag.api.EdgeProperty.DataSourceType;
+import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
import org.apache.tez.dag.api.client.DAGClient;
import org.apache.tez.dag.api.client.DAGStatus;
+import org.apache.tez.dag.app.dag.impl.RootInputVertexManager;
+import org.apache.tez.dag.library.vertexmanager.ShuffleVertexManager;
import org.apache.tez.runtime.api.AbstractLogicalIOProcessor;
import org.apache.tez.runtime.api.AbstractLogicalInput;
import org.apache.tez.runtime.api.AbstractLogicalOutput;
@@ -68,7 +79,9 @@ import org.apache.tez.runtime.api.Reader;
import org.apache.tez.runtime.api.Writer;
import org.apache.tez.runtime.api.events.InputDataInformationEvent;
import org.apache.tez.runtime.api.events.InputInitializerEvent;
+import org.apache.tez.runtime.api.events.VertexManagerEvent;
import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.apache.tez.test.TestAMRecovery.DoNothingProcessor;
import org.junit.Test;
public class TestExceptionPropagation {
@@ -168,7 +181,6 @@ public class TestExceptionPropagation {
tezConf.setInt(TezConfiguration.TEZ_AM_RESOURCE_MEMORY_MB, 500);
tezConf.set(TezConfiguration.TEZ_AM_LAUNCH_CMD_OPTS, " -Xmx256m");
tezConf.setBoolean(TezConfiguration.TEZ_AM_SESSION_MODE, false);
-
tezClient = TezClient.create("TestExceptionPropagation", tezConf);
tezClient.start();
}
@@ -280,16 +292,21 @@ public class TestExceptionPropagation {
// PROCESSOR_HANDLE_EVENTS
PROCESSOR_RUN_ERROR, PROCESSOR_CLOSE_ERROR, PROCESSOR_INITIALIZE_ERROR,
PROCESSOR_RUN_EXCEPTION, PROCESSOR_CLOSE_EXCEPTION, PROCESSOR_INITIALIZE_EXCEPTION,
+
+ // VM
+ VM_INITIALIZE, VM_ON_ROOTVERTEX_INITIALIZE,VM_ON_SOURCETASK_COMPLETED, VM_ON_VERTEX_STARTED,
+ VM_ON_VERTEXMANAGEREVENT_RECEIVED,
}
/**
- * create a DAG with single vertex, set payload on Input/Output/Processor to
+ * create a DAG with 2 vertices (v1 --> v2), set payload on Input/Output/Processor/VertexManagerPlugin to
* control where throw exception
*
* @param exLocation
* @return
+ * @throws IOException
*/
- private DAG createDAG(ExceptionLocation exLocation) {
+ private DAG createDAG(ExceptionLocation exLocation) throws IOException {
DAG dag = DAG.create("dag_" + exLocation.name());
UserPayload payload =
UserPayload.create(ByteBuffer.wrap(exLocation.name().getBytes()));
@@ -302,7 +319,17 @@ public class TestExceptionPropagation {
DataSourceDescriptor.create(inputDesc, iiDesc, null));
OutputDescriptor outputDesc = OutputWithException.getOutputDesc(payload);
v1.addDataSink("output", DataSinkDescriptor.create(outputDesc, null, null));
- dag.addVertex(v1);
+ v1.setVertexManagerPlugin(RootInputVertexManagerWithException.getVMDesc(payload));
+
+ Vertex v2 =
+ Vertex.create("v2", DoNothingProcessor.getProcDesc(), 1);
+ v2.setVertexManagerPlugin(ShuffleVertexManagerWithException.getVMDesc(exLocation));
+
+ dag.addVertex(v1)
+ .addVertex(v2)
+ .addEdge(Edge.create(v1, v2, EdgeProperty.create(DataMovementType.SCATTER_GATHER,
+ DataSourceType.PERSISTED, SchedulingType.SEQUENTIAL,
+ OutputWithException.getOutputDesc(payload), InputWithException.getInputDesc(payload))));
return dag;
}
@@ -438,8 +465,14 @@ public class TestExceptionPropagation {
public List<Event> close() throws Exception {
if (this.exLocation == ExceptionLocation.OUTPUT_CLOSE) {
throw new RuntimeException(this.exLocation.name());
+ } else if (this.exLocation == ExceptionLocation.VM_ON_VERTEXMANAGEREVENT_RECEIVED) {
+ // send VertexManagerEvent to v2
+ List<Event> events = new ArrayList<Event>();
+ events.add(VertexManagerEvent.create("v2", ByteBuffer.wrap(new byte[0])));
+ return events;
+ } else {
+ return null;
}
- return null;
}
@Override
@@ -513,4 +546,92 @@ public class TestExceptionPropagation {
.setUserPayload(payload);
}
}
+
+ public static class RootInputVertexManagerWithException extends RootInputVertexManager {
+
+ private ExceptionLocation exLocation;
+
+ public RootInputVertexManagerWithException(VertexManagerPluginContext context) {
+ super(context);
+ }
+
+ @Override
+ public void initialize() {
+ super.initialize();
+ this.exLocation =
+ ExceptionLocation.valueOf(new String(getContext().getUserPayload()
+ .deepCopyAsArray()));
+ if (this.exLocation == ExceptionLocation.VM_INITIALIZE) {
+ throw new RuntimeException(this.exLocation.name());
+ }
+ }
+
+ @Override
+ public void onRootVertexInitialized(String inputName,
+ InputDescriptor inputDescriptor, List<Event> events) {
+ if (this.exLocation == ExceptionLocation.VM_ON_ROOTVERTEX_INITIALIZE) {
+ throw new RuntimeException(this.exLocation.name());
+ }
+ super.onRootVertexInitialized(inputName, inputDescriptor, events);
+ }
+
+ @Override
+ public void onVertexStarted(Map<String, List<Integer>> completions) {
+ if (this.exLocation == ExceptionLocation.VM_ON_VERTEX_STARTED) {
+ throw new RuntimeException(this.exLocation.name());
+ }
+ super.onVertexStarted(completions);
+ }
+
+ public static VertexManagerPluginDescriptor getVMDesc(UserPayload payload) {
+ return VertexManagerPluginDescriptor.create(RootInputVertexManagerWithException.class.getName())
+ .setUserPayload(payload);
+ }
+ }
+
+ public static class ShuffleVertexManagerWithException extends ShuffleVertexManager {
+
+ private ExceptionLocation exLocation;
+ private static final String Test_ExceptionLocation = "Test.ExceptionLocation";
+
+ public ShuffleVertexManagerWithException(VertexManagerPluginContext context) {
+ super(context);
+ }
+
+ @Override
+ public void initialize() {
+ super.initialize();
+ Configuration conf;
+ try {
+ conf = TezUtils.createConfFromUserPayload(getContext().getUserPayload());
+ this.exLocation = ExceptionLocation.valueOf(conf.get(Test_ExceptionLocation));
+ } catch (IOException e) {
+ throw new TezUncheckedException(e);
+ }
+ }
+
+ @Override
+ public void onSourceTaskCompleted(String srcVertexName, Integer attemptId) {
+ if (this.exLocation == ExceptionLocation.VM_ON_SOURCETASK_COMPLETED) {
+ throw new RuntimeException(this.exLocation.name());
+ }
+ super.onSourceTaskCompleted(srcVertexName, attemptId);
+ }
+
+ @Override
+ public void onVertexManagerEventReceived(VertexManagerEvent vmEvent) {
+ if (this.exLocation == ExceptionLocation.VM_ON_VERTEXMANAGEREVENT_RECEIVED) {
+ throw new RuntimeException(this.exLocation.name());
+ }
+ super.onVertexManagerEventReceived(vmEvent);
+ }
+
+ public static VertexManagerPluginDescriptor getVMDesc(ExceptionLocation exLocation) throws IOException {
+ Configuration conf = new Configuration();
+ conf.set(Test_ExceptionLocation, exLocation.name());
+ UserPayload payload = TezUtils.createUserPayloadFromConf(conf);
+ return VertexManagerPluginDescriptor.create(ShuffleVertexManagerWithException.class.getName())
+ .setUserPayload(payload);
+ }
+ }
}
[2/2] git commit: TEZ-1267. Exception handling for VertexManager
(zjffdu)
Posted by zj...@apache.org.
TEZ-1267. Exception handling for VertexManager (zjffdu)
(cherry picked from commit b4674b485dd4e62c177d87581f302b243f24d797)
Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/eee882e9
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/eee882e9
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/eee882e9
Branch: refs/heads/branch-0.5
Commit: eee882e942e15d223e315f8dd368fb5bf68e3266
Parents: 87e473e
Author: Jeff Zhang <zj...@apache.org>
Authored: Thu Oct 23 07:43:38 2014 +0800
Committer: Jeff Zhang <zj...@apache.org>
Committed: Thu Oct 23 07:46:31 2014 +0800
----------------------------------------------------------------------
CHANGES.txt | 2 +-
.../apache/tez/dag/api/VertexManagerPlugin.java | 15 +-
.../tez/dag/app/dag/TaskTerminationCause.java | 7 +-
.../tez/dag/app/dag/VertexTerminationCause.java | 5 +
.../dag/app/dag/impl/AMUserCodeException.java | 56 ++
.../apache/tez/dag/app/dag/impl/VertexImpl.java | 624 +++++++++++--------
.../tez/dag/app/dag/impl/VertexManager.java | 41 +-
.../tez/dag/app/dag/impl/TestVertexImpl.java | 407 +++++++++++-
.../tez/dag/app/dag/impl/TestVertexManager.java | 5 +-
.../tez/test/TestExceptionPropagation.java | 131 +++-
10 files changed, 1016 insertions(+), 277 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/tez/blob/eee882e9/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index afccb99..dd5d1d8 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -51,7 +51,7 @@ ALL CHANGES:
TEZ-1643. DAGAppMaster kills DAG & shuts down, when RM is restarted.
TEZ-1684. upgrade mockito to latest release.
TEZ-1567. Avoid blacklisting nodes when the disable blacklisting threshold is about to be hit.
-
+ TEZ-1267. Exception handling for VertexManager.
Release 0.5.1: 2014-10-02
http://git-wip-us.apache.org/repos/asf/tez/blob/eee882e9/tez-api/src/main/java/org/apache/tez/dag/api/VertexManagerPlugin.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/VertexManagerPlugin.java b/tez-api/src/main/java/org/apache/tez/dag/api/VertexManagerPlugin.java
index f07734f..b494b12 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/VertexManagerPlugin.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/VertexManagerPlugin.java
@@ -54,36 +54,41 @@ public abstract class VertexManagerPlugin {
/**
* Initialize the plugin. Called when the vertex is initializing. This happens
* after all source vertices and inputs have initialized
+ * @throws Exception
*/
- public abstract void initialize();
+ public abstract void initialize() throws Exception;
/**
* Notification that the vertex is ready to start running tasks
* @param completions Source vertices and all their tasks that have already completed
+ * @throws Exception
*/
- public abstract void onVertexStarted(Map<String, List<Integer>> completions);
+ public abstract void onVertexStarted(Map<String, List<Integer>> completions) throws Exception;
/**
* Notification of a source vertex completion.
* @param srcVertexName
* @param taskId Index of the task that completed
+ * @throws Exception
*/
- public abstract void onSourceTaskCompleted(String srcVertexName, Integer taskId);
+ public abstract void onSourceTaskCompleted(String srcVertexName, Integer taskId) throws Exception;
/**
* Notification of an event directly sent to this vertex manager
* @param vmEvent
+ * @throws Exception
*/
- public abstract void onVertexManagerEventReceived(VertexManagerEvent vmEvent);
+ public abstract void onVertexManagerEventReceived(VertexManagerEvent vmEvent) throws Exception;
/**
* Notification that the inputs of this vertex have initialized
* @param inputName
* @param inputDescriptor
* @param events
+ * @throws Exception
*/
public abstract void onRootVertexInitialized(String inputName,
- InputDescriptor inputDescriptor, List<Event> events);
+ InputDescriptor inputDescriptor, List<Event> events) throws Exception;
/**
* Return ahe {@link org.apache.tez.dag.api.VertexManagerPluginContext} for this specific instance of
http://git-wip-us.apache.org/repos/asf/tez/blob/eee882e9/tez-dag/src/main/java/org/apache/tez/dag/app/dag/TaskTerminationCause.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/TaskTerminationCause.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/TaskTerminationCause.java
index 321dd01..eb0a2af 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/TaskTerminationCause.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/TaskTerminationCause.java
@@ -33,5 +33,10 @@ public enum TaskTerminationCause {
OTHER_TASK_FAILURE,
/** One of the tasks of the destination vertex failed. */
- OWN_TASK_FAILURE
+ OWN_TASK_FAILURE,
+
+ /** This vertex failed as its AM usercode (VertexManager/EdgeManager/InputInitializer)
+ * throw Exception
+ */
+ AM_USERCODE_FAILURE,
}
http://git-wip-us.apache.org/repos/asf/tez/blob/eee882e9/tez-dag/src/main/java/org/apache/tez/dag/app/dag/VertexTerminationCause.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/VertexTerminationCause.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/VertexTerminationCause.java
index d06fac3..4bfe001 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/VertexTerminationCause.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/VertexTerminationCause.java
@@ -32,6 +32,11 @@ public enum VertexTerminationCause {
/** Initialization failed for one of the root Inputs */
ROOT_INPUT_INIT_FAILURE,
+ /** This vertex failed as its AM usercode (VertexManager/EdgeManager/InputInitializer)
+ * throw Exception
+ */
+ AM_USERCODE_FAILURE,
+
/** One of the tasks for this vertex failed. */
OWN_TASK_FAILURE,
http://git-wip-us.apache.org/repos/asf/tez/blob/eee882e9/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/AMUserCodeException.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/AMUserCodeException.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/AMUserCodeException.java
new file mode 100644
index 0000000..b4e1849
--- /dev/null
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/AMUserCodeException.java
@@ -0,0 +1,56 @@
+/**
+ * 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.classification.InterfaceAudience.Private;
+
+
+/**
+ * Wrap the exception of user code in AM. Now we only have 3 kinds of user code in AM:
+ * <li>VertexManager</li>
+ * <li>EdgeManager</li>
+ * <li>InputInitializer</li>
+ */
+@Private
+public class AMUserCodeException extends Exception {
+
+ private static final long serialVersionUID = -3642816091492797520L;
+
+ public static enum Source {
+ VertexManager,
+ EdgeManager,
+ InputInitializer
+ }
+
+ private Source source;
+
+ public AMUserCodeException(Source source, String message, Throwable cause) {
+ super(message, cause);
+ this.source = source;
+ }
+
+ public AMUserCodeException(Source source, Throwable cause) {
+ super(cause);
+ this.source = source;
+ }
+
+ public Source getSource() {
+ return source;
+ }
+}
http://git-wip-us.apache.org/repos/asf/tez/blob/eee882e9/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 9f88523..9010cbc 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
@@ -157,6 +157,7 @@ import org.apache.tez.runtime.api.impl.GroupInputSpec;
import org.apache.tez.runtime.api.impl.InputSpec;
import org.apache.tez.runtime.api.impl.OutputSpec;
import org.apache.tez.runtime.api.impl.TezEvent;
+
import com.google.common.annotations.VisibleForTesting;
import com.google.common.base.Preconditions;
import com.google.common.collect.HashMultiset;
@@ -265,10 +266,12 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
EnumSet.of(VertexState.NEW),
VertexEventType.V_NULL_EDGE_INITIALIZED,
new NullEdgeInitializedTransition())
- .addTransition(VertexState.NEW, VertexState.NEW,
+ .addTransition(VertexState.NEW,
+ EnumSet.of(VertexState.NEW),
VertexEventType.V_ROUTE_EVENT,
ROUTE_EVENT_TRANSITION)
- .addTransition(VertexState.NEW, VertexState.NEW,
+ .addTransition(VertexState.NEW,
+ EnumSet.of(VertexState.NEW),
VertexEventType.V_SOURCE_TASK_ATTEMPT_COMPLETED,
SOURCE_TASK_ATTEMPT_COMPLETED_EVENT_TRANSITION)
.addTransition
@@ -342,10 +345,12 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
.addTransition(VertexState.INITIALIZING, VertexState.INITIALIZING,
VertexEventType.V_SOURCE_VERTEX_STARTED,
new SourceVertexStartedTransition())
- .addTransition(VertexState.INITIALIZING, VertexState.INITIALIZING,
+ .addTransition(VertexState.INITIALIZING,
+ EnumSet.of(VertexState.INITIALIZING),
VertexEventType.V_SOURCE_TASK_ATTEMPT_COMPLETED,
SOURCE_TASK_ATTEMPT_COMPLETED_EVENT_TRANSITION)
- .addTransition(VertexState.INITIALIZING, VertexState.INITIALIZING,
+ .addTransition(VertexState.INITIALIZING,
+ EnumSet.of(VertexState.INITIALIZING, VertexState.FAILED),
VertexEventType.V_ROUTE_EVENT,
ROUTE_EVENT_TRANSITION)
.addTransition(VertexState.INITIALIZING, VertexState.KILLED,
@@ -375,15 +380,17 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
EnumSet.of(VertexState.INITED),
VertexEventType.V_ONE_TO_ONE_SOURCE_SPLIT,
new OneToOneSourceSplitTransition())
- .addTransition(VertexState.INITED, VertexState.INITED,
+ .addTransition(VertexState.INITED,
+ EnumSet.of(VertexState.INITED),
VertexEventType.V_SOURCE_TASK_ATTEMPT_COMPLETED,
SOURCE_TASK_ATTEMPT_COMPLETED_EVENT_TRANSITION)
.addTransition(VertexState.INITED,
- EnumSet.of(VertexState.RUNNING, VertexState.INITED),
+ EnumSet.of(VertexState.RUNNING, VertexState.INITED, VertexState.TERMINATING),
VertexEventType.V_START,
new StartTransition())
.addTransition(VertexState.INITED,
- VertexState.INITED, VertexEventType.V_ROUTE_EVENT,
+ EnumSet.of(VertexState.INITED, VertexState.FAILED),
+ VertexEventType.V_ROUTE_EVENT,
ROUTE_EVENT_TRANSITION)
.addTransition(VertexState.INITED, VertexState.KILLED,
VertexEventType.V_TERMINATE,
@@ -396,7 +403,8 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
.addTransition(VertexState.RUNNING, VertexState.RUNNING,
VertexEventType.V_TASK_ATTEMPT_COMPLETED,
TASK_ATTEMPT_COMPLETED_EVENT_TRANSITION)
- .addTransition(VertexState.RUNNING, VertexState.RUNNING,
+ .addTransition(VertexState.RUNNING,
+ EnumSet.of(VertexState.RUNNING, VertexState.TERMINATING),
VertexEventType.V_SOURCE_TASK_ATTEMPT_COMPLETED,
SOURCE_TASK_ATTEMPT_COMPLETED_EVENT_TRANSITION)
.addTransition
@@ -427,7 +435,8 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
INTERNAL_ERROR_TRANSITION)
.addTransition(
VertexState.RUNNING,
- VertexState.RUNNING, VertexEventType.V_ROUTE_EVENT,
+ EnumSet.of(VertexState.RUNNING, VertexState.TERMINATING),
+ VertexEventType.V_ROUTE_EVENT,
ROUTE_EVENT_TRANSITION)
// Transitions from TERMINATING state.
@@ -463,7 +472,8 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
// Ignore-able events
.addTransition(
- VertexState.SUCCEEDED, VertexState.SUCCEEDED,
+ VertexState.SUCCEEDED,
+ EnumSet.of(VertexState.SUCCEEDED, VertexState.FAILED),
// accumulate these in case we get restarted
VertexEventType.V_ROUTE_EVENT,
ROUTE_EVENT_TRANSITION)
@@ -1121,8 +1131,16 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
tasksNotYetScheduled = false;
if (!pendingTaskEvents.isEmpty()) {
LOG.info("Routing pending task events for vertex: " + logIdentifier);
- VertexImpl.ROUTE_EVENT_TRANSITION.transition(this,
- new VertexEventRouteEvent(getVertexId(), pendingTaskEvents));
+ try {
+ handleRoutedTezEvents(this, pendingTaskEvents, false);
+ } catch (AMUserCodeException e) {
+ String msg = "Exception in " + e.getSource() +", vertex=" + logIdentifier;
+ LOG.error(msg, e);
+ addDiagnostic(msg + "," + ExceptionUtils.getStackTrace(e.getCause()));
+ // TODO add test for exception from EdgeManager in TEZ-1689
+ eventHandler.handle(new VertexEventTermination(vertexId, VertexTerminationCause.AM_USERCODE_FAILURE));
+ return;
+ }
pendingTaskEvents.clear();
}
for (TaskWithLocationHint task : tasksToSchedule) {
@@ -1539,7 +1557,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
new VertexCommitStartedEvent(vertex.vertexId,
vertex.clock.getTime())));
} catch (IOException e) {
- LOG.error("Failed to persist commit start event to recovery, vertexId="
+ LOG.error("Failed to persist commit start event to recovery, vertex="
+ vertex.logIdentifier, e);
vertex.trySetTerminationCause(VertexTerminationCause.INTERNAL_ERROR);
return vertex.finished(VertexState.FAILED);
@@ -1611,6 +1629,17 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
vertex.abortVertex(State.FAILED);
return vertex.finished(VertexState.FAILED);
}
+ else if (vertex.terminationCause == VertexTerminationCause.AM_USERCODE_FAILURE) {
+ vertex.setFinishTime();
+ String diagnosticMsg = "Vertex failed/killed due to VertexManager failed. "
+ + "failedTasks:"
+ + vertex.failedTaskCount
+ + " killedTasks:"
+ + vertex.killedTaskCount;
+ LOG.info(diagnosticMsg);
+ vertex.abortVertex(State.FAILED);
+ return vertex.finished(VertexState.FAILED);
+ }
else {
//should never occur
throw new TezUncheckedException("All tasks complete, but cannot determine final state of vertex"
@@ -1643,11 +1672,22 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
}
VertexState finished(VertexState finalState,
- VertexTerminationCause terminationCause) {
+ VertexTerminationCause terminationCause, String diag) {
if (finishTime == 0) setFinishTime();
+ if (terminationCause != null) {
+ trySetTerminationCause(terminationCause);
+ }
+ if (rootInputInitializerManager != null) {
+ rootInputInitializerManager.shutdown();
+ rootInputInitializerManager = null;
+ }
switch (finalState) {
case ERROR:
+ addDiagnostic("Vertex: " + logIdentifier + " error due to:" + terminationCause);
+ if (!StringUtils.isEmpty(diag)) {
+ addDiagnostic(diag);
+ }
eventHandler.handle(new DAGEvent(getDAGId(),
DAGEventType.INTERNAL_ERROR));
try {
@@ -1658,6 +1698,10 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
break;
case KILLED:
case FAILED:
+ addDiagnostic("Vertex " + logIdentifier + " killed/failed due to:" + terminationCause);
+ if (!StringUtils.isEmpty(diag)) {
+ addDiagnostic(diag);
+ }
eventHandler.handle(new DAGEventVertexCompleted(getVertexId(),
finalState, terminationCause));
try {
@@ -1686,7 +1730,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
}
VertexState finished(VertexState finalState) {
- return finished(finalState, null);
+ return finished(finalState, null, null);
}
@@ -1740,7 +1784,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
try {
initializeCommitters();
} catch (Exception e) {
- LOG.warn("Vertex Committer init failed, vertexId=" + logIdentifier, e);
+ LOG.warn("Vertex Committer init failed, vertex=" + logIdentifier, e);
addDiagnostic("Vertex init failed : "
+ ExceptionUtils.getStackTrace(e));
trySetTerminationCause(VertexTerminationCause.INIT_FAILURE);
@@ -1876,7 +1920,15 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
assignVertexManager();
- vertexManager.initialize();
+ try {
+ vertexManager.initialize();
+ } catch (AMUserCodeException e) {
+ String msg = "Exception in " + e.getSource()+ ", vertex:" + logIdentifier;
+ LOG.error(msg, e);
+ finished(VertexState.FAILED, VertexTerminationCause.AM_USERCODE_FAILURE,
+ msg + "," + ExceptionUtils.getStackTrace(e.getCause()));
+ return VertexState.FAILED;
+ }
// Setup tasks early if possible. If the VertexManager is not being used
// to set parallelism, sending events to Tasks is safe (and less confusing
@@ -2065,10 +2117,12 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
try {
vertex.initializeCommitters();
} catch (Exception e) {
- LOG.info("Failed to initialize committers"
- + ", vertex=" + vertex.logIdentifier, e);
+ String msg = "Failed to initialize committers"
+ + ", vertex=" + vertex.logIdentifier + ","
+ + ExceptionUtils.getStackTrace(e);
+ LOG.error(msg);
vertex.finished(VertexState.FAILED,
- VertexTerminationCause.INIT_FAILURE);
+ VertexTerminationCause.INIT_FAILURE, msg);
endState = VertexState.FAILED;
break;
}
@@ -2095,9 +2149,12 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
try {
vertex.initializeCommitters();
} catch (Exception e) {
- LOG.info("Failed to initialize committers", e);
+ String msg = "Failed to initialize committers"
+ + ", vertex=" + vertex.logIdentifier + ","
+ + ExceptionUtils.getStackTrace(e);
+ LOG.error(msg);
vertex.finished(VertexState.FAILED,
- VertexTerminationCause.INIT_FAILURE);
+ VertexTerminationCause.INIT_FAILURE, msg);
endState = VertexState.FAILED;
break;
}
@@ -2105,10 +2162,11 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
// if commit in progress and desired state is not a succeeded one,
// move to failed
if (vertex.recoveryCommitInProgress) {
- LOG.info("Recovered vertex was in the middle of a commit"
- + ", failing Vertex=" + vertex.logIdentifier);
+ String msg = "Recovered vertex was in the middle of a commit"
+ + ", failing Vertex=" + vertex.logIdentifier;
+ LOG.warn(msg);
vertex.finished(VertexState.FAILED,
- VertexTerminationCause.COMMIT_FAILURE);
+ VertexTerminationCause.COMMIT_FAILURE, msg);
endState = VertexState.FAILED;
break;
}
@@ -2118,8 +2176,16 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
vertex.eventHandler.handle(
new TaskEventRecoverTask(task.getTaskId()));
}
- vertex.vertexManager.onVertexStarted(vertex.pendingReportedSrcCompletions);
- endState = VertexState.RUNNING;
+ try {
+ vertex.vertexManager.onVertexStarted(vertex.pendingReportedSrcCompletions);
+ endState = VertexState.RUNNING;
+ } catch (AMUserCodeException e) {
+ String msg = "Exception in " + e.getSource() + ", vertex:" + vertex.getLogIdentifier();
+ LOG.error(msg, e);
+ vertex.finished(VertexState.FAILED, VertexTerminationCause.AM_USERCODE_FAILURE,
+ msg + ", " + ExceptionUtils.getStackTrace(e.getCause()));
+ endState = VertexState.FAILED;
+ }
} else {
// why succeeded here
endState = VertexState.SUCCEEDED;
@@ -2132,13 +2198,14 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
if (vertex.recoveredState == VertexState.SUCCEEDED
&& vertex.hasCommitter
&& vertex.summaryCompleteSeen && !vertex.vertexCompleteSeen) {
- LOG.warn("Cannot recover vertex as all recovery events not"
- + " found, vertexId=" + vertex.logIdentifier
+ String msg = "Cannot recover vertex as all recovery events not"
+ + " found, vertex=" + vertex.logIdentifier
+ ", hasCommitters=" + vertex.hasCommitter
+ ", summaryCompletionSeen=" + vertex.summaryCompleteSeen
- + ", finalCompletionSeen=" + vertex.vertexCompleteSeen);
+ + ", finalCompletionSeen=" + vertex.vertexCompleteSeen;
+ LOG.warn(msg);
vertex.finished(VertexState.FAILED,
- VertexTerminationCause.COMMIT_FAILURE);
+ VertexTerminationCause.COMMIT_FAILURE, msg);
endState = VertexState.FAILED;
} else {
vertex.tasksNotYetScheduled = false;
@@ -2161,8 +2228,16 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
new TaskEventRecoverTask(task.getTaskId(),
taskState));
}
- vertex.vertexManager.onVertexStarted(vertex.pendingReportedSrcCompletions);
- endState = VertexState.RUNNING;
+ try {
+ vertex.vertexManager.onVertexStarted(vertex.pendingReportedSrcCompletions);
+ endState = VertexState.RUNNING;
+ } catch (AMUserCodeException e) {
+ String msg = "Exception in " + e.getSource() +", vertex:" + vertex.getLogIdentifier();
+ LOG.error(msg, e);
+ vertex.finished(VertexState.FAILED, VertexTerminationCause.AM_USERCODE_FAILURE,
+ msg + "," + ExceptionUtils.getStackTrace(e.getCause()));
+ endState = VertexState.FAILED;
+ }
} else {
endState = vertex.recoveredState;
vertex.finished(endState);
@@ -2425,19 +2500,20 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
try {
vertex.initializeCommitters();
} catch (Exception e) {
- LOG.info("Failed to initialize committers, vertex="
- + vertex.logIdentifier, e);
+ String msg = "Failed to initialize committers, vertex="
+ + vertex.logIdentifier + "," + ExceptionUtils.getStackTrace(e);
+ LOG.error(msg);
vertex.finished(VertexState.FAILED,
- VertexTerminationCause.INIT_FAILURE);
+ VertexTerminationCause.INIT_FAILURE, msg);
endState = VertexState.FAILED;
break;
}
if (!vertex.setParallelism(0,
null, vertex.recoveredSourceEdgeManagers, vertex.recoveredRootInputSpecUpdates, true)) {
- LOG.info("Failed to recover edge managers, vertex="
- + vertex.logIdentifier);
+ String msg = "Failed to recover edge managers, vertex=" + vertex.logIdentifier;
+ LOG.error(msg);
vertex.finished(VertexState.FAILED,
- VertexTerminationCause.INIT_FAILURE);
+ VertexTerminationCause.INIT_FAILURE, msg);
endState = VertexState.FAILED;
break;
}
@@ -2469,24 +2545,27 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
LOG.info("Recovered vertex was in the middle of a commit"
+ ", failing Vertex=" + vertex.logIdentifier);
vertex.finished(VertexState.FAILED,
- VertexTerminationCause.COMMIT_FAILURE);
+ VertexTerminationCause.COMMIT_FAILURE, null);
endState = VertexState.FAILED;
break;
}
try {
vertex.initializeCommitters();
} catch (Exception e) {
- LOG.info("Failed to initialize committers", e);
+ String msg = "Failed to initialize committers, vertex="
+ + vertex.logIdentifier + "," + ExceptionUtils.getStackTrace(e);
+ LOG.error(msg);
vertex.finished(VertexState.FAILED,
- VertexTerminationCause.INIT_FAILURE);
+ VertexTerminationCause.INIT_FAILURE, msg);
endState = VertexState.FAILED;
break;
}
if (!vertex.setParallelism(0, null, vertex.recoveredSourceEdgeManagers,
vertex.recoveredRootInputSpecUpdates, true)) {
- LOG.info("Failed to recover edge managers");
+ String msg = "Failed to recover edge managers for vertex:" + vertex.logIdentifier;
+ LOG.error(msg);
vertex.finished(VertexState.FAILED,
- VertexTerminationCause.INIT_FAILURE);
+ VertexTerminationCause.INIT_FAILURE, msg);
endState = VertexState.FAILED;
break;
}
@@ -2496,8 +2575,16 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
vertex.eventHandler.handle(
new TaskEventRecoverTask(task.getTaskId()));
}
- vertex.vertexManager.onVertexStarted(vertex.pendingReportedSrcCompletions);
- endState = VertexState.RUNNING;
+ try {
+ vertex.vertexManager.onVertexStarted(vertex.pendingReportedSrcCompletions);
+ endState = VertexState.RUNNING;
+ } catch (AMUserCodeException e) {
+ String msg = "Exception in " + e.getSource() + ", vertex=" + vertex.getLogIdentifier();
+ LOG.error(msg, e);
+ vertex.finished(VertexState.FAILED, VertexTerminationCause.AM_USERCODE_FAILURE,
+ msg + "," + ExceptionUtils.getStackTrace(e.getCause()));
+ endState = VertexState.FAILED;
+ }
} else {
endState = VertexState.SUCCEEDED;
vertex.finished(endState);
@@ -2528,8 +2615,16 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
taskState));
}
// Wait for all tasks to recover and report back
- vertex.vertexManager.onVertexStarted(vertex.pendingReportedSrcCompletions);
- endState = VertexState.RUNNING;
+ try {
+ vertex.vertexManager.onVertexStarted(vertex.pendingReportedSrcCompletions);
+ endState = VertexState.RUNNING;
+ } catch (AMUserCodeException e) {
+ String msg = "Exception in " + e.getSource() +", vertex:" + vertex.getLogIdentifier();
+ LOG.error(msg, e);
+ vertex.finished(VertexState.FAILED, VertexTerminationCause.AM_USERCODE_FAILURE,
+ msg + "," + ExceptionUtils.getStackTrace(e.getCause()));
+ endState = VertexState.FAILED;
+ }
} else {
endState = vertex.recoveredState;
vertex.finished(endState);
@@ -2564,10 +2659,16 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
vertex.routeRecoveredEvents(endState, vertex.recoveredEvents);
vertex.recoveredEvents.clear();
if (!vertex.pendingRouteEvents.isEmpty()) {
- VertexImpl.ROUTE_EVENT_TRANSITION.transition(vertex,
- new VertexEventRouteEvent(vertex.getVertexId(),
- vertex.pendingRouteEvents));
- vertex.pendingRouteEvents.clear();
+ try {
+ handleRoutedTezEvents(vertex, vertex.pendingRouteEvents, false);
+ vertex.pendingRouteEvents.clear();
+ } catch (AMUserCodeException e) {
+ String msg = "Exception in " + e.getSource() + ", vertex=" + vertex.getLogIdentifier();
+ LOG.error(msg, e);
+ vertex.finished(VertexState.FAILED, VertexTerminationCause.AM_USERCODE_FAILURE,
+ msg + "," + ExceptionUtils.getStackTrace(e.getCause()));
+ endState = VertexState.FAILED;
+ }
}
} else {
// Ensure no recovered events
@@ -2776,13 +2877,22 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
VertexEventRootInputInitialized liInitEvent = (VertexEventRootInputInitialized) event;
VertexState state = vertex.getState();
if (state == VertexState.INITIALIZING) {
- List<TezEvent> inputInfoEvents =
- vertex.vertexManager.onRootVertexInitialized(
- liInitEvent.getInputName(),
- vertex.getAdditionalInputs().get(liInitEvent.getInputName())
- .getIODescriptor(), liInitEvent.getEvents());
- if (inputInfoEvents != null && !inputInfoEvents.isEmpty()) {
- ROUTE_EVENT_TRANSITION.transition(vertex, new VertexEventRouteEvent(vertex.vertexId, inputInfoEvents));
+ try {
+ List<TezEvent> inputInfoEvents =
+ vertex.vertexManager.onRootVertexInitialized(
+ liInitEvent.getInputName(),
+ vertex.getAdditionalInputs().get(liInitEvent.getInputName())
+ .getIODescriptor(), liInitEvent.getEvents());
+ if (inputInfoEvents != null && !inputInfoEvents.isEmpty()) {
+ VertexImpl.handleRoutedTezEvents(vertex, inputInfoEvents, false);
+ }
+ } catch (AMUserCodeException e) {
+ String msg = "Exception in " + e.getSource() + ", vertex:" + vertex.getLogIdentifier();
+ LOG.error(msg, e);
+ vertex.finished(VertexState.FAILED,
+ VertexTerminationCause.AM_USERCODE_FAILURE, msg
+ + "," + ExceptionUtils.getStackTrace(e.getCause()));
+ return VertexState.FAILED;
}
}
@@ -2962,7 +3072,15 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
"Vertex must be inited " + logIdentifier);
startedTime = clock.getTime();
- vertexManager.onVertexStarted(pendingReportedSrcCompletions);
+ try {
+ vertexManager.onVertexStarted(pendingReportedSrcCompletions);
+ } catch (AMUserCodeException e) {
+ String msg = "Exception in " + e.getSource() +", vertex=" + logIdentifier;
+ LOG.error(msg, e);
+ addDiagnostic(msg + "," + ExceptionUtils.getStackTrace(e.getCause()));
+ tryEnactKill(VertexTerminationCause.AM_USERCODE_FAILURE, TaskTerminationCause.AM_USERCODE_FAILURE);
+ return VertexState.TERMINATING;
+ }
pendingReportedSrcCompletions.clear();
logJobHistoryVertexStartedEvent();
@@ -3050,22 +3168,14 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
@Override
public void transition(VertexImpl vertex, VertexEvent event) {
VertexEventRootInputFailed fe = (VertexEventRootInputFailed) event;
- vertex.trySetTerminationCause(VertexTerminationCause.INIT_FAILURE);
- vertex.addDiagnostic("Vertex Input: " + fe.getInputName()
- + " initializer failed.");
+ String msg = "Vertex Input: " + fe.getInputName()
+ + " initializer failed, vertex=" + vertex.getLogIdentifier();
if (fe.getError() != null) {
- LOG.error("Vertex Input: " + fe.getInputName() + " initializer failed",
- fe.getError());
- if (fe.getError().getMessage() != null) {
- vertex.addDiagnostic(fe.getError().getMessage());
- }
- }
- if (vertex.rootInputInitializerManager != null) {
- vertex.rootInputInitializerManager.shutdown();
- vertex.rootInputInitializerManager = null;
+ msg = msg + ExceptionUtils.getStackTrace(fe.getError());
}
+ LOG.error(msg);
vertex.finished(VertexState.FAILED,
- VertexTerminationCause.ROOT_INPUT_INIT_FAILURE);
+ VertexTerminationCause.ROOT_INPUT_INIT_FAILURE, msg);
}
}
@@ -3099,10 +3209,6 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
@Override
public void transition(VertexImpl vertex, VertexEvent event) {
super.transition(vertex, event);
- if (vertex.rootInputInitializerManager != null) {
- vertex.rootInputInitializerManager.shutdown();
- vertex.rootInputInitializerManager = null;
- }
}
}
@@ -3121,6 +3227,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
case INVALID_NUM_OF_TASKS:
case INIT_FAILURE:
case INTERNAL_ERROR:
+ case AM_USERCODE_FAILURE:
case OTHER_VERTEX_FAILURE: vertex.tryEnactKill(trigger, TaskTerminationCause.OTHER_VERTEX_FAILURE); break;
default://should not occur
throw new TezUncheckedException("VertexKilledTransition: event.terminationCause is unexpected: " + trigger);
@@ -3136,9 +3243,9 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
* completed.
*/
private static class SourceTaskAttemptCompletedEventTransition implements
- SingleArcTransition<VertexImpl, VertexEvent> {
+ MultipleArcTransition<VertexImpl, VertexEvent, VertexState> {
@Override
- public void transition(VertexImpl vertex, VertexEvent event) {
+ public VertexState transition(VertexImpl vertex, VertexEvent event) {
VertexEventTaskAttemptCompleted completionEvent =
((VertexEventSourceTaskAttemptCompleted) event).getCompletionEvent();
LOG.info("Source task attempt completed for vertex: " + vertex.getVertexId()
@@ -3152,13 +3259,22 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
vertex.numSuccessSourceAttemptCompletions++;
if (vertex.getState() == VertexState.RUNNING) {
- // Inform the vertex manager about the source task completing.
- vertex.vertexManager.onSourceTaskCompleted(completionEvent
- .getTaskAttemptId().getTaskID());
+ try {
+ // Inform the vertex manager about the source task completing.
+ vertex.vertexManager.onSourceTaskCompleted(completionEvent
+ .getTaskAttemptId().getTaskID());
+ } catch (AMUserCodeException e) {
+ String msg = "Exception in " + e.getSource() + ", vertex:" + vertex.getLogIdentifier();
+ LOG.error(msg, e);
+ vertex.addDiagnostic(msg + "," + ExceptionUtils.getStackTrace(e.getCause()));
+ vertex.tryEnactKill(VertexTerminationCause.AM_USERCODE_FAILURE, TaskTerminationCause.AM_USERCODE_FAILURE);
+ return VertexState.TERMINATING;
+ }
} else {
vertex.pendingReportedSrcCompletions.add(completionEvent.getTaskAttemptId());
}
}
+ return vertex.getState();
}
}
@@ -3275,9 +3391,8 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
" completed with state " + vEvent.getState() + " after vertex succeeded.";
}
LOG.info(diagnosticMsg);
- vertex.addDiagnostic(diagnosticMsg);
vertex.abortVertex(finalStatus);
- vertex.finished(finalState, VertexTerminationCause.OWN_TASK_FAILURE);
+ vertex.finished(finalState, VertexTerminationCause.OWN_TASK_FAILURE, diagnosticMsg);
return finalState;
}
}
@@ -3304,9 +3419,8 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
LOG.info(diagnosticMsg);
vertex.tryEnactKill(VertexTerminationCause.OWN_TASK_FAILURE,
TaskTerminationCause.OWN_TASK_FAILURE);
- vertex.addDiagnostic(diagnosticMsg);
vertex.abortVertex(VertexStatus.State.FAILED);
- vertex.finished(VertexState.FAILED, VertexTerminationCause.OWN_TASK_FAILURE);
+ vertex.finished(VertexState.FAILED, VertexTerminationCause.OWN_TASK_FAILURE, diagnosticMsg);
return VertexState.FAILED;
}
}
@@ -3341,185 +3455,203 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
// }
private static class RouteEventTransition implements
- SingleArcTransition<VertexImpl, VertexEvent> {
+ MultipleArcTransition<VertexImpl, VertexEvent, VertexState> {
@Override
- public void transition(VertexImpl vertex, VertexEvent event) {
+ public VertexState transition(VertexImpl vertex, VertexEvent event) {
VertexEventRouteEvent rEvent = (VertexEventRouteEvent) event;
boolean recovered = rEvent.isRecovered();
List<TezEvent> tezEvents = rEvent.getEvents();
+ try {
+ VertexImpl.handleRoutedTezEvents(vertex, tezEvents, recovered);
+ } catch (AMUserCodeException e) {
+ String msg = "Exception in " + e.getSource() + ", vertex=" + vertex.getLogIdentifier();
+ LOG.error(msg, e);
+ if (vertex.getState() == VertexState.RUNNING) {
+ vertex.addDiagnostic(msg + "," + ExceptionUtils.getStackTrace(e.getCause()));
+ vertex.tryEnactKill(VertexTerminationCause.AM_USERCODE_FAILURE, TaskTerminationCause.AM_USERCODE_FAILURE);
+ return VertexState.TERMINATING;
+ } else {
+ vertex.finished(VertexState.FAILED, VertexTerminationCause.AM_USERCODE_FAILURE,
+ msg + "," + ExceptionUtils.getStackTrace(e.getCause()));
+ return VertexState.FAILED;
+ }
+ }
+ return vertex.getState();
+ }
+ }
- if (vertex.getAppContext().isRecoveryEnabled()
- && !recovered
- && !tezEvents.isEmpty()) {
- List<TezEvent> recoveryEvents =
- Lists.newArrayList();
- for (TezEvent tezEvent : tezEvents) {
- if (!isEventFromVertex(vertex, tezEvent.getSourceInfo())) {
- continue;
- }
- if (tezEvent.getEventType().equals(EventType.COMPOSITE_DATA_MOVEMENT_EVENT)
- || tezEvent.getEventType().equals(EventType.DATA_MOVEMENT_EVENT)
- || tezEvent.getEventType().equals(EventType.ROOT_INPUT_DATA_INFORMATION_EVENT)
- || tezEvent.getEventType().equals(EventType.ROOT_INPUT_INITIALIZER_EVENT)) {
- recoveryEvents.add(tezEvent);
- }
+ private static void handleRoutedTezEvents(VertexImpl vertex, List<TezEvent> tezEvents, boolean recovered) throws AMUserCodeException {
+ if (vertex.getAppContext().isRecoveryEnabled()
+ && !recovered
+ && !tezEvents.isEmpty()) {
+ List<TezEvent> recoveryEvents =
+ Lists.newArrayList();
+ for (TezEvent tezEvent : tezEvents) {
+ if (!isEventFromVertex(vertex, tezEvent.getSourceInfo())) {
+ continue;
}
- if (!recoveryEvents.isEmpty()) {
- VertexRecoverableEventsGeneratedEvent historyEvent =
- new VertexRecoverableEventsGeneratedEvent(vertex.vertexId,
- recoveryEvents);
- vertex.appContext.getHistoryHandler().handle(
- new DAGHistoryEvent(vertex.getDAGId(), historyEvent));
+ if (tezEvent.getEventType().equals(EventType.COMPOSITE_DATA_MOVEMENT_EVENT)
+ || tezEvent.getEventType().equals(EventType.DATA_MOVEMENT_EVENT)
+ || tezEvent.getEventType().equals(EventType.ROOT_INPUT_DATA_INFORMATION_EVENT)
+ || tezEvent.getEventType().equals(EventType.ROOT_INPUT_INITIALIZER_EVENT)) {
+ recoveryEvents.add(tezEvent);
}
}
- for(TezEvent tezEvent : tezEvents) {
- if (LOG.isDebugEnabled()) {
- LOG.debug("Vertex: " + vertex.getName() + " routing event: "
- + tezEvent.getEventType()
- + " Recovered:" + recovered);
- }
- EventMetaData sourceMeta = tezEvent.getSourceInfo();
- switch(tezEvent.getEventType()) {
- case INPUT_FAILED_EVENT:
- case DATA_MOVEMENT_EVENT:
- case COMPOSITE_DATA_MOVEMENT_EVENT:
- {
- if (isEventFromVertex(vertex, sourceMeta)) {
- // event from this vertex. send to destination vertex
- TezTaskAttemptID srcTaId = sourceMeta.getTaskAttemptID();
- if (tezEvent.getEventType() == EventType.DATA_MOVEMENT_EVENT) {
- ((DataMovementEvent) tezEvent.getEvent()).setVersion(srcTaId.getId());
- } else if (tezEvent.getEventType() == EventType.COMPOSITE_DATA_MOVEMENT_EVENT) {
- ((CompositeDataMovementEvent) tezEvent.getEvent()).setVersion(srcTaId.getId());
- } else {
- ((InputFailedEvent) tezEvent.getEvent()).setVersion(srcTaId.getId());
- }
- Vertex destVertex = vertex.getDAG().getVertex(sourceMeta.getEdgeVertexName());
- Edge destEdge = vertex.targetVertices.get(destVertex);
- if (destEdge == null) {
- throw new TezUncheckedException("Bad destination vertex: " +
- sourceMeta.getEdgeVertexName() + " for event vertex: " +
- vertex.getVertexId());
- }
- vertex.eventHandler.handle(new VertexEventRouteEvent(destVertex
- .getVertexId(), Collections.singletonList(tezEvent)));
+ if (!recoveryEvents.isEmpty()) {
+ VertexRecoverableEventsGeneratedEvent historyEvent =
+ new VertexRecoverableEventsGeneratedEvent(vertex.vertexId,
+ recoveryEvents);
+ vertex.appContext.getHistoryHandler().handle(
+ new DAGHistoryEvent(vertex.getDAGId(), historyEvent));
+ }
+ }
+ for(TezEvent tezEvent : tezEvents) {
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Vertex: " + vertex.getName() + " routing event: "
+ + tezEvent.getEventType()
+ + " Recovered:" + recovered);
+ }
+ EventMetaData sourceMeta = tezEvent.getSourceInfo();
+ switch(tezEvent.getEventType()) {
+ case INPUT_FAILED_EVENT:
+ case DATA_MOVEMENT_EVENT:
+ case COMPOSITE_DATA_MOVEMENT_EVENT:
+ {
+ if (isEventFromVertex(vertex, sourceMeta)) {
+ // event from this vertex. send to destination vertex
+ TezTaskAttemptID srcTaId = sourceMeta.getTaskAttemptID();
+ if (tezEvent.getEventType() == EventType.DATA_MOVEMENT_EVENT) {
+ ((DataMovementEvent) tezEvent.getEvent()).setVersion(srcTaId.getId());
+ } else if (tezEvent.getEventType() == EventType.COMPOSITE_DATA_MOVEMENT_EVENT) {
+ ((CompositeDataMovementEvent) tezEvent.getEvent()).setVersion(srcTaId.getId());
} else {
- // event not from this vertex. must have come from source vertex.
- // send to tasks
- if (vertex.tasksNotYetScheduled) {
- vertex.pendingTaskEvents.add(tezEvent);
- } else {
- Edge srcEdge = vertex.sourceVertices.get(vertex.getDAG().getVertex(
- sourceMeta.getTaskVertexName()));
- if (srcEdge == null) {
- throw new TezUncheckedException("Bad source vertex: " +
- sourceMeta.getTaskVertexName() + " for destination vertex: " +
- vertex.getVertexId());
- }
- srcEdge.sendTezEventToDestinationTasks(tezEvent);
- }
+ ((InputFailedEvent) tezEvent.getEvent()).setVersion(srcTaId.getId());
}
- }
- break;
- case ROOT_INPUT_DATA_INFORMATION_EVENT:
- if (vertex.tasksNotYetScheduled) {
- vertex.pendingTaskEvents.add(tezEvent);
- } else {
- checkEventSourceMetadata(vertex, sourceMeta);
- InputDataInformationEvent riEvent = (InputDataInformationEvent) tezEvent
- .getEvent();
- Task targetTask = vertex.getTask(riEvent.getTargetIndex());
- targetTask.registerTezEvent(tezEvent);
- }
- break;
- case VERTEX_MANAGER_EVENT:
- {
- // VM events on task success only can be changed as part of TEZ-1532
- VertexManagerEvent vmEvent = (VertexManagerEvent) tezEvent.getEvent();
- Vertex target = vertex.getDAG().getVertex(vmEvent.getTargetVertexName());
- Preconditions.checkArgument(target != null,
- "Event sent to unkown vertex: " + vmEvent.getTargetVertexName());
- if (target == vertex) {
- vertex.vertexManager.onVertexManagerEventReceived(vmEvent);
- } else {
- checkEventSourceMetadata(vertex, sourceMeta);
- vertex.eventHandler.handle(new VertexEventRouteEvent(target
- .getVertexId(), Collections.singletonList(tezEvent)));
- }
- }
- break;
- case ROOT_INPUT_INITIALIZER_EVENT:
- {
- InputInitializerEvent riEvent = (InputInitializerEvent) tezEvent.getEvent();
- Vertex target = vertex.getDAG().getVertex(riEvent.getTargetVertexName());
- Preconditions.checkArgument(target != null,
- "Event sent to unknown vertex: " + riEvent.getTargetVertexName());
- riEvent.setSourceVertexName(tezEvent.getSourceInfo().getTaskVertexName());
- if (target == vertex) {
- if (vertex.rootInputDescriptors == null ||
- !vertex.rootInputDescriptors.containsKey(riEvent.getTargetInputName())) {
- throw new TezUncheckedException(
- "InputInitializerEvent targeted at unknown initializer on vertex " +
- vertex.logIdentifier + ", Event=" + riEvent);
+ Vertex destVertex = vertex.getDAG().getVertex(sourceMeta.getEdgeVertexName());
+ Edge destEdge = vertex.targetVertices.get(destVertex);
+ if (destEdge == null) {
+ throw new TezUncheckedException("Bad destination vertex: " +
+ sourceMeta.getEdgeVertexName() + " for event vertex: " +
+ vertex.getVertexId());
}
- if (vertex.getState() == VertexState.NEW) {
- vertex.pendingInitializerEvents.add(tezEvent);
- } else if (vertex.getState() == VertexState.INITIALIZING) {
- vertex.rootInputInitializerManager.handleInitializerEvents(Collections.singletonList(tezEvent));
+ vertex.eventHandler.handle(new VertexEventRouteEvent(destVertex
+ .getVertexId(), Collections.singletonList(tezEvent)));
+ } else {
+ // event not from this vertex. must have come from source vertex.
+ // send to tasks
+ if (vertex.tasksNotYetScheduled) {
+ vertex.pendingTaskEvents.add(tezEvent);
} else {
- // Currently, INITED and subsequent states means Initializer complete / failure
- if (LOG.isDebugEnabled()) {
- LOG.debug("Dropping event" + tezEvent + " since state is not INITIALIZING in " + vertex.getLogIdentifier() + ", state=" + vertex.getState());
+ Edge srcEdge = vertex.sourceVertices.get(vertex.getDAG().getVertex(
+ sourceMeta.getTaskVertexName()));
+ if (srcEdge == null) {
+ throw new TezUncheckedException("Bad source vertex: " +
+ sourceMeta.getTaskVertexName() + " for destination vertex: " +
+ vertex.getVertexId());
}
+ srcEdge.sendTezEventToDestinationTasks(tezEvent);
}
- } else {
- checkEventSourceMetadata(vertex, sourceMeta);
- vertex.eventHandler.handle(new VertexEventRouteEvent(target.getVertexId(),
- Collections.singletonList(tezEvent)));
}
}
- break;
- case INPUT_READ_ERROR_EVENT:
- {
- checkEventSourceMetadata(vertex, sourceMeta);
- Edge srcEdge = vertex.sourceVertices.get(vertex.getDAG().getVertex(
- sourceMeta.getEdgeVertexName()));
- srcEdge.sendTezEventToSourceTasks(tezEvent);
- }
- break;
- case TASK_STATUS_UPDATE_EVENT:
- {
- checkEventSourceMetadata(vertex, sourceMeta);
- TaskStatusUpdateEvent sEvent =
- (TaskStatusUpdateEvent) tezEvent.getEvent();
- vertex.getEventHandler().handle(
- new TaskAttemptEventStatusUpdate(sourceMeta.getTaskAttemptID(),
- sEvent));
- }
- break;
- case TASK_ATTEMPT_COMPLETED_EVENT:
- {
- checkEventSourceMetadata(vertex, sourceMeta);
- vertex.getEventHandler().handle(
- new TaskAttemptEvent(sourceMeta.getTaskAttemptID(),
- TaskAttemptEventType.TA_DONE));
+ break;
+ case ROOT_INPUT_DATA_INFORMATION_EVENT:
+ if (vertex.tasksNotYetScheduled) {
+ vertex.pendingTaskEvents.add(tezEvent);
+ } else {
+ checkEventSourceMetadata(vertex, sourceMeta);
+ InputDataInformationEvent riEvent = (InputDataInformationEvent) tezEvent
+ .getEvent();
+ Task targetTask = vertex.getTask(riEvent.getTargetIndex());
+ targetTask.registerTezEvent(tezEvent);
+ }
+ break;
+ case VERTEX_MANAGER_EVENT:
+ {
+ // VM events on task success only can be changed as part of TEZ-1532
+ VertexManagerEvent vmEvent = (VertexManagerEvent) tezEvent.getEvent();
+ Vertex target = vertex.getDAG().getVertex(vmEvent.getTargetVertexName());
+ Preconditions.checkArgument(target != null,
+ "Event sent to unkown vertex: " + vmEvent.getTargetVertexName());
+ if (target == vertex) {
+ vertex.vertexManager.onVertexManagerEventReceived(vmEvent);
+ } else {
+ checkEventSourceMetadata(vertex, sourceMeta);
+ vertex.eventHandler.handle(new VertexEventRouteEvent(target
+ .getVertexId(), Collections.singletonList(tezEvent)));
+ }
+ }
+ break;
+ case ROOT_INPUT_INITIALIZER_EVENT:
+ {
+ InputInitializerEvent riEvent = (InputInitializerEvent) tezEvent.getEvent();
+ Vertex target = vertex.getDAG().getVertex(riEvent.getTargetVertexName());
+ Preconditions.checkArgument(target != null,
+ "Event sent to unknown vertex: " + riEvent.getTargetVertexName());
+ riEvent.setSourceVertexName(tezEvent.getSourceInfo().getTaskVertexName());
+ if (target == vertex) {
+ if (vertex.rootInputDescriptors == null ||
+ !vertex.rootInputDescriptors.containsKey(riEvent.getTargetInputName())) {
+ throw new TezUncheckedException(
+ "InputInitializerEvent targeted at unknown initializer on vertex " +
+ vertex.logIdentifier + ", Event=" + riEvent);
}
- break;
- case TASK_ATTEMPT_FAILED_EVENT:
- {
- checkEventSourceMetadata(vertex, sourceMeta);
- TaskAttemptFailedEvent taskFailedEvent =
- (TaskAttemptFailedEvent) tezEvent.getEvent();
- vertex.getEventHandler().handle(
- new TaskAttemptEventAttemptFailed(sourceMeta.getTaskAttemptID(),
- TaskAttemptEventType.TA_FAILED,
- "Error: " + taskFailedEvent.getDiagnostics()));
+ if (vertex.getState() == VertexState.NEW) {
+ vertex.pendingInitializerEvents.add(tezEvent);
+ } else if (vertex.getState() == VertexState.INITIALIZING) {
+ vertex.rootInputInitializerManager.handleInitializerEvents(Collections.singletonList(tezEvent));
+ } else {
+ // Currently, INITED and subsequent states means Initializer complete / failure
+ if (LOG.isDebugEnabled()) {
+ LOG.debug("Dropping event" + tezEvent + " since state is not INITIALIZING in " + vertex.getLogIdentifier() + ", state=" + vertex.getState());
+ }
}
- break;
- default:
- throw new TezUncheckedException("Unhandled tez event type: "
- + tezEvent.getEventType());
+ } else {
+ checkEventSourceMetadata(vertex, sourceMeta);
+ vertex.eventHandler.handle(new VertexEventRouteEvent(target.getVertexId(),
+ Collections.singletonList(tezEvent)));
+ }
+ }
+ break;
+ case INPUT_READ_ERROR_EVENT:
+ {
+ checkEventSourceMetadata(vertex, sourceMeta);
+ Edge srcEdge = vertex.sourceVertices.get(vertex.getDAG().getVertex(
+ sourceMeta.getEdgeVertexName()));
+ srcEdge.sendTezEventToSourceTasks(tezEvent);
}
+ break;
+ case TASK_STATUS_UPDATE_EVENT:
+ {
+ checkEventSourceMetadata(vertex, sourceMeta);
+ TaskStatusUpdateEvent sEvent =
+ (TaskStatusUpdateEvent) tezEvent.getEvent();
+ vertex.getEventHandler().handle(
+ new TaskAttemptEventStatusUpdate(sourceMeta.getTaskAttemptID(),
+ sEvent));
+ }
+ break;
+ case TASK_ATTEMPT_COMPLETED_EVENT:
+ {
+ checkEventSourceMetadata(vertex, sourceMeta);
+ vertex.getEventHandler().handle(
+ new TaskAttemptEvent(sourceMeta.getTaskAttemptID(),
+ TaskAttemptEventType.TA_DONE));
+ }
+ break;
+ case TASK_ATTEMPT_FAILED_EVENT:
+ {
+ checkEventSourceMetadata(vertex, sourceMeta);
+ TaskAttemptFailedEvent taskFailedEvent =
+ (TaskAttemptFailedEvent) tezEvent.getEvent();
+ vertex.getEventHandler().handle(
+ new TaskAttemptEventAttemptFailed(sourceMeta.getTaskAttemptID(),
+ TaskAttemptEventType.TA_FAILED,
+ "Error: " + taskFailedEvent.getDiagnostics()));
+ }
+ break;
+ default:
+ throw new TezUncheckedException("Unhandled tez event type: "
+ + tezEvent.getEventType());
}
}
}
http://git-wip-us.apache.org/repos/asf/tez/blob/eee882e9/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexManager.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexManager.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexManager.java
index 8488f85..892aea2 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexManager.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexManager.java
@@ -49,6 +49,7 @@ import org.apache.tez.dag.app.AppContext;
import org.apache.tez.dag.app.dag.Task;
import org.apache.tez.dag.app.dag.TaskAttempt;
import org.apache.tez.dag.app.dag.Vertex;
+import org.apache.tez.dag.app.dag.impl.AMUserCodeException.Source;
import org.apache.tez.dag.records.TezTaskAttemptID;
import org.apache.tez.dag.records.TezTaskID;
import org.apache.tez.runtime.api.Event;
@@ -227,17 +228,21 @@ public class VertexManager {
return plugin;
}
- public void initialize() {
+ public void initialize() throws AMUserCodeException {
pluginContext = new VertexManagerPluginContextImpl();
if (pluginDesc != null) {
plugin = ReflectionUtils.createClazzInstance(pluginDesc.getClassName(),
new Class[]{VertexManagerPluginContext.class}, new Object[]{pluginContext});
payload = pluginDesc.getUserPayload();
}
- plugin.initialize();
+ try {
+ plugin.initialize();
+ } catch (Exception e) {
+ throw new AMUserCodeException(Source.VertexManager, e);
+ }
}
- public void onVertexStarted(List<TezTaskAttemptID> completions) {
+ public void onVertexStarted(List<TezTaskAttemptID> completions) throws AMUserCodeException {
Map<String, List<Integer>> pluginCompletionsMap = Maps.newHashMap();
if (completions != null && !completions.isEmpty()) {
for (TezTaskAttemptID tezTaskAttemptID : completions) {
@@ -253,23 +258,39 @@ public class VertexManager {
taskIdList.add(taskId);
}
}
- plugin.onVertexStarted(pluginCompletionsMap);
+ try {
+ plugin.onVertexStarted(pluginCompletionsMap);
+ } catch (Exception e) {
+ throw new AMUserCodeException(Source.VertexManager, e);
+ }
}
- public void onSourceTaskCompleted(TezTaskID tezTaskId) {
+ public void onSourceTaskCompleted(TezTaskID tezTaskId) throws AMUserCodeException {
Integer taskId = new Integer(tezTaskId.getId());
String vertexName =
appContext.getCurrentDAG().getVertex(tezTaskId.getVertexID()).getName();
- plugin.onSourceTaskCompleted(vertexName, taskId);
+ try {
+ plugin.onSourceTaskCompleted(vertexName, taskId);
+ } catch (Exception e) {
+ throw new AMUserCodeException(Source.VertexManager, e);
+ }
}
- public void onVertexManagerEventReceived(VertexManagerEvent vmEvent) {
- plugin.onVertexManagerEventReceived(vmEvent);
+ public void onVertexManagerEventReceived(VertexManagerEvent vmEvent) throws AMUserCodeException {
+ try {
+ plugin.onVertexManagerEventReceived(vmEvent);
+ } catch (Exception e) {
+ throw new AMUserCodeException(Source.VertexManager, e);
+ }
}
public List<TezEvent> onRootVertexInitialized(String inputName,
- InputDescriptor inputDescriptor, List<Event> events) {
- plugin.onRootVertexInitialized(inputName, inputDescriptor, events);
+ InputDescriptor inputDescriptor, List<Event> events) throws AMUserCodeException {
+ try {
+ plugin.onRootVertexInitialized(inputName, inputDescriptor, events);
+ } catch (Exception e) {
+ throw new AMUserCodeException(Source.VertexManager, e);
+ }
if (LOG.isDebugEnabled()) {
LOG.debug("vertex:" + managedVertex.getName() + "; after call of VertexManagerPlugin.onRootVertexInitialized"
+ " on input:" + inputName + ", current task events size is " + rootInputInitEventQueue.size());
http://git-wip-us.apache.org/repos/asf/tez/blob/eee882e9/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 e71acb6..2489b1b 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
@@ -23,9 +23,7 @@ import java.nio.ByteBuffer;
import static org.junit.Assert.assertTrue;
import static org.mockito.Matchers.any;
import static org.mockito.Matchers.anyInt;
-import static org.mockito.Mockito.doReturn;
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.*;
import java.io.ByteArrayOutputStream;
import java.io.DataInput;
@@ -75,7 +73,6 @@ import org.apache.tez.dag.api.OutputDescriptor;
import org.apache.tez.dag.api.ProcessorDescriptor;
import org.apache.tez.dag.api.RootInputLeafOutput;
import org.apache.tez.dag.api.TezConfiguration;
-import org.apache.tez.dag.api.TezUncheckedException;
import org.apache.tez.dag.api.UserPayload;
import org.apache.tez.dag.api.VertexLocationHint;
import org.apache.tez.dag.api.EdgeProperty.DataMovementType;
@@ -129,6 +126,7 @@ import org.apache.tez.dag.app.dag.event.VertexEventTaskReschedule;
import org.apache.tez.dag.app.dag.event.VertexEventTermination;
import org.apache.tez.dag.app.dag.event.VertexEventType;
import org.apache.tez.dag.app.dag.impl.DAGImpl.VertexGroupInfo;
+import org.apache.tez.dag.app.dag.impl.TestVertexImpl.VertexManagerWithException.VMExceptionLocation;
import org.apache.tez.dag.app.rm.TaskSchedulerEventHandler;
import org.apache.tez.dag.history.HistoryEventHandler;
import org.apache.tez.dag.library.vertexmanager.InputReadyVertexManager;
@@ -388,6 +386,160 @@ public class TestVertexImpl {
return dag;
}
+ /**
+ * v1 -> v2
+ */
+ private DAGPlan createDAGPlanWithVMException(String initializerClassName, VMExceptionLocation exLocation) {
+ LOG.info("Setting up dag plan with VertexManager which would throw exception");
+ DAGPlan dag = DAGPlan.newBuilder()
+ .setName("initializerWith0Tasks")
+ .addVertex(
+ VertexPlan.newBuilder()
+ .setName("vertex1")
+ .setType(PlanVertexType.NORMAL)
+ .addInputs(
+ RootInputLeafOutputProto.newBuilder()
+ .setControllerDescriptor(
+ TezEntityDescriptorProto.newBuilder().setClassName(
+ initializerClassName))
+ .setName("input1")
+ .setIODescriptor(
+ TezEntityDescriptorProto.newBuilder()
+ .setClassName("InputClazz")
+ .build()
+ ).build()
+ )
+ .setTaskConfig(
+ PlanTaskConfiguration.newBuilder()
+ .setNumTasks(1)
+ .setVirtualCores(4)
+ .setMemoryMb(1024)
+ .setJavaOpts("")
+ .setTaskModule("x1.y1")
+ .build()
+ )
+ .addOutEdgeId("e1")
+ .setVertexManagerPlugin(TezEntityDescriptorProto.newBuilder()
+ .setClassName(VertexManagerWithException.class.getName())
+ .setUserPayload(ByteString.copyFrom(exLocation.name().getBytes())))
+ .build()
+ )
+ .addVertex(
+ VertexPlan.newBuilder()
+ .setName("vertex2")
+ .setType(PlanVertexType.NORMAL)
+ .setTaskConfig(
+ PlanTaskConfiguration.newBuilder()
+ .setNumTasks(1)
+ .setVirtualCores(4)
+ .setMemoryMb(1024)
+ .setJavaOpts("")
+ .setTaskModule("x2.y2")
+ .build()
+ )
+ .addInEdgeId("e1")
+ .setVertexManagerPlugin(TezEntityDescriptorProto.newBuilder()
+ .setClassName(VertexManagerWithException.class.getName())
+ .setUserPayload(ByteString.copyFrom(exLocation.name().getBytes())))
+ .build()
+ )
+ .addEdge(
+ EdgePlan.newBuilder()
+ .setEdgeDestination(TezEntityDescriptorProto.newBuilder().setClassName("v1_v2"))
+ .setInputVertexName("vertex1")
+ .setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o2"))
+ .setOutputVertexName("vertex2")
+ .setDataMovementType(PlanEdgeDataMovementType.BROADCAST)
+ .setId("e1")
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
+ .build()
+ )
+ .build();
+ return dag;
+ }
+
+ /**
+ * v1 -> v2
+ */
+ private DAGPlan createDAGPlanWithIIException() {
+ LOG.info("Setting up dag plan with VertexManager which would throw exception");
+ DAGPlan dag = DAGPlan.newBuilder()
+ .setName("initializerWith0Tasks")
+ .addVertex(
+ VertexPlan.newBuilder()
+ .setName("vertex1")
+ .setType(PlanVertexType.NORMAL)
+ .addInputs(
+ RootInputLeafOutputProto.newBuilder()
+ .setControllerDescriptor(
+ TezEntityDescriptorProto.newBuilder().setClassName(
+ "IrrelevantInitializerClassName"))
+ .setName("input1")
+ .setIODescriptor(
+ TezEntityDescriptorProto.newBuilder()
+ .setClassName("InputClazz")
+ .build()
+ )
+ .build()
+ )
+ .setTaskConfig(
+ PlanTaskConfiguration.newBuilder()
+ .setNumTasks(1)
+ .setVirtualCores(4)
+ .setMemoryMb(1024)
+ .setJavaOpts("")
+ .setTaskModule("x1.y1")
+ .build()
+ )
+ .addOutEdgeId("e1")
+ .build()
+ )
+ .addVertex(
+ VertexPlan.newBuilder()
+ .setName("vertex2")
+ .setType(PlanVertexType.NORMAL)
+ .setTaskConfig(
+ PlanTaskConfiguration.newBuilder()
+ .setNumTasks(1)
+ .setVirtualCores(4)
+ .setMemoryMb(1024)
+ .setJavaOpts("")
+ .setTaskModule("x2.y2")
+ .build()
+ )
+ .addInputs(
+ RootInputLeafOutputProto.newBuilder()
+ .setControllerDescriptor(
+ TezEntityDescriptorProto.newBuilder().setClassName(
+ "IrrelevantInitializerClassName"))
+ .setName("input1")
+ .setIODescriptor(
+ TezEntityDescriptorProto.newBuilder()
+ .setClassName("InputClazz")
+ .build()
+ )
+ .build()
+ )
+ .addInEdgeId("e1")
+ .build()
+ )
+ .addEdge(
+ EdgePlan.newBuilder()
+ .setEdgeDestination(TezEntityDescriptorProto.newBuilder().setClassName("v1_v2"))
+ .setInputVertexName("vertex1")
+ .setEdgeSource(TezEntityDescriptorProto.newBuilder().setClassName("o2"))
+ .setOutputVertexName("vertex2")
+ .setDataMovementType(PlanEdgeDataMovementType.BROADCAST)
+ .setId("e1")
+ .setDataSourceType(PlanEdgeDataSourceType.PERSISTED)
+ .setSchedulingType(PlanEdgeSchedulingType.SEQUENTIAL)
+ .build()
+ )
+ .build();
+ return dag;
+ }
+
private DAGPlan createDAGPlanWithMixedEdges() {
LOG.info("Setting up mixed edge dag plan");
org.apache.tez.dag.api.DAG dag = org.apache.tez.dag.api.DAG.create("MixedEdges");
@@ -1964,6 +2116,7 @@ public class TestVertexImpl {
String dagName = "dag0";
dispatcher = new DrainDispatcher();
appContext = mock(AppContext.class);
+ thh = mock(TaskHeartbeatHandler.class);
historyEventHandler = mock(HistoryEventHandler.class);
TaskSchedulerEventHandler taskScheduler = mock(TaskSchedulerEventHandler.class);
UserGroupInformation ugi;
@@ -3013,7 +3166,7 @@ public class TestVertexImpl {
}
@Test(timeout = 5000)
- public void testVertexWithOneToOneSplitWhileRunning() {
+ public void testVertexWithOneToOneSplitWhileRunning() throws Exception {
int numTasks = 5;
// create a diamond shaped dag with 1-1 edges.
setupPreDagCreation();
@@ -3050,7 +3203,7 @@ public class TestVertexImpl {
}
@Test(timeout = 5000)
- public void testVertexWithOneToOneSplitWhileInited() {
+ public void testVertexWithOneToOneSplitWhileInited() throws Exception {
int numTasks = 5;
// create a diamond shaped dag with 1-1 edges.
setupPreDagCreation();
@@ -3122,11 +3275,16 @@ public class TestVertexImpl {
Assert.assertEquals(VertexState.INITIALIZING, v1.getState());
RootInputInitializerManagerControlled initializerManager1 = v1.getRootInputInitializerManager();
initializerManager1.failInputInitialization();
+ dispatcher.await();
Assert.assertEquals(VertexState.FAILED, v1.getState());
Assert.assertEquals(RootInputVertexManager.class.getName(), v1
.getVertexManager().getPlugin().getClass().getName());
Assert.assertEquals(true, initializerManager1.hasShutDown);
+ Assert.assertTrue(StringUtils.join(v1.getDiagnostics(), ",")
+ .contains(VertexTerminationCause.ROOT_INPUT_INIT_FAILURE.name()));
+ Assert.assertTrue(StringUtils.join(v1.getDiagnostics(), ",")
+ .contains("MockInitializerFailed"));
VertexImplWithControlledInitializerManager v2 = (VertexImplWithControlledInitializerManager) vertices.get("vertex2");
Assert.assertEquals(VertexState.INITIALIZING, v2.getState());
@@ -3137,6 +3295,10 @@ public class TestVertexImpl {
Assert.assertEquals(RootInputVertexManager.class.getName(), v2
.getVertexManager().getPlugin().getClass().getName());
Assert.assertEquals(true, initializerManager2.hasShutDown);
+ Assert.assertTrue(StringUtils.join(v1.getDiagnostics(), ",")
+ .contains(VertexTerminationCause.ROOT_INPUT_INIT_FAILURE.name()));
+ Assert.assertTrue(StringUtils.join(v1.getDiagnostics(), ",")
+ .contains("MockInitializerFailed"));
}
@Test(timeout = 10000)
@@ -4070,7 +4232,7 @@ public class TestVertexImpl {
}
@SuppressWarnings("unchecked")
- @Test(timeout = 5000)
+ @Test(timeout = 500000)
public void testVertexWithInitializerSuccess() {
useCustomInitializer = true;
setupPreDagCreation();
@@ -4614,6 +4776,162 @@ public class TestVertexImpl {
Assert.assertEquals(VertexState.RUNNING, vC.getState());
}
+ @SuppressWarnings("unchecked")
+ @Test(timeout = 5000)
+ public void testExceptionFromVM_Initialize() {
+ useCustomInitializer = true;
+ setupPreDagCreation();
+ dagPlan = createDAGPlanWithVMException("TestInputInitializer", VMExceptionLocation.Initialize);
+ setupPostDagCreation();
+
+ VertexImplWithControlledInitializerManager v1 = (VertexImplWithControlledInitializerManager) vertices
+ .get("vertex1");
+ dispatcher.getEventHandler().handle(new VertexEvent(v1.getVertexId(),
+ VertexEventType.V_INIT));
+ dispatcher.await();
+
+ Assert.assertEquals(VertexState.FAILED, v1.getState());
+ String diagnostics = StringUtils.join(v1.getDiagnostics(), ",");
+ Assert.assertTrue(diagnostics.contains(VMExceptionLocation.Initialize.name()));
+ Assert.assertEquals(VertexTerminationCause.AM_USERCODE_FAILURE, v1.getTerminationCause());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(timeout = 5000)
+ public void testExceptionFromVM_OnRootVertexInitialized() {
+ useCustomInitializer = true;
+ setupPreDagCreation();
+ dagPlan = createDAGPlanWithVMException("TestInputInitializer", VMExceptionLocation.OnRootVertexInitialized);
+ setupPostDagCreation();
+
+ VertexImplWithControlledInitializerManager v1 = (VertexImplWithControlledInitializerManager) vertices
+ .get("vertex1");
+ dispatcher.getEventHandler().handle(new VertexEvent(v1.getVertexId(),
+ VertexEventType.V_INIT));
+ dispatcher.await();
+
+ RootInputInitializerManagerControlled initializerManager1 = v1.getRootInputInitializerManager();
+ initializerManager1.completeInputInitialization();
+ dispatcher.await();
+ Assert.assertEquals(VertexManagerWithException.class, v1.vertexManager.getPlugin().getClass());
+ Assert.assertEquals(VertexState.FAILED, v1.getState());
+ Assert.assertTrue(initializerManager1.hasShutDown);
+ String diagnostics = StringUtils.join(v1.getDiagnostics(), ",");
+ Assert.assertTrue(diagnostics.contains(VMExceptionLocation.OnRootVertexInitialized.name()));
+ Assert.assertEquals(VertexTerminationCause.AM_USERCODE_FAILURE, v1.getTerminationCause());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(timeout = 5000)
+ public void testExceptionFromVM_OnVertexStarted() {
+ useCustomInitializer = true;
+ setupPreDagCreation();
+ dagPlan = createDAGPlanWithVMException("TestInputInitializer", VMExceptionLocation.OnVertexStarted);
+ setupPostDagCreation();
+
+ VertexImplWithControlledInitializerManager v1 = (VertexImplWithControlledInitializerManager) vertices
+ .get("vertex1");
+ dispatcher.getEventHandler().handle(new VertexEvent(v1.getVertexId(),
+ VertexEventType.V_INIT));
+ // wait to ensure init is completed, so that rootInitManager is not null
+ dispatcher.await();
+
+ RootInputInitializerManagerControlled initializerManager1 = v1.getRootInputInitializerManager();
+ initializerManager1.completeInputInitialization();
+ dispatcher.getEventHandler().handle(new VertexEvent(v1.getVertexId(),
+ VertexEventType.V_START));
+ dispatcher.await();
+
+ Assert.assertEquals(VertexManagerWithException.class, v1.vertexManager.getPlugin().getClass());
+ Assert.assertEquals(VertexState.FAILED, v1.getState());
+ String diagnostics = StringUtils.join(v1.getDiagnostics(), ",");
+ Assert.assertTrue(diagnostics.contains(VMExceptionLocation.OnVertexStarted.name()));
+ Assert.assertEquals(VertexTerminationCause.AM_USERCODE_FAILURE, v1.getTerminationCause());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(timeout = 5000)
+ public void testExceptionFromVM_OnSourceTaskCompleted() {
+ useCustomInitializer = true;
+ setupPreDagCreation();
+ dagPlan = createDAGPlanWithVMException("TestInputInitializer", VMExceptionLocation.OnSourceTaskCompleted);
+ setupPostDagCreation();
+
+ VertexImplWithControlledInitializerManager v1 = (VertexImplWithControlledInitializerManager) vertices
+ .get("vertex1");
+ VertexImpl v2 = vertices.get("vertex2");
+
+ dispatcher.getEventHandler().handle(new VertexEvent(v1.getVertexId(),
+ VertexEventType.V_INIT));
+ // wait to ensure V_INIT is handled, so that rootInitManager is not null
+ dispatcher.await();
+ RootInputInitializerManagerControlled initializerManager1 = v1.getRootInputInitializerManager();
+ initializerManager1.completeInputInitialization();
+ dispatcher.getEventHandler().handle(new VertexEvent(v1.getVertexId(),
+ VertexEventType.V_START));
+
+ // ensure v2 go to RUNNING and start one task in v2
+ dispatcher.await();
+
+ TezTaskAttemptID taId = TezTaskAttemptID.getInstance(TezTaskID.getInstance(v1.getVertexId(), 0), 0);
+ v1.getEventHandler().handle(new VertexEventTaskAttemptCompleted(taId,
+ TaskAttemptStateInternal.SUCCEEDED));
+ dispatcher.await();
+
+ Assert.assertEquals(VertexManagerWithException.class, v1.vertexManager.getPlugin().getClass());
+ Assert.assertEquals(VertexManagerWithException.class, v2.vertexManager.getPlugin().getClass());
+ Assert.assertEquals(VertexState.FAILED, v2.getState());
+ String diagnostics = StringUtils.join(v2.getDiagnostics(), ",");
+ Assert.assertTrue(diagnostics.contains(VMExceptionLocation.OnSourceTaskCompleted.name()));
+ Assert.assertEquals(VertexTerminationCause.AM_USERCODE_FAILURE, v2.getTerminationCause());
+ }
+
+ @SuppressWarnings("unchecked")
+ @Test(timeout = 5000)
+ public void testExceptionFromVM_OnVertexManagerEventReceived() {
+ useCustomInitializer = true;
+ setupPreDagCreation();
+ dagPlan = createDAGPlanWithVMException("TestInputInitializer", VMExceptionLocation.OnVertexManagerEventReceived);
+ setupPostDagCreation();
+
+ VertexImplWithControlledInitializerManager v1 = (VertexImplWithControlledInitializerManager) vertices
+ .get("vertex1");
+ dispatcher.getEventHandler().handle(new VertexEvent(v1.getVertexId(),
+ VertexEventType.V_INIT));
+ // wait to ensure init is completed, so that rootInitManager is not null
+ dispatcher.await();
+ RootInputInitializerManagerControlled initializerManager1 = v1.getRootInputInitializerManager();
+ initializerManager1.completeInputInitialization();
+
+ Event vmEvent = VertexManagerEvent.create(v1.getName(), ByteBuffer.wrap(new byte[0]));
+ TezEvent tezEvent = new TezEvent(vmEvent, null);
+ dispatcher.getEventHandler().handle(new VertexEventRouteEvent(v1.getVertexId(), Lists.newArrayList(tezEvent)));
+ dispatcher.await();
+
+ Assert.assertEquals(VertexState.FAILED, v1.getState());
+ String diagnostics = StringUtils.join(v1.getDiagnostics(), ",");
+ Assert.assertTrue(diagnostics.contains(VMExceptionLocation.OnVertexManagerEventReceived.name()));
+ }
+
+ @Test(timeout = 5000)
+ public void testExceptionFromII_Initialize() {
+ useCustomInitializer = true;
+ customInitializer = new EventHandlingRootInputInitializer(null, IIExceptionLocation.Initialize);
+ EventHandlingRootInputInitializer initializer =
+ (EventHandlingRootInputInitializer) customInitializer;
+ setupPreDagCreation();
+ dagPlan = createDAGPlanWithIIException();
+ setupPostDagCreation();
+
+ VertexImplWithRunningInputInitializer v1 =
+ (VertexImplWithRunningInputInitializer) vertices.get("vertex1");
+ initVertex(v1);
+ String diagnostics = StringUtils.join(v1.getDiagnostics(), ",");
+ assertTrue(diagnostics.contains(IIExceptionLocation.Initialize.name()));
+ Assert.assertEquals(VertexTerminationCause.ROOT_INPUT_INIT_FAILURE, v1.getTerminationCause());
+ }
+
+
@InterfaceAudience.Private
public static class RootInputSpecUpdaterVertexManager extends VertexManagerPlugin {
@@ -4698,6 +5016,71 @@ public class TestVertexImpl {
}
@InterfaceAudience.Private
+ public static class VertexManagerWithException extends RootInputVertexManager{
+
+ public static enum VMExceptionLocation {
+ OnRootVertexInitialized,
+ OnSourceTaskCompleted,
+ OnVertexStarted,
+ OnVertexManagerEventReceived,
+ Initialize,
+ }
+
+ private VMExceptionLocation exLocation;
+
+ public VertexManagerWithException(VertexManagerPluginContext context) {
+ super(context);
+ }
+
+ @Override
+ public void initialize() {
+ super.initialize();
+ this.exLocation = VMExceptionLocation.valueOf(
+ new String(getContext().getUserPayload().deepCopyAsArray()));
+ if (this.exLocation == VMExceptionLocation.Initialize) {
+ throw new RuntimeException(this.exLocation.name());
+ }
+ }
+
+ @Override
+ public void onRootVertexInitialized(String inputName,
+ InputDescriptor inputDescriptor, List<Event> events) {
+ if (this.exLocation == VMExceptionLocation.OnRootVertexInitialized) {
+ throw new RuntimeException(this.exLocation.name());
+ }
+ super.onRootVertexInitialized(inputName, inputDescriptor, events);
+ }
+
+ @Override
+ public void onSourceTaskCompleted(String srcVertexName, Integer attemptId) {
+ if (this.exLocation == VMExceptionLocation.OnSourceTaskCompleted) {
+ throw new RuntimeException(this.exLocation.name());
+ }
+ super.onSourceTaskCompleted(srcVertexName, attemptId);
+ }
+
+ @Override
+ public void onVertexStarted(Map<String, List<Integer>> completions) {
+ if (this.exLocation == VMExceptionLocation.OnVertexStarted) {
+ throw new RuntimeException(this.exLocation.name());
+ }
+ super.onVertexStarted(completions);
+ }
+
+ @Override
+ public void onVertexManagerEventReceived(VertexManagerEvent vmEvent) {
+ super.onVertexManagerEventReceived(vmEvent);
+ if (this.exLocation == VMExceptionLocation.OnVertexManagerEventReceived) {
+ throw new RuntimeException(this.exLocation.name());
+ }
+ }
+ }
+
+ public static enum IIExceptionLocation {
+ Initialize,
+ }
+
+ @InterfaceAudience.Private
public static class EventHandlingRootInputInitializer extends InputInitializer
implements ContextSettableInputInitialzier {
@@ -4712,14 +5095,24 @@ public class TestVertexImpl {
private final List<InputInitializerEvent> initializerEvents = new LinkedList<InputInitializerEvent>();
private volatile InputInitializerContext context;
private volatile int numExpectedEvents = 1;
+ private IIExceptionLocation exLocation = null;
public EventHandlingRootInputInitializer(
InputInitializerContext initializerContext) {
super(initializerContext);
}
+ public EventHandlingRootInputInitializer(
+ InputInitializerContext initializerContext, IIExceptionLocation exLocation) {
+ super(initializerContext);
+ this.exLocation = exLocation;
+ }
+
@Override
public List<Event> initialize() throws Exception {
+ if (exLocation == IIExceptionLocation.Initialize) {
+ throw new Exception(exLocation.name());
+ }
context.registerForVertexStateUpdates("vertex1", null);
initStarted.set(true);
lock.lock();
http://git-wip-us.apache.org/repos/asf/tez/blob/eee882e9/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexManager.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexManager.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexManager.java
index b1825a5..ccd344b 100644
--- a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexManager.java
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexManager.java
@@ -50,7 +50,7 @@ import com.google.common.collect.Sets;
public class TestVertexManager {
@Test
- public void testOnRootVertexInitialized() {
+ public void testOnRootVertexInitialized() throws Exception {
Vertex mockVertex = mock(Vertex.class, RETURNS_DEEP_STUBS);
AppContext mockAppContext = mock(AppContext.class, RETURNS_DEEP_STUBS);
doReturn("vertex1").when(mockVertex).getName();
@@ -87,9 +87,10 @@ public class TestVertexManager {
/**
* TEZ-1647
* custom vertex manager generates events only when both i1 and i2 are initialized.
+ * @throws Exception
*/
@Test
- public void testOnRootVertexInitialized2() {
+ public void testOnRootVertexInitialized2() throws Exception {
Vertex mockVertex = mock(Vertex.class, RETURNS_DEEP_STUBS);
AppContext mockAppContext = mock(AppContext.class, RETURNS_DEEP_STUBS);
doReturn("vertex1").when(mockVertex).getName();