State Machines

Every long-lived entity inside the Tez ApplicationMaster is a state machine: DAGImpl, VertexImpl, TaskImpl, TaskAttemptImpl, AMContainerImpl, AMNodeImpl. Each one is built from the same factory API, driven by the same dispatcher discipline, and fails in the same recognizable ways. This chapter is the foundation for the four lifecycle chapters that follow (event-routing.md, vertex-lifecycle.md, task-lifecycle.md, task-attempt-lifecycle.md) — none of them makes sense until you can read a StateMachineFactory chain fluently.

After this chapter you can: locate any state machine in the AM and count its transitions, explain the difference between single-arc, multiple-arc, and ignorable transitions, describe exactly what happens when an InvalidStateTransitonException fires in each entity, explain what Tez's own StateMachineTez wrapper adds on top of Hadoop's factory, and generate a Graphviz picture of the whole machine from the build.

Where the framework lives

Tez does not ship its own transition engine. The factory, the transition interfaces, and the exception all come from hadoop-yarn-common:

cd /path/to/tez
grep -n "import org.apache.hadoop.yarn.state" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
import org.apache.hadoop.yarn.state.InvalidStateTransitonException;
import org.apache.hadoop.yarn.state.MultipleArcTransition;
import org.apache.hadoop.yarn.state.SingleArcTransition;
import org.apache.hadoop.yarn.state.StateMachine;
import org.apache.hadoop.yarn.state.StateMachineFactory;
// ...
import org.apache.tez.state.StateMachineTez;

Two things to burn in:

1. The exception really is spelled InvalidStateTransitonException — a historical typo in Hadoop, preserved for compatibility. A grep for InvalidStateTransitionException (correct spelling) finds nothing.
2. That last import is Tez's own: org.apache.tez.state.StateMachineTez, a thin decorator that lives in tez-dag and adds state-entered callbacks. More on it below.

Note: Because the engine is Hadoop's, the semantics are identical to the MapReduce AM's state machines. If you have read RMAppImpl or JobImpl in Hadoop, you already know the mechanics; what is Tez-specific is which states and events exist and what the transitions do.

The factory API

Each entity declares a single static final factory, built as one long chained expression, terminated by installTopology(). The real opening of TaskImpl's machine:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
private static final StateMachineFactory
             <TaskImpl, TaskStateInternal, TaskEventType, TaskEvent>
          stateMachineFactory
         = new StateMachineFactory<TaskImpl, TaskStateInternal, TaskEventType, TaskEvent>
             (TaskStateInternal.NEW)

  // Transitions from NEW state
  // Stay in NEW in recovery when Task is killed in the previous AM
  .addTransition(TaskStateInternal.NEW,
      EnumSet.of(TaskStateInternal.NEW, TaskStateInternal.SCHEDULED),
      TaskEventType.T_SCHEDULE, new InitialScheduleTransition())
  .addTransition(TaskStateInternal.NEW, TaskStateInternal.KILLED,
      TaskEventType.T_TERMINATE,
      new KillNewTransition())
  // ...
  .installTopology();

The four generic parameters, in order: the operand (the entity instance passed into every transition), the state enum, the event type enum, and the event class. The constructor argument is the initial state.

addTransition comes in several arities; the three you will meet constantly:

The two hook interfaces:

Tez leans on MultipleArcTransition far more than MapReduce did, because recovery replays events into machines that may already know their outcome. Look at InitialScheduleTransition above: T_SCHEDULE in NEW can legally land in NEW or SCHEDULED, because during recovery the task may need to wait for a replayed terminal event instead of scheduling an attempt.

installTopology() freezes the transition table and returns the factory. Every addTransition call actually returns a new immutable factory — that is why the whole thing is one chained expression assigned once to a static final field: the table is built once per class, then instantiated cheaply per object with make(this):

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
this.stateMachine = stateMachineFactory.make(this);

The census

Count the machines yourself — the counts drift between branches, so never quote them without re-running this:

for f in tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java \
         tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java \
         tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java \
         tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java \
         tez-dag/src/main/java/org/apache/tez/dag/app/rm/container/AMContainerImpl.java \
         tez-dag/src/main/java/org/apache/tez/dag/app/rm/node/AMNodeImpl.java; do
  echo "$(grep -c addTransition $f)  $f"
