Lab 5.2: Add a Missing TestVertexImpl Transition Test

Lab type: Fix-It (test coverage) Estimated time: 120 min Tez module: tez-dag Key class: org.apache.tez.dag.app.dag.impl.TestVertexImpl

Background

TestVertexImpl is the unit-test harness for the most complex state machine in Tez, VertexImpl. It has 100+ test methods and is ~7,600 lines long, but its structure is simple and worth learning cold, because TestDAGImpl, TestTaskImpl, and TestTaskAttemptImpl all follow the same pattern:

arrange a state → fire a VertexEvent → dispatcher.await() → assert on VertexState.

The engine that makes this deterministic is DrainDispatcher (tez-common/src/test/java/org/apache/tez/common/DrainDispatcher.java), a subclass of YARN's AsyncDispatcher whose await() busy-loops until the event queue is empty. Because the test blocks until every queued event has been processed, there is no race and no Thread.sleep.

In this lab you will dissect the harness, build a coverage matrix of which (VertexState, VertexEventType) transitions have tests, find a genuine gap, and add a new transition test in the exact style of the existing ones. "Add coverage for transition X" is one of the most common, most-welcomed first patches a new Tez contributor lands.

Why This Lab Matters for Contributors

State-machine coverage gaps are real, discoverable, and low-risk to fill — the ideal shape of a first tez-dag patch.
Writing one transition test forces you to understand the harness well enough to write the next one, which you will need for every state-machine bug fix you ever submit.
Committers reject state-machine changes with no transition test. This lab is where you learn to produce the artifact they demand.

Prerequisites

Lab 5.1 complete; export TEZ_SRC=/path/to/tez; mvn install -DskipTests -q succeeded.
Read the State Machines and Vertex Lifecycle deep dives, and Level 4's state-machine reading lab.
You can run one test method: mvn test -pl tez-dag -Dtest='TestVertexImpl#testVertexInit' -q.

Step-by-Step Tasks

Step 1: Enumerate the state machine

The transition table is the StateMachineFactory in VertexImpl:

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

Each addTransition call declares a (fromState, eventType) -> {toState(s), transitionAction} arc. The factory is declared like this (VertexImpl):

protected static final
  StateMachineFactory<VertexImpl, VertexState, VertexEventType, VertexEvent>
     stateMachineFactory
   = 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,
                new InitTransition())
        ...

The event alphabet is VertexEventType:

grep -n "V_" "$TEZ_SRC"/tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/VertexEventType.java

The states are VertexState: NEW, INITIALIZING, INITED, RUNNING, COMMITTING, TERMINATING, SUCCEEDED, FAILED, KILLED, ERROR (grep VertexState.java to confirm).

Note: What happens when a state has no addTransition for an incoming event? VertexImpl.handle catches the InvalidStateTransitonException (note the Hadoop typo in the class name), logs "Can't handle Invalid event ...", adds a diagnostic, and fires VertexEventType.V_INTERNAL_ERROR, which drives the vertex to VertexState.ERROR and emits a DAGEventType.INTERNAL_ERROR to the DAG. That behavior is exactly what the existing testInvalidEvent asserts — and the model for the test you will write.

The full VertexState alphabet is:

NEW → INITIALIZING → INITED → RUNNING → { COMMITTING } → SUCCEEDED
                                   ↘ TERMINATING ↘ { FAILED, KILLED }
   (any state, on an undefined event) → ERROR

The happy path and the invalid-event escape hatch, as a diagram:

stateDiagram-v2
    [*] --> NEW
    NEW --> INITIALIZING: V_INIT (needs initializer)
    NEW --> INITED: V_INIT (no initializer)
    INITIALIZING --> INITED: V_READY_TO_INIT
    INITED --> RUNNING: V_START
    RUNNING --> COMMITTING: V_TASK_COMPLETED (last task, has committer)
    RUNNING --> SUCCEEDED: V_TASK_COMPLETED (last task)
    COMMITTING --> SUCCEEDED: V_COMMIT_COMPLETED
    RUNNING --> TERMINATING: V_TERMINATE
    TERMINATING --> KILLED: V_TASK_COMPLETED
    RUNNING --> FAILED: V_TASK_COMPLETED (task failed)
    RUNNING --> ERROR: (undefined event) V_INTERNAL_ERROR
    INITED --> ERROR: (undefined event) V_INTERNAL_ERROR

