tez-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From bi...@apache.org
Subject [2/2] git commit: TEZ-1547. Make use of state change notifier in VertexManagerPlugins and fix TEZ-1494 without latency penalty (bikas) (cherry picked from commit cd0ed751ae502492733738f2d7d7d0ae8e4224e9)
Date Tue, 04 Nov 2014 21:09:14 GMT
TEZ-1547. Make use of state change notifier in VertexManagerPlugins and fix TEZ-1494 without latency penalty (bikas)
(cherry picked from commit cd0ed751ae502492733738f2d7d7d0ae8e4224e9)


Project: http://git-wip-us.apache.org/repos/asf/tez/repo
Commit: http://git-wip-us.apache.org/repos/asf/tez/commit/bf8d7a5e
Tree: http://git-wip-us.apache.org/repos/asf/tez/tree/bf8d7a5e
Diff: http://git-wip-us.apache.org/repos/asf/tez/diff/bf8d7a5e

Branch: refs/heads/branch-0.5
Commit: bf8d7a5e25b506985f7e0b358b63d4fb725b2972
Parents: 6e6b55d
Author: Bikas Saha <bikas@apache.org>
Authored: Tue Nov 4 13:07:06 2014 -0800
Committer: Bikas Saha <bikas@apache.org>
Committed: Tue Nov 4 13:08:20 2014 -0800

----------------------------------------------------------------------
 CHANGES.txt                                     |   2 +
 .../apache/tez/dag/api/VertexManagerPlugin.java |  21 ++
 .../tez/dag/api/VertexManagerPluginContext.java |  50 ++++
 .../apache/tez/dag/api/event/VertexState.java   |  17 +-
 .../tez/runtime/api/InputInitializer.java       |   4 +-
 .../java/org/apache/tez/dag/app/dag/Vertex.java |   5 +-
 .../event/VertexEventManagerUserCodeError.java  |  36 +++
 .../tez/dag/app/dag/event/VertexEventType.java  |   1 +
 .../dag/impl/ImmediateStartVertexManager.java   |  93 ++++---
 .../apache/tez/dag/app/dag/impl/VertexImpl.java | 180 ++++++++++--
 .../tez/dag/app/dag/impl/VertexManager.java     | 150 ++++++++--
 .../tez/dag/app/dag/impl/TestDAGImpl.java       |   7 -
 .../impl/TestImmediateStartVertexManager.java   | 131 +++++++++
 .../tez/dag/app/dag/impl/TestVertexImpl.java    | 279 ++++++++++++++-----
 .../tez/dag/app/dag/impl/TestVertexManager.java |   5 +-
 .../dag/app/dag/impl/TestVertexRecovery.java    |  48 ++++
 .../tez/examples/SortMergeJoinExample.java      |  16 +-
 .../vertexmanager/ShuffleVertexManager.java     |  60 +++-
 .../vertexmanager/TestShuffleVertexManager.java | 122 +++++---
 .../org/apache/tez/test/TestAMRecovery.java     |   5 +-
 20 files changed, 1005 insertions(+), 227 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/tez/blob/bf8d7a5e/CHANGES.txt
----------------------------------------------------------------------
diff --git a/CHANGES.txt b/CHANGES.txt
index 882aa07..ee3f358 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -94,6 +94,8 @@ ALL CHANGES:
   TEZ-1731. OnDiskMerger can end up clobbering files across tasks with LocalDiskFetch enabled.
   TEZ-1735. Allow setting basic info per DAG for Tez UI.
   TEZ-1728. Remove local host name from Fetcher thread name.
+  TEZ-1547. Make use of state change notifier in VertexManagerPlugins and fix
+  TEZ-1494 without latency penalty
 
 Release 0.5.1: 2014-10-02
 

http://git-wip-us.apache.org/repos/asf/tez/blob/bf8d7a5e/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 b494b12..6aa18d6 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
@@ -23,6 +23,7 @@ import java.util.Map;
 
 import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
+import org.apache.tez.dag.api.event.VertexStateUpdate;
 import org.apache.tez.runtime.api.Event;
 import org.apache.tez.runtime.api.events.VertexManagerEvent;
 
@@ -99,4 +100,24 @@ public abstract class VertexManagerPlugin {
   public final VertexManagerPluginContext getContext() {
     return this.context;
   }
+  
+  /**
+   * Receive notifications on vertex state changes.
+   * <p/>
+   * State changes will be received based on the registration via
+   * {@link VertexManagerPluginContext#registerForVertexStateUpdates(String, java.util.Set)}
+   * . Notifications will be received for all registered state changes, and not
+   * just for the latest state update. They will be in order in which the state
+   * change occurred.
+   * </p><br>This method may be invoked concurrently with {@link #onVertexStarted(Map)} etc. and 
+   * multi-threading/concurrency implications must be considered.
+   *  
+   * @param stateUpdate
+   *          an event indicating the name of the vertex, and it's updated
+   *          state. Additional information may be available for specific
+   *          events, Look at the type hierarchy for
+   *          {@link org.apache.tez.dag.api.event.VertexStateUpdate}
+   */
+  public void onVertexStateUpdated(VertexStateUpdate stateUpdate) throws Exception {
+  }  
  }

http://git-wip-us.apache.org/repos/asf/tez/blob/bf8d7a5e/tez-api/src/main/java/org/apache/tez/dag/api/VertexManagerPluginContext.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/VertexManagerPluginContext.java b/tez-api/src/main/java/org/apache/tez/dag/api/VertexManagerPluginContext.java
index c1f4bcd..dfa9287 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/VertexManagerPluginContext.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/VertexManagerPluginContext.java
@@ -30,6 +30,7 @@ import org.apache.hadoop.classification.InterfaceStability.Unstable;
 import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.tez.dag.api.TaskLocationHint;
+import org.apache.tez.dag.api.event.VertexState;
 import org.apache.tez.runtime.api.InputSpecUpdate;
 import org.apache.tez.runtime.api.events.InputDataInformationEvent;
 
@@ -169,4 +170,53 @@ public interface VertexManagerPluginContext {
    * @return DAG Attempt number
    */
   public int getDAGAttemptNumber();
+  
+  /**
+   * Register to get notifications on updates to the specified vertex. Notifications will be sent
+   * via {@link VertexManagerPlugin#onVertexStateUpdated(org.apache.tez.dag.api.event.VertexStateUpdate)}
+   *
+   * This method can only be invoked once. Duplicate invocations will result in an error.
+   *
+   * @param vertexName the vertex name for which notifications are required.
+   * @param stateSet   the set of states for which notifications are required. null implies all
+   */
+  void registerForVertexStateUpdates(String vertexName, @Nullable Set<VertexState> stateSet);
+  
+  /**
+   * Optional API. No need to call this when the vertex is not fully defined to
+   * start with. E.g. vertex parallelism is not defined, or edges are not
+   * configured. In that case, Tez will assume that the vertex needs
+   * reconfiguration. If the vertex is already fully defined, but the
+   * {@link VertexManagerPlugin} wants to reconfigure the vertex, then it must
+   * use this API to inform Tez about its intention. Without invoking this
+   * method, it is invalid to re-configure the vertex, e.g. via the
+   * {@link #setVertexParallelism(int, VertexLocationHint, Map, Map)} method if
+   * the vertex is already fully defined. This can be invoked at any time until
+   * {@link VertexManagerPlugin#initialize()} has completed. Its invalid to
+   * invoke this method after {@link VertexManagerPlugin#initialize()} has
+   * completed<br>
+   * If this API is invoked, then {@link #doneReconfiguringVertex()} must be
+   * invoked after the {@link VertexManagerPlugin} is done reconfiguring the
+   * vertex, . Actions like scheduling tasks or sending events do not count as
+   * reconfiguration.
+   */
+  public void vertexReconfigurationPlanned();
+  
+  /**
+   * Optional API. This needs to be called only if {@link #vertexReconfigurationPlanned()} has been 
+   * invoked. This must be called after {@link #vertexReconfigurationPlanned()} is called.
+   */
+  public void doneReconfiguringVertex();
+  
+  /**
+   * Optional API. This API can be invoked to declare that the
+   * {@link VertexManagerPlugin} is done with its work. After this the system
+   * will not invoke the plugin methods any more. Its invalid for the plugin to
+   * make further invocations of the context APIs after this. This can be used
+   * to stop receiving further {@link VertexState} notifications after the
+   * plugin has made all changes.
+   */
+  // TODO must be done later after TEZ-1714
+  //public void vertexManagerDone();
+
 }