done

On current master that prints roughly: VertexImpl 61, TaskAttemptImpl 48, DAGImpl 46, AMContainerImpl 41, AMNodeImpl 29, TaskImpl 26. The state enums live in separate files — internal states under tez-dag/src/main/java/org/apache/tez/dag/app/dag/ (VertexState, TaskStateInternal, TaskAttemptStateInternal, DAGState) and the scheduler-side ones under .../dag/app/rm/ (AMContainerState, AMNodeState). The internal/external state split (why TaskStateInternal exists at all) is covered in task-lifecycle.md.

StateMachineTez: Tez's one addition

ls tez-dag/src/main/java/org/apache/tez/state/
grep -n "new StateMachineTez" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/*.java

DAGImpl, VertexImpl, and TaskImpl do not use the raw Hadoop machine; they wrap it:

// tez-dag/src/main/java/org/apache/tez/state/StateMachineTez.java
public class StateMachineTez<STATE extends Enum<STATE>, EVENTTYPE extends Enum<EVENTTYPE>,
    EVENT, OPERAND> implements StateMachine<STATE, EVENTTYPE, EVENT> {

  private final Map<STATE, OnStateChangedCallback> callbackMap = new HashMap<>();
  // ...
  @Override
  public STATE doTransition(EVENTTYPE eventType, EVENT event) throws
      InvalidStateTransitonException {
    STATE oldState = realStatemachine.getCurrentState();
    STATE newState = realStatemachine.doTransition(eventType, event);
    if (newState != oldState) {
      OnStateChangedCallback callback = callbackMap.get(newState);
      if (callback != null) {
        callback.onStateChanged(operand, newState);
      }
    }
    return newState;
  }
}

It fires a registered OnStateChangedCallback whenever a transition enters a state — but only on genuine state changes, not self-loops. VertexImpl uses this to notify the StateChangeNotifier (which feeds VertexManagerPlugins and InputInitializers that subscribed to vertex state updates):

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
private void augmentStateMachine() {
  stateMachine
      .registerStateEnteredCallback(VertexState.SUCCEEDED, STATE_CHANGED_CALLBACK)
      .registerStateEnteredCallback(VertexState.FAILED, STATE_CHANGED_CALLBACK)
      .registerStateEnteredCallback(VertexState.KILLED, STATE_CHANGED_CALLBACK)
      .registerStateEnteredCallback(VertexState.RUNNING, STATE_CHANGED_CALLBACK)
      .registerStateEnteredCallback(VertexState.INITIALIZING, STATE_CHANGED_CALLBACK);
}

TaskAttemptImpl is the odd one out: it uses the raw stateMachineFactory.make(this) with no wrapper, because nothing external subscribes to attempt-level state entry.

Tip: When you need "run code every time we land in state X, regardless of which arc got us there", a state-entered callback is cleaner than editing every inbound transition. That is precisely the pattern registerStateEnteredCallback exists for.

Reading a multiple-arc hook: a worked example

The single skill that separates someone who can modify a Tez state machine from someone who can only stare at one is reading a MultipleArcTransition and knowing, before you run anything, which state it will return. Work through TaskImpl's InitialScheduleTransition — the arc that opens the whole task machine — because it exercises every pattern you will meet:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
private static class InitialScheduleTransition
  implements MultipleArcTransition<TaskImpl, TaskEvent, TaskStateInternal> {
  @Override
  public TaskStateInternal transition(TaskImpl task, TaskEvent event) {
    if (task.recoveryData != null) {
      TaskStartedEvent tStartedEvent = task.recoveryData.getTaskStartedEvent();
      TaskFinishedEvent tFinishedEvent = task.recoveryData.getTaskFinishedEvent();
      // If TaskStartedEvent is not seen but TaskFinishedEvent is seen, that means
      // Task is killed before it is started. Just send T_TERMINATE to itself to move to KILLED
      if (tStartedEvent == null && tFinishedEvent != null) {
        // ... send TaskEventTermination, then:
        return TaskStateInternal.NEW;
      }
    } else {
      task.scheduledTime = task.clock.getTime();
      task.logJobHistoryTaskStartedEvent();
      task.vertex.reportTaskStartTime(task.getLaunchTime());
    }
    // ... normal path: create and schedule the first attempt
    if (!task.addAndScheduleAttempt(null)) {
      return TaskStateInternal.FAILED;
    }
    // ...  returns SCHEDULED on success
  }
}

The declared EnumSet on the addTransition line is EnumSet.of(NEW, SCHEDULED), yet the body clearly can return FAILED too via addAndScheduleAttempt failing. That mismatch is exactly the kind of thing the factory validates at runtime — if this hook ever returns a state not in the declared set, the factory throws. So the declared set is a contract you must keep in sync with the body. Three lessons:

1. The return value is the arc. There is no "next state" argument on a multiple-arc addTransition; the hook is the decision.
2. Recovery is a first-class branch. task.recoveryData != null is the replay path; nearly every opening transition in every Tez machine forks on it. When reading, split the hook into "fresh run" and "recovery" halves.
3. The declared EnumSet is load-bearing. Widen it when you add a return path, or you will ship a latent InvalidStateTransiton-adjacent crash that only fires on the rare arc.

The handle() contract

State machines are not thread-safe. Correctness rests on two mechanisms working together: the single dispatcher thread (see event-routing.md) and a per-entity ReentrantReadWriteLock. Every entity's handle() looks like VertexImpl's:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
public void handle(VertexEvent event) {
  // ...
  try {
    writeLock.lock();
    VertexState oldState = getInternalState();
    try {
       getStateMachine().doTransition(event.getType(), event);
    } catch (InvalidStateTransitonException e) {
      String message = "Invalid event " + event.getType() + " on vertex " /* ... */;
      LOG.error("Can't handle " + message, e);
      addDiagnostic(message);
      eventHandler.handle(new VertexEvent(this.vertexId,
          VertexEventType.V_INTERNAL_ERROR));
    } catch (RuntimeException e) {
      // ... same idea, guarded by internalErrorTriggered.getAndSet(true)
    }
    if (oldState != getInternalState()) {
      LOG.info(logIdentifier + " transitioned from " + oldState + " to "
          + getInternalState() + " due to event " + event.getType());
    }
  } finally {
    writeLock.unlock();
  }
}