The invalid-event escape hatch is in VertexImpl.handle (quote it, don't cite a line):

// From VertexImpl.handle(VertexEvent event)
VertexState oldState = getInternalState();
try {
  getStateMachine().doTransition(event.getType(), event);
} catch (InvalidStateTransitonException e) {
  String message = "Invalid event " + event.getType() + " on vertex " + this.vertexName + ...;
  LOG.error("Can't handle " + message, e);
  addDiagnostic(message);
  eventHandler.handle(new VertexEvent(this.vertexId, VertexEventType.V_INTERNAL_ERROR));
}

That emitted V_INTERNAL_ERROR is what carries the vertex to ERROR and, via the DAG's handler, records one DAGEventType.INTERNAL_ERROR. Every "unexpected event in state X" test keys off this exact path.

Step 2: Read the harness

grep -n "class DrainDispatcher\|@BeforeClass\|@Before\b\|@After\b\|setupPreDagCreation\|setupPostDagCreation\|setupVertices\|initAllVertices\|initVertex\|startVertex\|class VertexEventDispatcher\|class DagEventDispatcher" \
  "$TEZ_SRC"/tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java

The lifecycle (methods in TestVertexImpl):

@BeforeClass beforeClass() → MockDNSToSwitchMapping.initializeMockRackResolver().
@Before setup() → setupPreDagCreation(); builds the dagPlan; setupPostDagCreation(false).
setupPreDagCreation() builds conf, appAttemptId, dagId, and a mocked TaskSpecificLaunchCmdOption.
setupPostDagCreation(...) is where the dispatcher and mocks are born:

// From TestVertexImpl.setupPostDagCreation(...)
if (dispatcher != null) {
  dispatcher.stop();
}
dispatcher = new DrainDispatcher();
appContext = mock(AppContext.class);
when(appContext.getHadoopShim()).thenReturn(new DefaultHadoopShim());
// ... many more mocks: ContainerLauncherManager, TaskCommunicatorManagerInterface, AMContainerMap ...
dispatcher.init(conf);
dispatcher.start();

setupVertices(...) constructs each vertex directly from the DAG plan — note the real constructor call:

// From TestVertexImpl.setupVertices(...)
v = new VertexImpl(vertexId, vPlan, vPlan.getName(), conf,
        dispatcher.getEventHandler(), taskCommunicatorManagerInterface,
        clock, thh, true, appContext, locationHint, vertexGroups,
        taskSpecificLaunchCmdOption, updateTracker, dagConf);
vertices.put(vName, v);
vertexIdMap.put(vertexId, v);

The inner VertexEventDispatcher routes a VertexEvent back to the right vertex via vertexIdMap:

private class VertexEventDispatcher implements EventHandler<VertexEvent> {
  @Override
  public void handle(VertexEvent event) {
    VertexImpl vertex = vertexIdMap.get(event.getVertexID());
    ((EventHandler<VertexEvent>) vertex).handle(event);
  }
}

The inner DagEventDispatcher counts DAG events, so a test can assert INTERNAL_ERROR was emitted:

private class DagEventDispatcher implements EventHandler<DAGEvent> {
  public Map<DAGEventType, Integer> eventCount = new HashMap<>();
  // increments eventCount.get(event.getType()) on each handle(...)
}

The state-driving helpers you will reuse:

Helper	What it does
`initVertex(v)`	fires `V_INIT`, `await()`s; asserts the vertex leaves `NEW`
`initAllVertices(VertexState.INITED)`	inits every source vertex and asserts they all reach `INITED`
`startVertex(v)`	fires `V_START`, `await()`s, asserts `RUNNING`
`killVertex(v)`	fires `VertexEventTermination(..., DAG_TERMINATED)`, asserts `KILLED`
`completeAllTasksSuccessfully(v)`	fires `VertexEventTaskCompleted(task, SUCCEEDED)` for every task

Step 3: Study the model test — `testInvalidEvent`

This is the existing test you will pattern-match. It sends V_START to a vertex that is still in NEW (the NEW state has no V_START arc), and asserts the invalid-event handling drives it to ERROR:

@Test(timeout = 5000)
public void testInvalidEvent() {
  VertexImpl v = vertices.get("vertex2");
  dispatcher.getEventHandler().handle(new VertexEvent(v.getVertexId(), VertexEventType.V_START));
  dispatcher.await();
  Assert.assertEquals(VertexState.ERROR, v.getState());
  Assert.assertEquals(1,
      dagEventDispatcher.eventCount.get(DAGEventType.INTERNAL_ERROR).intValue());
}

Run it and confirm it passes:

mvn test -pl tez-dag -Dtest='TestVertexImpl#testInvalidEvent' -q 2>&1 | tail -6

Step 4: Build a coverage matrix

You want to find a (state, event) pair that is not exercised by any test. A fast proxy: count how often each event type appears in the test file. Events that never appear are certainly untested; events that appear only in one state are candidates for gaps in other states.

cd "$TEZ_SRC"
SRC=tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
TST=tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java
for ev in $(grep -oE "V_[A-Z_]+" tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/VertexEventType.java | sort -u); do
  printf "%-32s defined:%-3s referenced-in-test:%s\n" "$ev" \
    "$(grep -cw "$ev" "$SRC")" "$(grep -cw "$ev" "$TST")"
done

Now narrow to invalid transitions — the cheapest kind of test to write because they only require reaching a state and firing an event that state doesn't handle. For a chosen state, list the events it does handle, and anything in VertexEventType not in that list is an invalid-transition candidate:

# Events RUNNING handles (arcs whose 'from' is RUNNING). Read the block by eye:
grep -n "addTransition" "$SRC" | sed -n '/RUNNING/p'   # or open the "Transitions from RUNNING state" block

From the RUNNING block you will find arcs for V_ROOT_INPUT_FAILED, V_TASK_ATTEMPT_COMPLETED, V_SOURCE_TASK_ATTEMPT_COMPLETED, V_TASK_COMPLETED, V_TERMINATE, V_MANAGER_USER_CODE_ERROR, V_TASK_RESCHEDULED, V_COMPLETED, V_INTERNAL_ERROR, and V_ROUTE_EVENT. There is no arc for V_START from RUNNING. So: sending V_START to a vertex already in RUNNING is an untested invalid transition — a genuine, completable gap, and the direct analog of testInvalidEvent (which covers V_START from NEW, not from RUNNING).

Tip: Verify the gap before you write the test — don't trust this doc. grep -n "testStart\|RUNNING\|V_START" "$TST" and confirm no existing method starts a vertex and then fires V_START again asserting ERROR. Transitions move between releases; the method for finding a gap is the durable skill.

Step 5: Write the new transition test

Add this method next to testInvalidEvent in TestVertexImpl. It arranges RUNNING with the existing helpers, fires the invalid V_START, drains, and asserts the invalid-event contract:

@Test(timeout = 5000)
public void testStartWhileRunning() {
  // Arrange: drive vertex2 to RUNNING using the existing helpers.
  initAllVertices(VertexState.INITED);
  VertexImpl v = vertices.get("vertex2");
  startVertex(v);                                   // fires V_START; asserts RUNNING
  Assert.assertEquals(VertexState.RUNNING, v.getState());

  // Act: fire V_START again. RUNNING has no V_START arc -> InvalidStateTransitonException.
  dispatcher.getEventHandler().handle(new VertexEvent(v.getVertexId(), VertexEventType.V_START));
  dispatcher.await();

  // Assert: invalid-event handling drives the vertex to ERROR and emits a DAG INTERNAL_ERROR.
  Assert.assertEquals(VertexState.ERROR, v.getState());
  Assert.assertEquals(1,
      dagEventDispatcher.eventCount.get(DAGEventType.INTERNAL_ERROR).intValue());
}

Why each line is there:

initAllVertices + startVertex reuse the harness so you don't hand-roll the arrange step.
The second V_START has no defined arc from RUNNING; VertexImpl.handle catches InvalidStateTransitonException and fires V_INTERNAL_ERROR.
dispatcher.await() guarantees both the invalid V_START and the follow-up V_INTERNAL_ERROR are fully processed before you assert. This is the whole reason DrainDispatcher exists — never assert before await().

Step 6: Prove the test is meaningful

A good transition test fails if the behavior regresses. Temporarily verify it is actually asserting something by weakening it (then revert):

// TEMPORARY sanity check — comment out the second V_START, keep the asserts.
// The test should now FAIL, proving the asserts depend on the invalid event.

Run just your method:

mvn test -pl tez-dag -Dtest='TestVertexImpl#testStartWhileRunning' -q 2>&1 | tail -8

Restore the second V_START. Expected passing tail:

Tests run: 1, Failures: 0, Errors: 0, Skipped: 0

Step 7: Run the whole class and format the patch

Your test must not disturb the other 100+:

mvn test -pl tez-dag -Dtest=TestVertexImpl -q 2>&1 | tail -6
git -C "$TEZ_SRC" diff -- tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestVertexImpl.java

Draft the JIRA (Tez tracks tests as first-class work):

Summary: TestVertexImpl missing coverage for V_START in RUNNING state
Component: test
Priority: Trivial
Description:
  VertexImpl has no (RUNNING, V_START) transition, so an errant V_START drives the vertex to
  ERROR via the invalid-event path. TestVertexImpl covers (NEW, V_START) in testInvalidEvent but
  not (RUNNING, V_START). This patch adds testStartWhileRunning, asserting VertexState.ERROR and
  one DAGEventType.INTERNAL_ERROR, following the existing testInvalidEvent pattern.

Deliverables

The coverage-matrix command output, with the untested (state, event) pair you chose highlighted.
The new @Test method added to TestVertexImpl, using DrainDispatcher.await() (no Thread.sleep).
Evidence the test is meaningful: it failed when you weakened it (Step 6) and passes when restored.
A green full-class run (-Dtest=TestVertexImpl) proving no regressions.
A git diff and a JIRA summary/description in the format above.

Troubleshooting

Symptom	Cause	Fix
`NullPointerException` on `dagEventDispatcher.eventCount.get(...)`	The event you fired did not actually trigger `INTERNAL_ERROR` (the transition was valid)	You picked a `(state, event)` pair that has an arc; re-check Step 4 and choose a truly undefined pair
Assertion sees `RUNNING`/`INITED`, not `ERROR`	You asserted before draining	Add `dispatcher.await()` after firing the event
Test passes even with the event commented out	Your asserts don't depend on the event	Assert on both `VertexState.ERROR` and the `INTERNAL_ERROR` count, like `testInvalidEvent`
`vertex2` is `null`	Vertex names come from the test DAG plan; `createTestDAGPlan` defines `vertex1..vertexN`	Grep `createTestDAGPlan` for the real vertex names in your checkout
Whole-class run breaks other tests	Shared mutable state or you edited a helper	Only add a method; never mutate `setup`/helpers for one test

Stretch Goals

A second gap. Find another invalid pair — e.g. V_INIT fired at an INITED vertex (INITED has no V_INIT arc) — and add testInitWhileInited by the same recipe.
A valid transition. Pick a valid but untested arc (use the matrix) and assert both the resulting state and a side effect (a counter, a diagnostic string via v.getDiagnostics(), or an emitted event). Model it on testVertexFailure, which asserts VertexState.FAILED, the OWN_TASK_FAILURE termination cause, and a diagnostics substring.
Understand Clock mocking. Some tests inject a mock Clock. Grep for clock in the harness and explain why a state-machine test would ever need to control time (hint: speculation and timeouts).

Deeper Understanding

#	Question
1	What is the difference between `VertexState.FAILED` and `VertexState.ERROR`? When does the AM produce each? (Hint: `FAILED` is a normal terminal outcome — e.g. `OWN_TASK_FAILURE`; `ERROR` is the invalid-event escape hatch.)
2	`setupPostDagCreation` builds a fresh `DrainDispatcher` every time it runs, calling `dispatcher.stop()` on the previous one first. Why must a test that reconfigures the DAG mid-method rebuild the dispatcher rather than reuse it?
3	`VertexImpl` calls methods on a mocked `AppContext` constantly. Grep `appContext.` in `VertexImpl.java` — which methods dominate, and why must the test mock return sensible values for them (not `null`)?
4	Why is `DrainDispatcher` used in tests instead of the production `AsyncDispatcher`? What does `await()` do that `AsyncDispatcher` cannot give a test?
5	Some `TestVertexImpl` tests inject a mock `Clock`. Why would a state-machine test ever need to control time? (Hint: speculation, and `@Test(timeout=...)` interacting with real wall-clock waits.)

Validation / Self-check

What are the four things every arc of the stateMachineFactory declares?
What does VertexImpl.handle do when it receives an event with no defined transition from the current state? Name the exception and the two observable effects.
Why must every assertion come after dispatcher.await()? What would a Thread.sleep(50) instead of await() cost you?
Which helper drives a vertex to RUNNING, and what does it assert internally?
How does the DagEventDispatcher let you assert that a DAGEventType.INTERNAL_ERROR was emitted?
How did you prove your new test is meaningful rather than vacuously passing?
Give the exact Maven command to run only your new method.

When your new transition test passes, the full class is green, and you can explain the invalid-event path, continue to Lab 5.3: Build It — MiniTezCluster Integration Test.

Open-Source Engineer & Contributor