\ No newline at end of file

http://git-wip-us.apache.org/repos/asf/tez/blob/bf8d7a5e/tez-api/src/main/java/org/apache/tez/dag/api/event/VertexState.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/dag/api/event/VertexState.java b/tez-api/src/main/java/org/apache/tez/dag/api/event/VertexState.java
index ab296a5..c9c2d58 100644
--- a/tez-api/src/main/java/org/apache/tez/dag/api/event/VertexState.java
+++ b/tez-api/src/main/java/org/apache/tez/dag/api/event/VertexState.java
@@ -28,12 +28,14 @@ import org.apache.hadoop.classification.InterfaceStability;
 @InterfaceStability.Unstable
 public enum VertexState {
   /**
-   * Indicates that the Vertex had entered the SUCCEEDED state. A vertex could go back into RUNNING state after SUCCEEDING
+   * Indicates that the Vertex had entered the SUCCEEDED state. A vertex could
+   * go back into RUNNING state after SUCCEEDING
    */
   SUCCEEDED,
   /**
-   * Indicates that the Vertex had entered the RUNNING state. This state can be reached after SUCCEEDED, if some
-   * tasks belonging to the vertex are restarted due to errors
+   * Indicates that the Vertex had entered the RUNNING state. This state can be
+   * reached after SUCCEEDED, if some tasks belonging to the vertex are
+   * restarted due to errors
    */
   RUNNING,
   /**
@@ -47,5 +49,12 @@ public enum VertexState {
   /**
    * Indicates that the parallelism for the vertex had changed.
    */
-  PARALLELISM_UPDATED
+  PARALLELISM_UPDATED,
+  /**
+   * Indicates that the vertex has been completely configured. Parallelism, edges, edge
+   * properties, inputs/outputs have been set and will not be changed any
+   * further. Listeners can depend on the vertex's configured state after
+   * receiving this notification.
+   */
+  CONFIGURED
 }

http://git-wip-us.apache.org/repos/asf/tez/blob/bf8d7a5e/tez-api/src/main/java/org/apache/tez/runtime/api/InputInitializer.java
----------------------------------------------------------------------
diff --git a/tez-api/src/main/java/org/apache/tez/runtime/api/InputInitializer.java b/tez-api/src/main/java/org/apache/tez/runtime/api/InputInitializer.java
index d9d6517..cc33205 100644
--- a/tez-api/src/main/java/org/apache/tez/runtime/api/InputInitializer.java
+++ b/tez-api/src/main/java/org/apache/tez/runtime/api/InputInitializer.java
@@ -19,6 +19,7 @@
 package org.apache.tez.runtime.api;
 
 import java.util.List;
+import java.util.Map;
 
 import org.apache.hadoop.classification.InterfaceStability.Unstable;
 import org.apache.hadoop.classification.InterfaceAudience.Public;