The pattern to internalize:

• Writes happen only inside handle(), under writeLock. Transition hooks mutate fields freely because they always execute under that lock.
• Getters take readLock. Other threads (the RPC handler answering DAGClient status calls, the web UI, the speculator) read a consistent snapshot without blocking dispatch except momentarily.
• The transition log line is your debugging bread and butter. Every real state change prints transitioned from X to Y due to event Z. When you reconstruct an incident from an AM log, you grep for exactly these lines.

Each entity handles InvalidStateTransitonException slightly differently, and the differences matter when you read logs:

That is worth restating: a single unregistered (state, event) pair anywhere in the AM takes the whole DAG to ERROR. Tez treats an invalid transition as a protocol bug, not a recoverable hiccup. This is why terminal states carry long lists of ignorable transitions — races that deliver late events to finished entities are normal, and each one must be declared harmless.

The best-documented example is in TaskImpl, complete with the JIRA that motivated it:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
// Transitions from KILLED state
// Ignorable event: T_ATTEMPT_KILLED
// Refer to TEZ-2379
// T_ATTEMPT_KILLED can show up in KILLED state as
// a SUCCEEDED attempt can still transition to KILLED after receiving
// a KILL event.
// ...
.addTransition(TaskStateInternal.KILLED, TaskStateInternal.KILLED,
    EnumSet.of(
        TaskEventType.T_TERMINATE,
        TaskEventType.T_SCHEDULE,
        TaskEventType.T_ADD_SPEC_ATTEMPT,
        TaskEventType.T_ATTEMPT_LAUNCHED,
        TaskEventType.T_ATTEMPT_SUCCEEDED,
        TaskEventType.T_ATTEMPT_FAILED,
        TaskEventType.T_ATTEMPT_KILLED))

Read that comment carefully — it describes a specific race (kill arrives, attempt succeeds concurrently, task gets both T_ATTEMPT_SUCCEEDED and T_ATTEMPT_KILLED from the same attempt) and the no-op that absorbs it. This is the standard of justification expected when you add an ignorable transition.

