VertexImpl Lifecycle

VertexImpl is the AM-side representation of a single vertex in a running DAG. It is the largest and most intricate state machine in Tez — on current master it registers around 61 transitions across ten states — because a vertex sits at the crossroads of input initialization (deciding how many tasks to run), task scheduling (via a pluggable VertexManagerPlugin), edge routing (moving data-movement events between producer and consumer tasks), and output commit (finalizing a DataSink on the AM). This chapter walks the happy path NEW → INITIALIZING → INITED → RUNNING → COMMITTING → SUCCEEDED, the failure/kill machinery through TERMINATING, and the two features that make Tez vertices special: deferred parallelism and AM-side commit.

Read state-machines.md first — this chapter assumes you can read a StateMachineFactory chain and know why VertexImpl wraps its machine in StateMachineTez.

After this chapter you can: name every VertexState and predict the next state for any event, explain how a vertex with parallelism = -1 decides its task count, describe the commit protocol and the config that toggles it, and read the TERMINATING drain logic that turns "something failed" into a clean terminal state.

The file and the states

tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java

The internal state enum is small and exact — quote it, don't paraphrase:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/VertexState.java
public enum VertexState {
  NEW,
  INITIALIZING,
  INITED,
  RUNNING,
  SUCCEEDED,
  FAILED,
  KILLED,
  ERROR,
  TERMINATING,
  COMMITTING,
}

Ten states. Note there is no separate RECOVERING state — recovery is handled by replaying history events through the normal machine (the V_RECOVER event and RecoverTransition), landing the vertex in whatever state its recorded history implies. Meanings:

grep -n "stateMachineFactory\|installTopology" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
grep -c "addTransition" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java

The happy path, transition by transition

The factory opens with the V_INIT arc out of NEW, and it is a multiple-arc transition declaring five possible landing states:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
= new StateMachineFactory<VertexImpl, VertexState, VertexEventType, VertexEvent>
          (VertexState.NEW)
      // Transitions from NEW state
      .addTransition
          (VertexState.NEW,
              EnumSet.of(VertexState.NEW, VertexState.INITED,
                  VertexState.INITIALIZING, VertexState.FAILED, VertexState.KILLED),
              VertexEventType.V_INIT,
              /* ... InitTransition ... */)

Why five? Because V_INIT does very different things depending on the vertex:

• NEW — the vertex has source vertices that are not all initialized yet; it waits (numInitedSourceVertices hasn't reached the source count).
• INITIALIZING — the vertex has root-input initializers or a VertexManagerPlugin that needs to run before the task count is known.
• INITED — the task count was already fixed statically, so init completes synchronously.
• FAILED/KILLED — recovery: the recorded history says this vertex already finished in a prior AM attempt.

The InitTransition hook shows the source-vertex gating and the cascade to downstream vertices:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
public static class InitTransition implements
    MultipleArcTransition<VertexImpl, VertexEvent, VertexState> {
  @Override
  public VertexState transition(VertexImpl vertex, VertexEvent event) {
    // recover from recovery data (NEW->FAILED/KILLED)
    if (vertex.recoveryData != null && !vertex.recoveryData.isVertexInited()
        && vertex.recoveryData.isVertexFinished()) {
      VertexFinishedEvent finishedEvent = vertex.recoveryData.getVertexFinishedEvent();
      vertex.diagnostics.add(finishedEvent.getDiagnostics());
      return vertex.finished(finishedEvent.getState());
    }
    VertexState vertexState = VertexState.NEW;
    vertex.numInitedSourceVertices++;
    if (vertex.sourceVertices == null || vertex.sourceVertices.isEmpty() ||
        (vertex.numInitedSourceVertices == vertex.sourceVertices.size())) {
      vertexState = handleInitEvent(vertex);
      if (vertexState != VertexState.FAILED) {
        if (vertex.targetVertices != null && !vertex.targetVertices.isEmpty()) {
          for (Vertex target : vertex.targetVertices.keySet()) {
            vertex.getEventHandler().handle(new VertexEvent(target.getVertexId(),
                VertexEventType.V_INIT));
          }
        }
      }
    }
    return vertexState;
  }

That last loop is how V_INIT propagates down the DAG: initializing one vertex sends V_INIT to each of its targets. Initialization ripples from roots to leaves.

A simplified happy-path map (the authoritative picture is the visualize Graphviz output from state-machines.md):

stateDiagram-v2
    [*] --> NEW
    NEW --> INITIALIZING: V_INIT (initializers / VM)
    NEW --> INITED: V_INIT (static parallelism)
    INITIALIZING --> INITED: V_READY_TO_INIT
    INITIALIZING --> TERMINATING: V_ROOT_INPUT_FAILED
    INITED --> RUNNING: V_START
    RUNNING --> COMMITTING: all tasks SUCCEEDED, has committer
    RUNNING --> SUCCEEDED: all tasks SUCCEEDED, no committer
    RUNNING --> TERMINATING: V_TERMINATE / task failed beyond budget
    COMMITTING --> SUCCEEDED: V_COMMIT_COMPLETED (all)
    COMMITTING --> TERMINATING: V_TERMINATE / commit failed
    TERMINATING --> FAILED
    TERMINATING --> KILLED
    SUCCEEDED --> [*]
    FAILED --> [*]
    KILLED --> [*]

Initialization: deferred parallelism

The signature Tez feature at the vertex level is that a vertex can start life not knowing how many tasks it will run. numTasks == -1 means "decide later." The InitTransition hook (handleInitEvent → setupVertex) branches on this:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java (handleInitEvent)
// 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.getLogIdentifier());
  if (vertex.hasInputInitializers()) {
    // ... run RootInputInitializerManager, stay in INITIALIZING
  }
}

The three ways parallelism gets fixed:

1. Root input initializers. A VertexInputInitializer (e.g. Hive's split generator) runs on the AM, computes input splits, and reports back via V_ROOT_INPUT_INITIALIZED. The RootInputInitializedTransition routes those events into the VertexManagerPlugin:

   // tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
   public static class RootInputInitializedTransition implements
       MultipleArcTransition<VertexImpl, VertexEvent, VertexState> {
     @Override
     public VertexState transition(VertexImpl vertex, VertexEvent event) {
       VertexEventRootInputInitialized liInitEvent = (VertexEventRootInputInitialized) event;
       VertexState state = vertex.getState();
       if (state == VertexState.INITIALIZING) {
         vertex.vertexManager.onRootVertexInitialized(liInitEvent.getInputName(), /* ... */
             liInitEvent.getEvents());
       }
       vertex.numInitializedInputs++;
       if (vertex.numInitializedInputs == vertex.inputsWithInitializers.size()) {
         vertex.rootInputInitializerManager.shutdown();
         vertex.rootInputInitializerManager = null;
       }
       return vertex.getState();
     }
   }
   

2. VertexManagerPlugin.reconfigureVertex(...). The plugin (which owns scheduling policy) can call back into the vertex to set parallelism and edge properties. VertexManager.VertexManagerPluginContextImpl forwards to managedVertex.setParallelism(...) / reconfigureVertex(...):

   grep -n "setParallelism\|reconfigureVertex\|onVertexStarted\|onRootVertexInitialized" \
     tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexManager.java
   

3. A 1-1 edge from an already-sized source. If the vertex has a one-to-one edge from a source whose parallelism is known, its count is inherited.

Until one of these fires, the vertex stays in INITIALIZING. The most common "stuck" bug in all of Tez is a vertex parked in INITIALIZING because an initializer never reported back — see the symptoms table.

Note: INITED is distinct from RUNNING for a reason. At INITED the tasks exist as TaskImpl objects but no attempt has been requested from the scheduler. V_START is what triggers VertexManagerPlugin.onVertexStarted, which drives the first T_SCHEDULE events. A DAG may init all its vertices up front but start them in dependency order, which is why the two states are separate.

Scheduling is delegated, not owned

VertexImpl never decides which tasks to schedule when — that policy lives in the pluggable VertexManagerPlugin, wrapped by VertexManager. The vertex merely forwards lifecycle facts to the plugin and lets it call back to schedule tasks or reconfigure parallelism:

grep -n "onVertexStarted\|onSourceTaskCompleted\|scheduleTasks\|reconfigureVertex" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexManager.java | head

On V_START, VertexManager.onVertexStarted runs the plugin's onVertexStarted callback; as upstream attempts complete, the vertex forwards V_SOURCE_TASK_ATTEMPT_COMPLETED into the plugin, which is how a ShuffleVertexManager decides to lower a reducer's parallelism once it has seen enough source output sizes. The plugin's callbacks run through VertexManagerEvent wrappers on the AM's executor, not inline on the dispatch thread — the same "heavy work off the dispatcher" discipline as the committer. This separation is why the same VertexImpl runs a fixed-parallelism map, an auto-reducing shuffle join, and a custom user vertex manager without any special-casing in the state machine itself.

Task completion and the commit decision

When a task completes, TaskImpl emits V_TASK_COMPLETED, and the vertex's TaskCompletedTransition calls the shared bookkeeping routine checkTasksForCompletion. It logs a line you will grep for constantly during incidents:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
private static String constructCheckTasksForCompletionLog(VertexImpl vertex) {
  String logLine = vertex.logIdentifier
      + ", tasks=" + vertex.numTasks
      + ", failed=" + vertex.failedTaskCount
      + ", killed=" + vertex.killedTaskCount
      + ", success=" + vertex.succeededTaskCount
      + ", completed=" + vertex.completedTaskCount
      + ", commits=" + vertex.commitFutures.size()
      + ", err=" + vertex.terminationCause;
  return logLine;
}

When all tasks have succeeded, the vertex must decide whether to commit now or defer to DAG success. This is the crux of the commit protocol.

commit-on-vertex-success vs commit-on-DAG-success

grep -n "TEZ_AM_COMMIT_ALL_OUTPUTS_ON_DAG_SUCCESS" \
  tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java

// tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java
public static final String TEZ_AM_COMMIT_ALL_OUTPUTS_ON_DAG_SUCCESS =
    TEZ_AM_PREFIX + "commit-all-outputs-on-dag-success";
public static final boolean TEZ_AM_COMMIT_ALL_OUTPUTS_ON_DAG_SUCCESS_DEFAULT = true;

The default is true: outputs are committed only after the whole DAG succeeds, coordinated by DAGImpl, not by individual vertices. This is the safe default — if a later vertex fails, nothing has been published yet, so a partial DAG never leaves visible garbage in the sink.

Set it to false and each vertex commits its own outputs as soon as it succeeds. That lets downstream consumers or external readers see a vertex's output earlier, at the cost of publishing data that a later DAG failure cannot un-publish. DAGImpl reads the flag once and stores it:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java
commitAllOutputsOnSuccess = dagConf.getBoolean(
    TezConfiguration.TEZ_AM_COMMIT_ALL_OUTPUTS_ON_DAG_SUCCESS,
    TezConfiguration.TEZ_AM_COMMIT_ALL_OUTPUTS_ON_DAG_SUCCESS_DEFAULT);
// ...
if (!commitAllOutputsOnSuccess && isCommittable()) {
  // per-vertex commit path
}

How the commit actually runs

Crucially, the committer does not run on the dispatch thread. When a vertex enters commit, it submits each OutputCommitter.commitOutput() to the AM's shared executor as a CallableEvent, and stays in COMMITTING until the futures complete:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
CallableEvent commitCallableEvent = new CallableEvent(commitCallback) {
  @Override
  public Void call() throws Exception {
    vertex.dagUgi.doAs(new PrivilegedExceptionAction<Void>() {
      @Override
      public Void run() throws Exception {
        LOG.info("Invoking committer commit for output=" + outputName
            + ", vertexId=" + vertex.logIdentifier);
        committer.commitOutput();
        return null;
      }
    });
    return null;
  }
};
ListenableFuture<Void> commitFuture =
    vertex.getAppContext().getExecService().submit(commitCallableEvent);
Futures.addCallback(commitFuture, commitCallableEvent.getCallback(), GuavaShim.directExecutor());
vertex.commitFutures.put(outputName, commitFuture);
// ...
if (vertex.commitFutures.isEmpty()) {
  return vertex.finished(VertexState.SUCCEEDED);
} else {
  return VertexState.COMMITTING;
}

Each committer's completion fires a V_COMMIT_COMPLETED event back into the vertex; when the last one lands, the CommitCompletedTransition moves COMMITTING → SUCCEEDED. A commit failure moves COMMITTING → TERMINATING. This is why COMMITTING is a distinct state and not folded into RUNNING: a vertex can spend real wall-clock time in commit, and it must remain interruptible (a V_TERMINATE during commit is legal and routes to TERMINATING).

Vertex groups. When several vertices write to a shared VertexGroup (a named output shared across vertices), the commit is coordinated at the DAG level using VertexGroupCommitStartedEvent / VertexGroupCommitFinishedEvent, so the group is committed exactly once rather than once per member. Grep the DAG side:

grep -n "VertexGroupCommit\|commitAllOutputsOnSuccess\|isVertexGroupCommit" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java

The TERMINATING drain

A vertex cannot jump straight to FAILED or KILLED while tasks are still running — those tasks hold containers and may be mid-commit. TERMINATING is the drain state: on entry the vertex issues kills to all non-terminal tasks, then waits. As each task reports completion (success, failure, or killed), checkTasksForCompletion re-evaluates; when completedTaskCount reaches the task total, finished(...) selects the terminal state from the recorded terminationCause. A VERTEX_KILL-family cause yields KILLED; a failure cause yields FAILED.

The events that drive RUNNING → TERMINATING are worth memorizing because they are the vertex's failure surface:

• V_TERMINATE — an explicit kill (client, DAG kill, upstream vertex failure).
• V_TASK_COMPLETED carrying a TaskState.FAILED when a task exhausted its attempt budget (see task-lifecycle.md).
• V_ROOT_INPUT_FAILED — an initializer failed (only reachable from INITIALIZING, not RUNNING, but it lands in TERMINATING too).
• V_MANAGER_USER_CODE_ERROR — the VertexManagerPlugin threw.
• V_INTERNAL_ERROR — routes to ERROR, a distinct terminal.

Warning: If a container's heartbeat lingers past a kill, a task can sit in the drain longer than expected and the vertex appears stuck in TERMINATING with one task outstanding. That is a heartbeat-timeout tuning issue (tez.task.timeout-ms, default 300000), not a state-machine bug — see task-attempt-lifecycle.md.

Rolling up to the DAG

A vertex is a leaf in the DAG's own state machine. When a vertex reaches a terminal state, its VertexStateChangedCallback (registered via augmentStateMachine, see state-machines.md) notifies subscribers, and a DAGEventVertexCompleted is sent to DAGImpl. The DAG keeps its own completion accounting, mirroring the vertex's checkTasksForCompletion at one level up:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java
static DAGState checkVerticesForCompletion(DAGImpl dag) {
  // ... logs numCompletedVertices=... etc.
  if (dag.numCompletedVertices > dag.numVertices) {
    LOG.error("vertex completion accounting issue: numCompletedVertices > numVertices" /* ... */);
  }
  // ...
}

The DAGState enum is the vertex enum minus a couple of states — verify:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/DAGState.java
public enum DAGState {
  NEW, INITED, RUNNING, SUCCEEDED, FAILED, KILLED, ERROR, TERMINATING, COMMITTING,
}

Note the DAG has COMMITTING and TERMINATING for the same reasons the vertex does: with tez.am.commit-all-outputs-on-dag-success=true (the default), the DAG enters COMMITTING after all vertices succeed and drives every vertex's committer from there. This is why the default-mode commit is DAG-coordinated — the vertex signals readiness by succeeding, and the DAG owns the actual commit phase. Trace it:

grep -n "DAG_VERTEX_COMPLETED\|checkVerticesForCompletion\|commitAllOutputsOnSuccess\|COMMITTING" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java | head

---

## Reading exercise

```bash
cd /path/to/tez
# 1. The full state machine block
sed -n '/= new StateMachineFactory<VertexImpl/,/installTopology()/p' \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java | less

# 2. Every transition group header
grep -n "Transitions from" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java

# 3. The deferred-parallelism branch
grep -n "numTasks == -1\|setParallelism\|hasInputInitializers" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java | head

# 4. The commit machinery
grep -n "commitOutput\|commitFutures\|V_COMMIT_COMPLETED\|COMMITTING" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java | head -20

Then answer:

1. List every reachable state in the V_INIT arc from NEW and, from the InitTransition body, the condition that selects each.
2. Why does RootInputInitializedTransition route events through vertexManager.onRootVertexInitialized rather than fixing parallelism itself?
3. What determines FAILED vs KILLED when TERMINATING drains? Find the finished(...) call and the terminationCause field.
4. With tez.am.commit-all-outputs-on-dag-success at its default, which class runs the commit and when — the vertex or the DAG?
5. Why is commitOutput() submitted to execService instead of run inline in the transition? (Cross-reference event-routing.md on queue backup.)
6. Trace a V_TERMINATE that arrives while the vertex is in COMMITTING. What happens to the outstanding commit futures?

Common bugs and symptoms

Validation: prove you understand this

1. From memory, list all ten VertexState values with a one-line meaning, then verify against VertexState.java.
2. Predict the next state for each pair and verify against the source: (NEW, V_TERMINATE), (INITIALIZING, V_TERMINATE), (RUNNING, V_TASK_RESCHEDULED), (COMMITTING, V_TERMINATE).
3. Set tez.am.commit-all-outputs-on-dag-success=false on a two-vertex DAG in MiniTezCluster. Observe from the logs which vertex commits when, and contrast with the default.
4. Read one VertexManagerPlugin (e.g. ShuffleVertexManager in tez-runtime-library) and trace how it calls reconfigureVertex to set parallelism — then follow the call into VertexImpl.
5. Write a TestVertexImpl-style test that drives a vertex NEW → SUCCEEDED with a mock initializer, using DrainDispatcher.await() between phases.
6. Add a no-op ignorable transition for some (state, event) pair absent from VertexImpl, update TestVertexImpl in the same patch, and compile.