@@ -97,7 +98,8 @@ public abstract class InputInitializer {
    *
    * Extensive processing should not be performed via this method call. Instead this should just be
    * used as a notification mechanism to the main initialization, which is via the initialize method.
-   *
+   * <br>This method may be invoked concurrently with {@link #initialize()} etc. and 
+   * multi-threading/concurrency implications must be considered.
    * @param stateUpdate an event indicating the name of the vertex, and it's updated state.
    *                    Additional information may be available for specific events, Look at the
    *                    type hierarchy for {@link org.apache.tez.dag.api.event.VertexStateUpdate}

http://git-wip-us.apache.org/repos/asf/tez/blob/bf8d7a5e/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 fa1f2c4..cfedc41 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
@@ -89,8 +89,11 @@ public interface Vertex extends Comparable<Vertex> {
 
   void setParallelism(int parallelism, VertexLocationHint vertexLocationHint,
       Map<String, EdgeManagerPluginDescriptor> sourceEdgeManagers,
-      Map<String, InputSpecUpdate> rootInputSpecUpdate) throws AMUserCodeException;
+      Map<String, InputSpecUpdate> rootInputSpecUpdate, boolean fromVertexManager)
+      throws AMUserCodeException;
   void setVertexLocationHint(VertexLocationHint vertexLocationHint);
+  void vertexReconfigurationPlanned();
+  void doneReconfiguringVertex();
 
   // CHANGE THESE TO LISTS AND MAINTAIN ORDER?
   void setInputVertices(Map<Vertex, Edge> inVertices);

http://git-wip-us.apache.org/repos/asf/tez/blob/bf8d7a5e/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/VertexEventManagerUserCodeError.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/VertexEventManagerUserCodeError.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/VertexEventManagerUserCodeError.java
new file mode 100644
index 0000000..022620a
--- /dev/null
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/VertexEventManagerUserCodeError.java
@@ -0,0 +1,36 @@
+/**
+* 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.event;
+
+import org.apache.tez.dag.app.dag.impl.AMUserCodeException;
+import org.apache.tez.dag.records.TezVertexID;
+
+public class VertexEventManagerUserCodeError extends VertexEvent {
+  final AMUserCodeException e;
+  
+  public VertexEventManagerUserCodeError(TezVertexID vertexId, AMUserCodeException e) {
+    super(vertexId, VertexEventType.V_MANAGER_USER_CODE_ERROR);
+    this.e = e;
+  }
+  
+  public AMUserCodeException getError() {
+    return e;
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/tez/blob/bf8d7a5e/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/VertexEventType.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/VertexEventType.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/VertexEventType.java
index e649095..b4f7e29 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/VertexEventType.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/VertexEventType.java
@@ -42,6 +42,7 @@ public enum VertexEventType {
   
   //Producer:Any component
   V_INTERNAL_ERROR,
+  V_MANAGER_USER_CODE_ERROR,
   
   V_ROUTE_EVENT,
   V_ONE_TO_ONE_SOURCE_SPLIT,

http://git-wip-us.apache.org/repos/asf/tez/blob/bf8d7a5e/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ImmediateStartVertexManager.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ImmediateStartVertexManager.java b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ImmediateStartVertexManager.java
index 773426b..00b5306 100644
--- a/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ImmediateStartVertexManager.java
+++ b/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ImmediateStartVertexManager.java
@@ -18,8 +18,10 @@
 
 package org.apache.tez.dag.app.dag.impl;
 
+import com.google.common.base.Preconditions;
 import com.google.common.collect.Lists;
 import com.google.common.collect.Maps;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.tez.dag.api.EdgeProperty;
@@ -27,11 +29,15 @@ import org.apache.tez.dag.api.InputDescriptor;
 import org.apache.tez.dag.api.VertexManagerPlugin;
 import org.apache.tez.dag.api.VertexManagerPluginContext;
 import org.apache.tez.dag.api.VertexManagerPluginContext.TaskWithLocationHint;
+import org.apache.tez.dag.api.event.VertexState;
+import org.apache.tez.dag.api.event.VertexStateUpdate;
 import org.apache.tez.runtime.api.Event;
 import org.apache.tez.runtime.api.events.VertexManagerEvent;
 
+import java.util.EnumSet;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 /**
  * Starts all tasks immediately on vertex start
@@ -40,18 +46,10 @@ public class ImmediateStartVertexManager extends VertexManagerPlugin {
 
   private static final Log LOG = LogFactory.getLog(ImmediateStartVertexManager.class);
 
-  private final Map<String, SourceVertexInfo> srcVertexInfo = Maps.newHashMap();
+  private final Map<String, Boolean> srcVertexConfigured = Maps.newConcurrentMap();
   private int managedTasks;
   private boolean tasksScheduled = false;
-
-  class SourceVertexInfo {
-    EdgeProperty edgeProperty;
-    int numFinishedTasks;
-
-    SourceVertexInfo(EdgeProperty edgeProperty) {
-      this.edgeProperty = edgeProperty;
-    }
-  }
+  private AtomicBoolean onVertexStartedDone = new AtomicBoolean(false);
 
   public ImmediateStartVertexManager(VertexManagerPluginContext context) {
     super(context);
@@ -63,37 +61,35 @@ public class ImmediateStartVertexManager extends VertexManagerPlugin {
     Map<String, EdgeProperty> edges = getContext().getInputVertexEdgeProperties();
     for (Map.Entry<String, EdgeProperty> entry : edges.entrySet()) {
       String srcVertex = entry.getKey();
-      EdgeProperty edgeProp = entry.getValue();
-      LOG.info("Task count in " + srcVertex + ": " + getContext().getVertexNumTasks(srcVertex));
       //track vertices with task count > 0
       if (getContext().getVertexNumTasks(srcVertex) > 0) {
-        srcVertexInfo.put(srcVertex, new SourceVertexInfo(edgeProp));
+        LOG.info("Task count in " + srcVertex + ": " + getContext().getVertexNumTasks(srcVertex));
+        srcVertexConfigured.put(srcVertex, false);
+        getContext().registerForVertexStateUpdates(srcVertex, EnumSet.of(VertexState.CONFIGURED));
       } else {
         LOG.info("Vertex: " + getContext().getVertexName() + "; Ignoring " + srcVertex
             + " as it has got 0 tasks");
       }
     }
-
-    //handle completions
-    for (Map.Entry<String, List<Integer>> entry : completions.entrySet()) {
-      for (Integer task : entry.getValue()) {
-        handleSourceTaskFinished(entry.getKey(), task);
-      }
-    }
+    onVertexStartedDone.set(true);
     scheduleTasks();
   }
 
-  private void handleSourceTaskFinished(String vertex, Integer taskId) {
-    SourceVertexInfo srcInfo = srcVertexInfo.get(vertex);
-    //Not mandatory to check for duplicate completions here
-    srcInfo.numFinishedTasks++;
-  }
-
   private void scheduleTasks() {
-    if (!canScheduleTasks()) {
+    if (!onVertexStartedDone.get()) {
+      // vertex not started yet
+      return;
+    }
+    if (tasksScheduled) {
+      // already scheduled
       return;
     }
 
+    if (!canScheduleTasks()) {
+      return;
+    }
+    
+    tasksScheduled = true;
     List<TaskWithLocationHint> tasksToStart = Lists.newArrayListWithCapacity(managedTasks);
     for (int i = 0; i < managedTasks; ++i) {
       tasksToStart.add(new TaskWithLocationHint(new Integer(i), null));
@@ -103,35 +99,42 @@ public class ImmediateStartVertexManager extends VertexManagerPlugin {
       LOG.info("Starting " + tasksToStart.size() + " in " + getContext().getVertexName());
       getContext().scheduleVertexTasks(tasksToStart);
     }
-    tasksScheduled = true;
+    // all tasks scheduled. Can call vertexManagerDone().
+    // TODO TEZ-1714 for locking issues getContext().vertexManagerDone();
   }
 
   private boolean canScheduleTasks() {
-    //Check if at least 1 task is finished from each source vertex (in case of broadcast &
-    // one-to-one or custom)
-    for (Map.Entry<String, SourceVertexInfo> entry : srcVertexInfo.entrySet()) {
-      SourceVertexInfo srcVertexInfo = entry.getValue();
-      switch(srcVertexInfo.edgeProperty.getDataMovementType()) {
-      case ONE_TO_ONE:
-      case BROADCAST:
-      case CUSTOM:
-        if (srcVertexInfo.numFinishedTasks == 0) {
-          //do not schedule tasks until a task from source task is complete
-          return false;
+    // check for source vertices completely configured
+    for (Map.Entry<String, Boolean> entry : srcVertexConfigured.entrySet()) {
+      if (!entry.getValue().booleanValue()) {
+        // vertex not configured
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Waiting for vertex: " + entry.getKey() + " in vertex: " + getContext().getVertexName());
         }
-      default:
-        break;
+        return false;
       }
     }
+
     return true;
   }
+  
+  @Override
+  public void onVertexStateUpdated(VertexStateUpdate stateUpdate) {
+    Preconditions.checkArgument(stateUpdate.getVertexState() == VertexState.CONFIGURED,
+        "Received incorrect state notification : " + stateUpdate.getVertexState() + " for vertex: "
+            + stateUpdate.getVertexName() + " in vertex: " + getContext().getVertexName());
+    Preconditions.checkArgument(srcVertexConfigured.containsKey(stateUpdate.getVertexName()),
+        "Received incorrect vertex notification : " + stateUpdate.getVertexState() + " for vertex: "
+            + stateUpdate.getVertexName() + " in vertex: " + getContext().getVertexName());
+    Preconditions.checkState(srcVertexConfigured.put(stateUpdate.getVertexName(), true)
+        .booleanValue() == false);
+    LOG.info("Received configured notification: " + stateUpdate.getVertexState() + " for vertex: "
+        + stateUpdate.getVertexName() + " in vertex: " + getContext().getVertexName());
+    scheduleTasks();
+  }
 
   @Override
   public void onSourceTaskCompleted(String srcVertexName, Integer attemptId) {
-    handleSourceTaskFinished(srcVertexName, attemptId);
-    if (!tasksScheduled) {
-      scheduleTasks();
-    }
   }
 
   @Override

http://git-wip-us.apache.org/repos/asf/tez/blob/bf8d7a5e/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 4a88949..593ecca 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
@@ -110,6 +110,7 @@ import org.apache.tez.dag.app.dag.event.TaskEventRecoverTask;
 import org.apache.tez.dag.app.dag.event.TaskEventTermination;
 import org.apache.tez.dag.app.dag.event.TaskEventType;
 import org.apache.tez.dag.app.dag.event.VertexEvent;
+import org.apache.tez.dag.app.dag.event.VertexEventManagerUserCodeError;
 import org.apache.tez.dag.app.dag.event.VertexEventNullEdgeInitialized;
 import org.apache.tez.dag.app.dag.event.VertexEventOneToOneSourceSplit;
 import org.apache.tez.dag.app.dag.event.VertexEventRecoverVertex;
@@ -320,7 +321,11 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
               (VertexState.RECOVERING, VertexState.RECOVERING,
                   VertexEventType.V_TERMINATE,
                   new TerminateDuringRecoverTransition())
-
+          .addTransition
+              (VertexState.RECOVERING, EnumSet.of(VertexState.RECOVERING),
+                  VertexEventType.V_MANAGER_USER_CODE_ERROR,
+                  new VertexManagerUserCodeErrorTransition())
+          
           // Transitions from INITIALIZING state
           .addTransition(VertexState.INITIALIZING,
               EnumSet.of(VertexState.INITIALIZING, VertexState.INITED,
@@ -353,6 +358,9 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
               EnumSet.of(VertexState.INITIALIZING, VertexState.FAILED),
               VertexEventType.V_ROUTE_EVENT,
               ROUTE_EVENT_TRANSITION)
+          .addTransition(VertexState.INITIALIZING, EnumSet.of(VertexState.FAILED),
+              VertexEventType.V_MANAGER_USER_CODE_ERROR,
+              new VertexManagerUserCodeErrorTransition())
           .addTransition(VertexState.INITIALIZING, VertexState.KILLED,
               VertexEventType.V_TERMINATE,
               new TerminateInitingVertexTransition())
@@ -399,6 +407,9 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
           .addTransition(VertexState.INITED, VertexState.KILLED,
               VertexEventType.V_TERMINATE,
               new TerminateInitedVertexTransition())
+          .addTransition(VertexState.INITED, EnumSet.of(VertexState.FAILED),
+              VertexEventType.V_MANAGER_USER_CODE_ERROR,
+              new VertexManagerUserCodeErrorTransition())
           .addTransition(VertexState.INITED, VertexState.ERROR,
               VertexEventType.V_INTERNAL_ERROR,
               INTERNAL_ERROR_TRANSITION)
@@ -429,6 +440,9 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
           .addTransition(VertexState.RUNNING, VertexState.TERMINATING,
               VertexEventType.V_TERMINATE,
               new VertexKilledTransition())
+          .addTransition(VertexState.RUNNING, EnumSet.of(VertexState.TERMINATING),
+              VertexEventType.V_MANAGER_USER_CODE_ERROR,
+              new VertexManagerUserCodeErrorTransition())
           .addTransition(VertexState.RUNNING, VertexState.RUNNING,
               VertexEventType.V_TASK_RESCHEDULED,
               new TaskRescheduledTransition())
@@ -460,6 +474,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
           // Ignore-able events
           .addTransition(VertexState.TERMINATING, VertexState.TERMINATING,
               EnumSet.of(VertexEventType.V_TERMINATE,
+                  VertexEventType.V_MANAGER_USER_CODE_ERROR,
                   VertexEventType.V_ROOT_INPUT_FAILED,
                   VertexEventType.V_SOURCE_VERTEX_STARTED,
                   VertexEventType.V_ROOT_INPUT_INITIALIZED,
@@ -512,6 +527,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
           // Ignore-able events
           .addTransition(VertexState.FAILED, VertexState.FAILED,
               EnumSet.of(VertexEventType.V_TERMINATE,
+                  VertexEventType.V_MANAGER_USER_CODE_ERROR,
                   VertexEventType.V_ROOT_INPUT_FAILED,
                   VertexEventType.V_SOURCE_VERTEX_STARTED,
                   VertexEventType.V_TASK_RESCHEDULED,
@@ -534,6 +550,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
           // Ignore-able events
           .addTransition(VertexState.KILLED, VertexState.KILLED,
               EnumSet.of(VertexEventType.V_TERMINATE,
+                  VertexEventType.V_MANAGER_USER_CODE_ERROR,
                   VertexEventType.V_ROOT_INPUT_FAILED,
                   VertexEventType.V_INIT,
                   VertexEventType.V_SOURCE_VERTEX_STARTED,
@@ -559,6 +576,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
                   VertexEventType.V_START,
                   VertexEventType.V_ROUTE_EVENT,
                   VertexEventType.V_TERMINATE,
+                  VertexEventType.V_MANAGER_USER_CODE_ERROR,
                   VertexEventType.V_TASK_COMPLETED,
                   VertexEventType.V_TASK_ATTEMPT_COMPLETED,
                   VertexEventType.V_ONE_TO_ONE_SOURCE_SPLIT,
@@ -627,6 +645,10 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
 
   private final String vertexName;
   private final ProcessorDescriptor processorDescriptor;
+  
+  private boolean vertexToBeReconfiguredByManager = false;
+  AtomicBoolean vmIsInitialized = new AtomicBoolean(false);
+  AtomicBoolean completelyConfiguredSent = new AtomicBoolean(false);
 
   @VisibleForTesting
   Map<Vertex, Edge> sourceVertices;
@@ -1212,14 +1234,16 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
   @Override
   public void setParallelism(int parallelism, VertexLocationHint vertexLocationHint,
       Map<String, EdgeManagerPluginDescriptor> sourceEdgeManagers,
-      Map<String, InputSpecUpdate> rootInputSpecUpdates) throws AMUserCodeException {
-    setParallelism(parallelism, vertexLocationHint, sourceEdgeManagers, rootInputSpecUpdates, false);
+      Map<String, InputSpecUpdate> rootInputSpecUpdates, boolean fromVertexManager)
+      throws AMUserCodeException {
+    setParallelism(parallelism, vertexLocationHint, sourceEdgeManagers, rootInputSpecUpdates,
+        false, fromVertexManager);
   }
 
   private void setParallelism(int parallelism, VertexLocationHint vertexLocationHint,
       Map<String, EdgeManagerPluginDescriptor> sourceEdgeManagers,
       Map<String, InputSpecUpdate> rootInputSpecUpdates,
-      boolean recovering) throws AMUserCodeException {
+      boolean recovering, boolean fromVertexManager) throws AMUserCodeException {
     if (recovering) {
       writeLock.lock();
       try {
@@ -1255,6 +1279,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
     + parallelism + " for vertex: " + logIdentifier);
     setVertexLocationHint(vertexLocationHint);
     writeLock.lock();
+    
     try {
       if (parallelismSet == true) {
         String msg = "Parallelism can only be set dynamically once per vertex: " + logIdentifier; 
@@ -1262,6 +1287,15 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
         throw new TezUncheckedException(msg);
       }
 
+      if (fromVertexManager && canInitVertex()) {
+        // vertex is fully defined. setParallelism has been called. VertexManager should have 
+        // informed us about this. Otherwise we would have notified listeners that we are fully 
+        // defined before we are actually fully defined
+        Preconditions.checkState(vertexToBeReconfiguredByManager, "Vertex is fully configured but still"
+            + " the reconfiguration API has been called. VertexManager must notify the framework using " 
+            + " context.vertexReconfigurationPlanned() before re-configuring the vertex.");
+      }
+      
       parallelismSet = true;
 
       // Input initializer/Vertex Manager/1-1 split expected to set parallelism.
@@ -1421,6 +1455,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
     }
   }
 
+  @Override
   public void setVertexLocationHint(VertexLocationHint vertexLocationHint) {
     writeLock.lock();
     try {
@@ -1432,6 +1467,46 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
       writeLock.unlock();
     }
   }
+  
+  @Override
+  public void vertexReconfigurationPlanned() {
+    vertexReconfigurationPlanned(false);
+  }
+  
+  public void vertexReconfigurationPlanned(boolean testOverride) {
+    writeLock.lock();
+    try {
+      if (testOverride) {
+        Preconditions.checkState(vmIsInitialized.get() && completelyConfiguredSent.get(),
+            "test should override only failed cases");
+      } else {
+        Preconditions.checkState(!vmIsInitialized.get(),
+            "context.vertexReconfigurationPlanned() cannot be called after initialize()");
+        Preconditions.checkState(!completelyConfiguredSent.get(), "vertexReconfigurationPlanned() "
+            + " cannot be invoked after the vertex has been configured.");
+      }
+      this.vertexToBeReconfiguredByManager = true;
+    } finally {
+      writeLock.unlock();
+    }      
+  }
+
+  @Override
+  public void doneReconfiguringVertex() {
+    writeLock.lock();
+    try {
+      Preconditions.checkState(vertexToBeReconfiguredByManager, "doneReconfiguringVertex() can be "
+          + "invoked only after vertexReconfigurationPlanned() is invoked");
+      this.vertexToBeReconfiguredByManager = false;
+      if (completelyConfiguredSent.compareAndSet(false, true)) {
+        // vertex already started and at that time this event was not sent. Send now.
+        stateChangeNotifier.stateChanged(vertexId, new VertexStateUpdate(vertexName,
+            org.apache.tez.dag.api.event.VertexState.CONFIGURED));
+      }
+    } finally {
+      writeLock.unlock();
+    }    
+  }
 
   @Override
   /**
@@ -1970,6 +2045,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
 
     try {
       vertexManager.initialize();
+      vmIsInitialized.set(true);
     } catch (AMUserCodeException e) {
       String msg = "Exception in " + e.getSource()+ ", vertex:" + logIdentifier;
       LOG.error(msg, e);
@@ -2042,7 +2118,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
               .getVertexManagerPlugin());
       LOG.info("Setting user vertex manager plugin: "
           + pluginDesc.getClassName() + " on vertex: " + getName());
-      vertexManager = new VertexManager(pluginDesc, this, appContext);
+      vertexManager = new VertexManager(pluginDesc, this, appContext, stateChangeNotifier);
     } else {
       // Intended order of picking a vertex manager
       // If there is an InputInitializer then we use the RootInputVertexManager. May be fixed by TEZ-703
@@ -2055,26 +2131,26 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
             + logIdentifier);
         vertexManager = new VertexManager(
             VertexManagerPluginDescriptor.create(RootInputVertexManager.class.getName()),
-            this, appContext);
+            this, appContext, stateChangeNotifier);
       } else if (hasOneToOne && !hasCustom) {
         LOG.info("Setting vertexManager to InputReadyVertexManager for "
             + logIdentifier);
         vertexManager = new VertexManager(
             VertexManagerPluginDescriptor.create(InputReadyVertexManager.class.getName()),
-            this, appContext);
+            this, appContext, stateChangeNotifier);
       } else if (hasBipartite && !hasCustom) {
         LOG.info("Setting vertexManager to ShuffleVertexManager for "
             + logIdentifier);
         // shuffle vertex manager needs a conf payload
         vertexManager = new VertexManager(ShuffleVertexManager.createConfigBuilder(conf).build(),
-            this, appContext);
+            this, appContext, stateChangeNotifier);
       } else {
         // schedule all tasks upon vertex start. Default behavior.
         LOG.info("Setting vertexManager to ImmediateStartVertexManager for "
             + logIdentifier);
         vertexManager = new VertexManager(
             VertexManagerPluginDescriptor.create(ImmediateStartVertexManager.class.getName()),
-            this, appContext);
+            this, appContext, stateChangeNotifier);
       }
     }
   }
@@ -2222,7 +2298,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
                   new TaskEventRecoverTask(task.getTaskId()));
             }
             try {
-              vertex.vertexManager.onVertexStarted(vertex.pendingReportedSrcCompletions);
+              vertex.recoveryCodeSimulatingStart();
               endState = VertexState.RUNNING;
             } catch (AMUserCodeException e) {
               String msg = "Exception in " + e.getSource() + ", vertex:" + vertex.getLogIdentifier();
@@ -2274,7 +2350,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
                         taskState));
               }
               try {
-                vertex.vertexManager.onVertexStarted(vertex.pendingReportedSrcCompletions);
+                vertex.recoveryCodeSimulatingStart();
                 endState = VertexState.RUNNING;
               } catch (AMUserCodeException e) {
                 String msg = "Exception in " + e.getSource() +", vertex:" + vertex.getLogIdentifier();
@@ -2335,6 +2411,13 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
     }
 
   }
+  
+  private void recoveryCodeSimulatingStart() throws AMUserCodeException {
+    vertexManager.onVertexStarted(pendingReportedSrcCompletions);
+    // This code is duplicated from startVertex() because recovery does not follow normal
+    // transitions. To be removed after recovery code is fixed.
+    maybeSendConfiguredEvent();
+  }
 
   private void routeRecoveredEvents(VertexState vertexState,
       List<TezEvent> tezEvents) {
@@ -2556,7 +2639,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
           boolean successSetParallelism ;
           try {
             vertex.setParallelism(0,
-              null, vertex.recoveredSourceEdgeManagers, vertex.recoveredRootInputSpecUpdates, true);
+              null, vertex.recoveredSourceEdgeManagers, vertex.recoveredRootInputSpecUpdates, true, false);
             successSetParallelism = true;
           } catch (Exception e) {
             successSetParallelism = false;
@@ -2614,7 +2697,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
           }
           try {
             vertex.setParallelism(0, null, vertex.recoveredSourceEdgeManagers,
-              vertex.recoveredRootInputSpecUpdates, true);
+              vertex.recoveredRootInputSpecUpdates, true, false);
             successSetParallelism = true;
           } catch (Exception e) {
             successSetParallelism = false;
@@ -2634,7 +2717,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
                   new TaskEventRecoverTask(task.getTaskId()));
             }
             try {
-              vertex.vertexManager.onVertexStarted(vertex.pendingReportedSrcCompletions);
+              vertex.recoveryCodeSimulatingStart();
               endState = VertexState.RUNNING;
             } catch (AMUserCodeException e) {
               String msg = "Exception in " + e.getSource() + ", vertex=" + vertex.getLogIdentifier();
@@ -2674,7 +2757,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
             }
             // Wait for all tasks to recover and report back
             try {
-              vertex.vertexManager.onVertexStarted(vertex.pendingReportedSrcCompletions);
+              vertex.recoveryCodeSimulatingStart();
               endState = VertexState.RUNNING;
             } catch (AMUserCodeException e) {
               String msg = "Exception in " + e.getSource() +", vertex:" + vertex.getLogIdentifier();
@@ -2825,10 +2908,12 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
 
       // Create tasks based on initial configuration, but don't start them yet.
       if (vertex.numTasks == -1) {
+        // this block must always return VertexState.INITIALIZING
         LOG.info("Num tasks is -1. Expecting VertexManager/InputInitializers/1-1 split"
             + " to set #tasks for the vertex " + vertex.getVertexId());
 
         if (vertex.inputsWithInitializers != null) {
+          LOG.info("Vertex will initialize from input initializer. " + vertex.logIdentifier);
           vertex.setupInputInitializerManager();
           return VertexState.INITIALIZING;
         } else {
@@ -2857,8 +2942,9 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
       } else {
         LOG.info("Creating " + vertex.numTasks + " for vertex: " + vertex.logIdentifier);
         vertex.createTasks();
-
+        // this block may return VertexState.INITIALIZING
         if (vertex.inputsWithInitializers != null) {
+          LOG.info("Vertex will initialize from input initializer. " + vertex.logIdentifier);
           vertex.setupInputInitializerManager();
           return VertexState.INITIALIZING;
         }
@@ -2867,6 +2953,8 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
           return VertexState.INITIALIZING;
         }
         LOG.info("Directly initializing vertex: " + vertex.logIdentifier);
+        // vertex is completely configured. Send out notification now.
+        vertex.maybeSendConfiguredEvent();
         boolean isInitialized = vertex.initializeVertex();
         if (isInitialized) {
           return VertexState.INITED;
@@ -3020,7 +3108,7 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
           " numTasks " + splitEvent.getNumTasks());
       vertex.originalOneToOneSplitSource = originalSplitSource;
       try {
-        vertex.setParallelism(splitEvent.getNumTasks(), null, null, null);
+        vertex.setParallelism(splitEvent.getNumTasks(), null, null, null, false);
       } catch (Exception e) {
         // ingore this exception, should not happen
         LOG.error("Unexpected exception, Just set Parallelims to a specified value, not involve EdgeManager,"
@@ -3130,8 +3218,25 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
       return vertex.startVertex();
     }
   }
+  
+  private void maybeSendConfiguredEvent() {
+    // the vertex is fully configured by the time it starts. Always notify completely configured
+    // unless the vertex manager has told us that it is going to reconfigure it further
+    Preconditions.checkState(canInitVertex());
+    if (!this.vertexToBeReconfiguredByManager) {
+      // this vertex will not be reconfigured by its manager
+      if (completelyConfiguredSent.compareAndSet(false, true)) {
+        stateChangeNotifier.stateChanged(vertexId, new VertexStateUpdate(vertexName,
+            org.apache.tez.dag.api.event.VertexState.CONFIGURED));
+      }
+    }    
+  }
 
   private VertexState startVertex() {
+    // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
+    // IMPORTANT - Until Recovery is fixed to use normal state transitions, if any code is added 
+    // here then please check if it needs to be duplicated in recoveryCodeSimulatingStart().
+    // !!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!
     Preconditions.checkState(getState() == VertexState.INITED,
         "Vertex must be inited " + logIdentifier);
 
@@ -3147,7 +3252,16 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
     }
     pendingReportedSrcCompletions.clear();
     logJobHistoryVertexStartedEvent();
-
+    
+    // the vertex is fully configured by the time it starts. Always notify completely configured
+    // unless the vertex manager has told us that it is going to reconfigure it further.
+    // If the vertex was pre-configured then the event would have been sent out earlier. Calling again 
+    // would be a no-op. If the vertex was not fully configured and waiting for that to complete then
+    // we would start immediately after that. Either parallelism updated (now) or IPO changed (future) 
+    // or vertex added (future). Simplify these cases by sending the event now automatically for the 
+    // user as if they had invoked the planned()/done() API's.
+    maybeSendConfiguredEvent();
+    
     // TODO: Metrics
     //job.metrics.runningJob(job);
 
@@ -3309,6 +3423,36 @@ public class VertexImpl implements org.apache.tez.dag.app.dag.Vertex,
     }
   }
 
+  private static class VertexManagerUserCodeErrorTransition implements
+    MultipleArcTransition<VertexImpl, VertexEvent, VertexState> {
+    @Override
+    public VertexState transition(VertexImpl vertex, VertexEvent event) {
+      VertexEventManagerUserCodeError errEvent = ((VertexEventManagerUserCodeError) event);
+      AMUserCodeException e = errEvent.getError();
+      String msg = "Exception in " + e.getSource() + ", vertex:" + vertex.getLogIdentifier();
+      LOG.error(msg, e);
+      
+      if (vertex.getState() == VertexState.RECOVERING) {
+        LOG.info("Received a user code error during recovering, setting recovered"
+            + " state to FAILED");
+        vertex.addDiagnostic(msg + "," + ExceptionUtils.getStackTrace(e.getCause()));
+        vertex.terminationCause = VertexTerminationCause.AM_USERCODE_FAILURE;
+        vertex.recoveredState = VertexState.FAILED;
+        return VertexState.RECOVERING;
+      } else 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;
+      }
+    }
+  }
+  
   /**
    * Here, the Vertex is being told that one of it's source task-attempts
    * completed.

http://git-wip-us.apache.org/repos/asf/tez/blob/bf8d7a5e/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 1bfb0f9..dd38c2a 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
@@ -27,6 +27,7 @@ import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.annotation.Nullable;
 
@@ -46,11 +47,16 @@ import org.apache.tez.dag.api.VertexLocationHint;
 import org.apache.tez.dag.api.VertexManagerPlugin;
 import org.apache.tez.dag.api.VertexManagerPluginContext;
 import org.apache.tez.dag.api.VertexManagerPluginDescriptor;
+import org.apache.tez.dag.api.event.VertexState;
+import org.apache.tez.dag.api.event.VertexStateUpdate;
 import org.apache.tez.dag.app.AppContext;
+import org.apache.tez.dag.app.dag.StateChangeNotifier;
 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.event.VertexEventManagerUserCodeError;
 import org.apache.tez.dag.app.dag.impl.AMUserCodeException.Source;
+import org.apache.tez.dag.app.dag.VertexStateUpdateListener;
 import org.apache.tez.dag.records.TezTaskAttemptID;
 import org.apache.tez.dag.records.TezTaskID;
 import org.apache.tez.runtime.api.Event;
@@ -75,18 +81,27 @@ public class VertexManager {
   UserPayload payload = null;
   AppContext appContext;
   BlockingQueue<TezEvent> rootInputInitEventQueue;
+  StateChangeNotifier stateChangeNotifier;
 
   private static final Log LOG = LogFactory.getLog(VertexManager.class);
 
-  class VertexManagerPluginContextImpl implements VertexManagerPluginContext {
-    // TODO Add functionality to allow VertexManagers to send VertexManagerEvents
+  class VertexManagerPluginContextImpl implements VertexManagerPluginContext, VertexStateUpdateListener {
 
     private EventMetaData rootEventSourceMetadata = new EventMetaData(EventProducerConsumerType.INPUT,
         managedVertex.getName(), "NULL_VERTEX", null);
     private Map<String, EventMetaData> destinationEventMetadataMap = Maps.newHashMap();
+    private final List<String> notificationRegisteredVertices = Lists.newArrayList();
+    AtomicBoolean isComplete = new AtomicBoolean(false);
 
+    private void checkAndThrowIfDone() {
+      if (isComplete()) {
+        throw new TezUncheckedException("Cannot invoke context methods after reporting done");
+      }
+    }
+    
     @Override
-    public Map<String, EdgeProperty> getInputVertexEdgeProperties() {
+    public synchronized Map<String, EdgeProperty> getInputVertexEdgeProperties() {
+      checkAndThrowIfDone();
       // TODO Something similar for Initial Inputs - payload etc visible
       Map<Vertex, Edge> inputs = managedVertex.getInputVertices();
       Map<String, EdgeProperty> vertexEdgeMap =
@@ -98,22 +113,25 @@ public class VertexManager {
     }
 
     @Override
-    public String getVertexName() {
+    public synchronized String getVertexName() {
+      checkAndThrowIfDone();
       return managedVertex.getName();
     }
 
     @Override
-    public int getVertexNumTasks(String vertexName) {
+    public synchronized int getVertexNumTasks(String vertexName) {
+      checkAndThrowIfDone();
       return appContext.getCurrentDAG().getVertex(vertexName).getTotalTasks();
     }
 
     @Override
-    public void setVertexParallelism(int parallelism, VertexLocationHint vertexLocationHint,
+    public synchronized void setVertexParallelism(int parallelism, VertexLocationHint vertexLocationHint,
         Map<String, EdgeManagerPluginDescriptor> sourceEdgeManagers,
         Map<String, InputSpecUpdate> rootInputSpecUpdate) {
+      checkAndThrowIfDone();
       try {
         managedVertex.setParallelism(parallelism, vertexLocationHint, sourceEdgeManagers,
-            rootInputSpecUpdate);
+            rootInputSpecUpdate, true);
       } catch (AMUserCodeException e) {
         // workaround: convert it to TezUncheckedException which would be caught in VM
         throw new TezUncheckedException(e);
@@ -121,13 +139,15 @@ public class VertexManager {
     }
 
     @Override
-    public void scheduleVertexTasks(List<TaskWithLocationHint> tasks) {
+    public synchronized void scheduleVertexTasks(List<TaskWithLocationHint> tasks) {
+      checkAndThrowIfDone();
       managedVertex.scheduleTasks(tasks);
     }
 
     @Nullable
     @Override
-    public Set<String> getVertexInputNames() {
+    public synchronized Set<String> getVertexInputNames() {
+      checkAndThrowIfDone();
       Set<String> inputNames = null;
       Map<String, RootInputLeafOutput<InputDescriptor, InputInitializerDescriptor>>
           inputs = managedVertex.getAdditionalInputs();
@@ -138,13 +158,15 @@ public class VertexManager {
     }
 
     @Override
-    public UserPayload getUserPayload() {
+    public synchronized UserPayload getUserPayload() {
+      checkAndThrowIfDone();
       return payload;
     }
 
     @Override
-    public void addRootInputEvents(final String inputName,
+    public synchronized void addRootInputEvents(final String inputName,
         Collection<InputDataInformationEvent> events) {
+      checkAndThrowIfDone();
       verifyIsRootInput(inputName);
       Collection<TezEvent> tezEvents = Collections2.transform(events,
           new Function<InputDataInformationEvent, TezEvent>() {
@@ -166,13 +188,15 @@ public class VertexManager {
 
 
     @Override
-    public void setVertexLocationHint(VertexLocationHint locationHint) {
+    public synchronized void setVertexLocationHint(VertexLocationHint locationHint) {
+      checkAndThrowIfDone();
       Preconditions.checkNotNull(locationHint, "locationHint is null");
       managedVertex.setVertexLocationHint(locationHint);
     }
 
     @Override
-    public int getDAGAttemptNumber() {
+    public synchronized int getDAGAttemptNumber() {
+      checkAndThrowIfDone();
       return appContext.getApplicationAttemptId().getAttemptId();
     }
 
@@ -192,22 +216,26 @@ public class VertexManager {
     }
 
     @Override
-    public Resource getVertexTaskResource() {
+    public synchronized Resource getVertexTaskResource() {
+      checkAndThrowIfDone();
       return managedVertex.getTaskResource();
     }
 
     @Override
-    public Resource getTotalAvailableResource() {
+    public synchronized Resource getTotalAvailableResource() {
+      checkAndThrowIfDone();
       return appContext.getTaskScheduler().getTotalResources();
     }
 
     @Override
-    public int getNumClusterNodes() {
+    public synchronized int getNumClusterNodes() {
+      checkAndThrowIfDone();
       return appContext.getTaskScheduler().getNumClusterNodes();
     }
 
     @Override
-    public Container getTaskContainer(String vertexName, Integer taskIndex) {
+    public synchronized Container getTaskContainer(String vertexName, Integer taskIndex) {
+      checkAndThrowIfDone();
       Vertex vertex = appContext.getCurrentDAG().getVertex(vertexName);
       Task task = vertex.getTask(taskIndex.intValue());
       TaskAttempt attempt = task.getSuccessfulAttempt();
@@ -216,16 +244,82 @@ public class VertexManager {
       }
       return null;
     }
+
+    @Override
+    public synchronized void registerForVertexStateUpdates(String vertexName, Set<VertexState> stateSet) {
+      checkAndThrowIfDone();
+      synchronized(notificationRegisteredVertices) {
+        notificationRegisteredVertices.add(vertexName);
+      }
+      stateChangeNotifier.registerForVertexUpdates(vertexName, stateSet, this);
+    }
+
+    private void unregisterForVertexStateUpdates() {
+      synchronized (notificationRegisteredVertices) {
+        for (String vertexName : notificationRegisteredVertices) {
+          stateChangeNotifier.unregisterForVertexUpdates(vertexName, this);
+        }
+
+      }
+    }
+    
+    boolean isComplete() {
+      return (isComplete.get() == true);
+    }
+
+    // TODO add later after TEZ-1714 @Override
+    public synchronized void vertexManagerDone() {
+      checkAndThrowIfDone();
+      LOG.info("Vertex Manager reported done for : " + managedVertex.getLogIdentifier());
+      this.isComplete.set(true);
+      unregisterForVertexStateUpdates();
+    }
+
+    @Override
+    public synchronized void vertexReconfigurationPlanned() {
+      checkAndThrowIfDone();
+      managedVertex.vertexReconfigurationPlanned();
+    }
+
+    @Override
+    public synchronized void doneReconfiguringVertex() {
+      checkAndThrowIfDone();
+      managedVertex.doneReconfiguringVertex();
+    }
+
+    @SuppressWarnings("unchecked")
+    @Override
+    public synchronized void onStateUpdated(VertexStateUpdate event) {
+      if (isComplete()) {
+        if (LOG.isDebugEnabled()) {
+          LOG.debug("Dropping state update for vertex=" + event.getVertexName() + ", state=" +
+              event.getVertexState() +
+              " since vertexmanager for " + managedVertex.getLogIdentifier() + " is complete.");
+        }
+      } else {
+        try {
+          plugin.onVertexStateUpdated(event);
+        } catch (Exception e) {
+          // state change must be triggered via an event transition
+          appContext.getEventHandler().handle(
+              new VertexEventManagerUserCodeError(managedVertex.getVertexId(),
+                  new AMUserCodeException(Source.VertexManager, e)));
+        }
+      }
+    }
+
   }
 
   public VertexManager(VertexManagerPluginDescriptor pluginDesc,
-      Vertex managedVertex, AppContext appContext) {
+      Vertex managedVertex, AppContext appContext, StateChangeNotifier stateChangeNotifier) {
     checkNotNull(pluginDesc, "pluginDesc is null");
     checkNotNull(managedVertex, "managedVertex is null");
     checkNotNull(appContext, "appContext is null");
+    checkNotNull(stateChangeNotifier, "notifier is null");
     this.pluginDesc = pluginDesc;
     this.managedVertex = managedVertex;
     this.appContext = appContext;
+    this.stateChangeNotifier = stateChangeNotifier;
     // don't specify the size of rootInputInitEventQueue, otherwise it will fail when addAll
     this.rootInputInitEventQueue = new LinkedBlockingQueue<TezEvent>();
   }
@@ -242,7 +336,9 @@ public class VertexManager {
       payload = pluginDesc.getUserPayload();
     }
     try {
-      plugin.initialize(); 
+      if (!pluginContext.isComplete()) {
+        plugin.initialize();
+      }
     } catch (Exception e) {
       throw new AMUserCodeException(Source.VertexManager, e);
     }
@@ -265,7 +361,9 @@ public class VertexManager {
       }
     }
     try {
-      plugin.onVertexStarted(pluginCompletionsMap);
+      if (!pluginContext.isComplete()) {
+        plugin.onVertexStarted(pluginCompletionsMap);
+      }
     } catch (Exception e) {
       throw new AMUserCodeException(Source.VertexManager, e);
     }
@@ -276,7 +374,9 @@ public class VertexManager {
     String vertexName =
         appContext.getCurrentDAG().getVertex(tezTaskId.getVertexID()).getName();
     try {
-      plugin.onSourceTaskCompleted(vertexName, taskId);
+      if (!pluginContext.isComplete()) {
+        plugin.onSourceTaskCompleted(vertexName, taskId);
+      }
     } catch (Exception e) {
       throw new AMUserCodeException(Source.VertexManager, e);
     }
@@ -284,7 +384,9 @@ public class VertexManager {
 
   public void onVertexManagerEventReceived(VertexManagerEvent vmEvent) throws AMUserCodeException {
     try {
-      plugin.onVertexManagerEventReceived(vmEvent);
+      if (!pluginContext.isComplete()) {
+        plugin.onVertexManagerEventReceived(vmEvent);
+      }
     } catch (Exception e) {
       throw new AMUserCodeException(Source.VertexManager, e);
     }
@@ -293,7 +395,9 @@ public class VertexManager {
   public List<TezEvent> onRootVertexInitialized(String inputName,
       InputDescriptor inputDescriptor, List<Event> events) throws AMUserCodeException {
     try {
-      plugin.onRootVertexInitialized(inputName, inputDescriptor, events);
+      if (!pluginContext.isComplete()) {
+        plugin.onRootVertexInitialized(inputName, inputDescriptor, events);
+      }
     } catch (Exception e) {
       throw new AMUserCodeException(Source.VertexManager, e);
     }

http://git-wip-us.apache.org/repos/asf/tez/blob/bf8d7a5e/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 f1961aa..d859ae0 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
@@ -898,13 +898,10 @@ public class TestDAGImpl {
     dispatcher.getEventHandler().handle(new DAGEventStartDag(dagWithCustomEdge.getID(),
         null));
     dispatcher.await();
-    Assert.assertEquals(DAGState.RUNNING, dagWithCustomEdge.getState());
 
     VertexImpl v2 = (VertexImpl)dagWithCustomEdge.getVertex("vertex2");
     LOG.info(v2.getTasks().size());
     Task t1= v2.getTask(0);
-    dispatcher.getEventHandler().handle(new TaskEvent(t1.getTaskId(), TaskEventType.T_SCHEDULE));
-    dispatcher.await();
     TaskAttemptImpl ta1= (TaskAttemptImpl)t1.getAttempt(TezTaskAttemptID.getInstance(t1.getTaskId(), 0));
 
     Assert.assertEquals(TaskAttemptStateInternal.FAILED, ta1.getInternalState());
@@ -947,7 +944,6 @@ public class TestDAGImpl {
 
     VertexImpl v1 = (VertexImpl)dagWithCustomEdge.getVertex("vertex1");
     VertexImpl v2 = (VertexImpl)dagWithCustomEdge.getVertex("vertex2");
-    v2.scheduleTasks(Collections.singletonList(new TaskWithLocationHint(new Integer(0), null)));
     dispatcher.await();
     Task t1= v2.getTask(0);
     TaskAttemptImpl ta1= (TaskAttemptImpl)t1.getAttempt(TezTaskAttemptID.getInstance(t1.getTaskId(), 0));
@@ -977,7 +973,6 @@ public class TestDAGImpl {
 
     VertexImpl v1 = (VertexImpl)dagWithCustomEdge.getVertex("vertex1");
     VertexImpl v2 = (VertexImpl)dagWithCustomEdge.getVertex("vertex2");
-    v2.scheduleTasks(Collections.singletonList(new TaskWithLocationHint(new Integer(0), null)));
     dispatcher.await();
 
     Task t1= v2.getTask(0);
@@ -1007,7 +1002,6 @@ public class TestDAGImpl {
 
     VertexImpl v1 = (VertexImpl)dagWithCustomEdge.getVertex("vertex1");
     VertexImpl v2 = (VertexImpl)dagWithCustomEdge.getVertex("vertex2");
-    v2.scheduleTasks(Collections.singletonList(new TaskWithLocationHint(new Integer(0), null)));
     dispatcher.await();
 
     Task t1= v2.getTask(0);
@@ -1038,7 +1032,6 @@ public class TestDAGImpl {
 
     VertexImpl v1 = (VertexImpl)dagWithCustomEdge.getVertex("vertex1");
     VertexImpl v2 = (VertexImpl)dagWithCustomEdge.getVertex("vertex2");
-    v2.scheduleTasks(Collections.singletonList(new TaskWithLocationHint(new Integer(0), null)));
     dispatcher.await();
 
     Task t1= v2.getTask(0);

http://git-wip-us.apache.org/repos/asf/tez/blob/bf8d7a5e/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestImmediateStartVertexManager.java
----------------------------------------------------------------------
diff --git a/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestImmediateStartVertexManager.java b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestImmediateStartVertexManager.java
new file mode 100644
index 0000000..6d071a7
--- /dev/null
+++ b/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestImmediateStartVertexManager.java
@@ -0,0 +1,131 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.tez.dag.app.dag.impl;
+
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+
+import org.apache.tez.dag.api.EdgeManagerPluginDescriptor;
+import org.apache.tez.dag.api.EdgeProperty;
+import org.apache.tez.dag.api.InputDescriptor;
+import org.apache.tez.dag.api.OutputDescriptor;
+import org.apache.tez.dag.api.VertexManagerPluginContext;
+import org.apache.tez.dag.api.EdgeProperty.SchedulingType;
+import org.apache.tez.dag.api.VertexManagerPluginContext.TaskWithLocationHint;
+import org.apache.tez.dag.api.event.VertexState;
+import org.apache.tez.dag.api.event.VertexStateUpdate;
+import org.junit.Assert;
+import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import static org.mockito.Matchers.anyList;
+import static org.mockito.Mockito.anySet;
+import static org.mockito.Mockito.anyString;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+public class TestImmediateStartVertexManager {
+  
+  @SuppressWarnings({ "unchecked", "rawtypes" })
+  @Test (timeout=5000)
+  public void testBasic() {
+    HashMap<String, EdgeProperty> mockInputVertices = 
+        new HashMap<String, EdgeProperty>();
+    final String mockSrcVertexId1 = "Vertex1";
+    EdgeProperty eProp1 = EdgeProperty.create(
+        EdgeProperty.DataMovementType.SCATTER_GATHER,
+        EdgeProperty.DataSourceType.PERSISTED,
+        SchedulingType.SEQUENTIAL,
+        OutputDescriptor.create("out"),
+        InputDescriptor.create("in"));
+    final String mockSrcVertexId2 = "Vertex2";
+    EdgeProperty eProp2 = EdgeProperty.create(mock(EdgeManagerPluginDescriptor.class),
+        EdgeProperty.DataSourceType.PERSISTED,
+        SchedulingType.SEQUENTIAL,
+        OutputDescriptor.create("out"),
+        InputDescriptor.create("in"));
+    final String mockSrcVertexId3 = "Vertex3";
+    EdgeProperty eProp3 = EdgeProperty.create(
+        EdgeProperty.DataMovementType.BROADCAST,
+        EdgeProperty.DataSourceType.PERSISTED,
+        SchedulingType.SEQUENTIAL,
+        OutputDescriptor.create("out"),
+        InputDescriptor.create("in"));
+    
+    final String mockManagedVertexId = "Vertex4";
+    
+    mockInputVertices.put(mockSrcVertexId1, eProp1);
+    mockInputVertices.put(mockSrcVertexId2, eProp2);
+    mockInputVertices.put(mockSrcVertexId3, eProp3);
+
+    final VertexManagerPluginContext mockContext = mock(VertexManagerPluginContext.class);
+    when(mockContext.getInputVertexEdgeProperties()).thenReturn(mockInputVertices);
+    when(mockContext.getVertexName()).thenReturn(mockManagedVertexId);
+    when(mockContext.getVertexNumTasks(mockManagedVertexId)).thenReturn(4);
+    when(mockContext.getVertexNumTasks(mockSrcVertexId1)).thenReturn(2);
+    when(mockContext.getVertexNumTasks(mockSrcVertexId2)).thenReturn(2);
+    when(mockContext.getVertexNumTasks(mockSrcVertexId3)).thenReturn(2);
+    
+    final HashSet<Integer> scheduledTasks = new HashSet<Integer>();
+    doAnswer(new Answer() {
+      public Object answer(InvocationOnMock invocation) {
+          Object[] args = invocation.getArguments();
+          scheduledTasks.clear();
+          List<TaskWithLocationHint> tasks = (List<TaskWithLocationHint>)args[0];
+          for (TaskWithLocationHint task : tasks) {
+            scheduledTasks.add(task.getTaskIndex());
+          }
+          return null;
+      }}).when(mockContext).scheduleVertexTasks(anyList());
+    
+    ImmediateStartVertexManager manager = new ImmediateStartVertexManager(mockContext);
+    manager.initialize();
+    manager.onVertexStarted(null);
+    verify(mockContext, times(0)).scheduleVertexTasks(anyList());
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId1,
+        VertexState.CONFIGURED));
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId2,
+        VertexState.CONFIGURED));
+    manager.onVertexStateUpdated(new VertexStateUpdate(mockSrcVertexId3,
+        VertexState.CONFIGURED));
+    verify(mockContext, times(1)).scheduleVertexTasks(anyList());
+    Assert.assertEquals(4, scheduledTasks.size());
+
+    // simulate race between onVertexStarted and notifications
+    scheduledTasks.clear();
+    final ImmediateStartVertexManager raceManager = new ImmediateStartVertexManager(mockContext);
+    doAnswer(new Answer() {
+      public Object answer(InvocationOnMock invocation) throws Exception {
+        raceManager.onVertexStateUpdated(new VertexStateUpdate((String)invocation.getArguments()[0],
+            VertexState.CONFIGURED));
+        scheduledTasks.clear();
+        return null;
+    }}).when(mockContext).registerForVertexStateUpdates(anyString(), anySet());
+    raceManager.initialize();
+    raceManager.onVertexStarted(null);
+    verify(mockContext, times(2)).scheduleVertexTasks(anyList());
+    Assert.assertEquals(4, scheduledTasks.size());
+  }
+  
+}


Mime
View raw message