sequenceDiagram
    participant Q as AsyncDispatcher queue
    participant D as Dispatch thread
    participant E as VertexImpl (writeLock)
    participant SM as StateMachineTez
    participant CB as OnStateChangedCallback

    Q->>D: take() event
    D->>E: handle(event)
    E->>E: writeLock.lock()
    E->>SM: doTransition(type, event)
    SM->>SM: hook.transition(operand, event)
    alt state actually changed
        SM->>CB: onStateChanged(operand, newState)
    end
    alt no registered (state, event) arc
        SM-->>E: InvalidStateTransitonException
        E->>Q: V_INTERNAL_ERROR (DAG will go to ERROR)
    end
    E->>E: writeLock.unlock()

Warning: Never do blocking work inside a transition hook. The hook runs on the central dispatch thread holding the entity's write lock; a slow HDFS call there stalls every event in the AM. Tez's own committers run on a separate execService via CallableEvent for exactly this reason — see the commit machinery in vertex-lifecycle.md.

How to read a state machine

The idiom, once you have done it twice, takes minutes:

1. grep -n "stateMachineFactory" <file> — jump to the factory.
2. The chain is grouped by comments: // Transitions from NEW state, // Transitions from RUNNING state, and so on. Read one group at a time.
3. For every multiple-arc transition, open the hook class (they are private static class ...Transition in the same file) and find its return statements — those are the actual arcs.
4. Note the ignorable EnumSet no-ops in terminal states; each should have a comment explaining the race it absorbs.
5. Cross-check against the test: TestTaskImpl, TestVertexImpl, TestTaskAttempt in tez-dag/src/test/java/.../dag/impl/ drive these machines directly with a DrainDispatcher.

# All transition hook classes in VertexImpl
grep -n "private static class.*Transition\|public static class.*Transition" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java | head -30

# All state-group comments in TaskAttemptImpl
grep -n "Transitions from" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java

DrainDispatcher: deterministic tests

Tez carries its own test-side dispatcher, tez-common/src/test/java/org/apache/tez/common/DrainDispatcher.java (same idea as YARN's, but matching Tez's AsyncDispatcher). Tests enqueue events and then call await(), which blocks until the queue is empty — turning an asynchronous machine into a synchronous, assertable one:

grep -rln "DrainDispatcher" tez-dag/src/test/java | head

Forgetting await() between "send event" and "assert state" is the number one cause of flaky state-machine tests: locally the dispatch thread wins the race, on a loaded CI box it doesn't.

Visualizing the machines

You do not have to draw these by hand. tez-dag/pom.xml ships a visualize Maven profile that runs Hadoop's VisualizeStateMachine over all six classes and emits Graphviz:

<!-- tez-dag/pom.xml -->
<tez.dag.state.classes>org.apache.tez.dag.app.dag.impl.DAGImpl,
org.apache.tez.dag.app.dag.impl.VertexImpl,
org.apache.tez.dag.app.dag.impl.TaskImpl,
org.apache.tez.dag.app.dag.impl.TaskAttemptImpl,
org.apache.tez.dag.app.rm.node.AMNodeImpl,
org.apache.tez.dag.app.rm.container.AMContainerImpl</tez.dag.state.classes>
<tez.graphviz.output.file>${project.build.directory}/Tez.gv</tez.graphviz.output.file>

# From the tez source root (needs a prior full build for dependencies)
mvn -pl tez-dag compile -Pvisualize -DskipTests
# Then render with graphviz:
dot -Tpng tez-dag/target/Tez.gv -o /tmp/tez-state-machines.png

The main class is org.apache.hadoop.yarn.state.VisualizeStateMachine — the factory carries enough metadata at runtime to dump its own topology. Generate this once and pin the image next to your monitor while you work through the lifecycle chapters; it is the authoritative diagram, unlike anything hand-drawn (including the mermaid sketches in this book, which are simplified).

As a small concrete example, here is AMContainerImpl's machine (7 states, the simplest of the six after AMNodeImpl) — verify against tez-dag/src/main/java/org/apache/tez/dag/app/rm/container/AMContainerState.java:

stateDiagram-v2
    [*] --> ALLOCATED
    ALLOCATED --> LAUNCHING: launch request
    LAUNCHING --> IDLE: launched, no task
    IDLE --> RUNNING: task assigned
    RUNNING --> IDLE: task completed (container reuse)
    IDLE --> STOP_REQUESTED: no more work / timeout
    RUNNING --> STOP_REQUESTED: stop
    STOP_REQUESTED --> STOPPING: NM stop failed, RM stop issued
    STOP_REQUESTED --> COMPLETED: NM confirmed
    STOPPING --> COMPLETED: RM confirmed
    COMPLETED --> [*]

The IDLE ⇄ RUNNING loop is what container reuse looks like at the state level — see container-reuse.md.

How to add a transition safely

The review checklist Tez committers apply to any state-machine patch:

1. Read the whole group for the source state first. Your event may already be handled by an EnumSet arc you missed.
2. Decide the shape. If the next state depends on runtime data, use MultipleArcTransition and declare every reachable state in the EnumSet — the factory validates returned states against it at runtime.
3. Justify no-ops in a comment, ideally with the JIRA, following the TEZ-2379 pattern above. An unexplained no-op is a review blocker.
4. Sweep the terminal states. If you introduce a new event type, every terminal state (SUCCEEDED, FAILED, KILLED, plus KILL_WAIT-style draining states) that could receive it late needs an ignorable arc.
5. Check recovery. Recovery replays history events through the same machines (see the recovery arcs in task-attempt-lifecycle.md); your new arc must tolerate replayed sequences.
6. Patch the test in the same commit — TestVertexImpl, TestTaskImpl, or TestTaskAttempt, using DrainDispatcher.await() between event and assertion.

Reading exercise

cd /path/to/tez
# 1. The factory chain start and terminator in TaskImpl
grep -n "stateMachineFactory\|installTopology" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java

# 2. Every multiple-arc transition in TaskImpl and its declared state sets
grep -n -A3 "EnumSet.of(TaskStateInternal" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java | head -40

# 3. Tez's wrapper and its callback interface
cat tez-dag/src/main/java/org/apache/tez/state/OnStateChangedCallback.java

# 4. The exception's usage (note the spelling)
grep -rn "InvalidStateTransitonException" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/ | wc -l

Then answer:

1. Why is the factory static final while the machine from make(this) is an instance field? What would break if the factory were per-instance?
2. InitialScheduleTransition in TaskImpl can return NEW or SCHEDULED for the same T_SCHEDULE event. Read its body: what decides?
3. Which of the six machines has no StateMachineTez wrapper, and why does it not need state-entered callbacks?
4. Find the internalErrorTriggered AtomicBoolean in VertexImpl.handle. What double-fire does it prevent?
5. In the TEZ-2379 comment, reconstruct the exact event interleaving that delivers T_ATTEMPT_KILLED to a task already in KILLED.
6. Run the visualize profile. Compare VertexImpl's rendered graph with the happy-path table in vertex-lifecycle.md — what arcs does the table omit?

Common bugs and symptoms

Validation: prove you understand this

1. From memory, write the addTransition signature for all three shapes (single-arc, multiple-arc, ignorable) and state what each hook returns.
2. Without looking: which entity converts an invalid transition into V_INTERNAL_ERROR sent to itself, and which two send DAG-level INTERNAL_ERROR directly? Verify with grep.
3. Implement a toy three-state machine (OFF/ON/BROKEN, events TOGGLE, BREAK) against org.apache.hadoop.yarn.state.StateMachineFactory, including one ignorable arc, and unit-test it.
4. Generate Tez.gv via the visualize profile and produce a PNG of just the TaskImpl subgraph.
5. Pick one MultipleArcTransition in VertexImpl (e.g. the V_INIT arc from NEW, which declares EnumSet.of(NEW, INITED, INITIALIZING, FAILED, KILLED)), read its hook, and write one sentence per reachable state describing the condition that selects it.
6. Add a hypothetical ignorable arc to a scratch copy of TaskImpl for a (state, event) pair currently absent — then explain why master may already have it and what that tells you about session-mode DAG re-submission.