Lab 3.1: Trace a DAG Submission End-to-End

Background

A DAG starts life as a plain Java object — a DAG you assembled with Vertex.create and Edge.create. It ends as a Processor.run() call inside a container JVM on some other machine. Between those two points is a chain of method calls, one protobuf serialization, one IPC hop, a state-machine cascade in the AM, a container launch, and a second IPC handshake back to the AM over the task umbilical. That chain crosses three JVMs (client, AM, task container) and roughly six class boundaries.

This is a code-reading lab. You will not write production code. You will follow one real submission — TezClient.submitDAG(dag) — from the client all the way to LogicalIOProcessorRuntimeTask.run(), using grep, your IDE, and a debugger. The skill being drilled is the same one every Tez committer uses on the first day of a bug report: reconstruct the path from the source, not from documentation.

You will produce two concrete trace recipes and one filled-in trace table:

• (a) an IDE breakpoint session in local mode, with the exact classes and methods to stop at, in order; and
• (b) a log-based trace — the real log lines that mark each hop, and the grep commands that find them.

Everything here is verified against a current Apache Tez master checkout. Class and method names are stable; line numbers are not, so this lab gives you grep/find locators rather than fabricated line numbers. Companion deep dives: tez-client.md, dag-app-master.md, and the Level 3 overview.

Why This Lab Matters for Contributors

When a user files "my DAG hangs in NEW state" or "submitDAG throws SessionNotRunning," the maintainer's first move is to locate the exact hop that failed. Was the plan rejected at the client before it was ever sent? Did the AM's readiness gate never open? Did DAG_INIT fail inside initializeDAG() and silently mark the DAG FAILED? You cannot triage any of these without a mental model of the submission chain that is accurate to the method.

Every later Tez lab — vertex-manager work, scheduler changes, recovery, the multi-input project in Lab 3.3 — assumes you can get from submitDAG to a running processor in your head. Build that model now, from the code, so it is real.

Prerequisites

• Apache Tez cloned and building (mvn -T1C -DskipTests install succeeds). The checkout is your source of truth for every claim in this lab.
• The Lab 3.3 companion project or any local-mode Tez program you can run — you need something that actually calls submitDAG to set breakpoints on.
• An IDE with a debugger and "Find Usages" / "Call Hierarchy" (IntelliJ Ctrl-Alt-H).
• A scratch file for your trace notes:

mkdir -p ~/tez-notes
: > ~/tez-notes/trace-3.1.md

• Set an env var so the grep commands below are short:

export TEZ=~/src/tez            # <-- your Tez checkout root
cd "$TEZ"

The Path at a Glance

sequenceDiagram
    autonumber
    participant App as Your main()
    participant TC as TezClient (tez-api)
    participant FC as FrameworkClient<br/>(LocalClient / YARN)
    participant PB as DAGClientAMProtocol<br/>BlockingPBServerImpl
    participant DH as DAGClientHandler
    participant AM as DAGAppMaster
    participant DAG as DAGImpl
    participant V as VertexImpl
    participant TA as TaskAttemptImpl
    participant CL as ContainerLauncher / NM
    participant Child as TezChild (container JVM)
    participant RT as LogicalIOProcessor<br/>RuntimeTask

    App->>TC: submitDAG(dag)
    TC->>TC: prepareAndCreateDAGPlan(dag) -> DAGPlan
    TC->>FC: submitDag(dag, SubmitDAGRequestProto{DAGPlan}, ...)
    Note over TC,FC: session: IPC to AM · non-session: DAGPlan as tez-dag.pb in launch context
    FC->>PB: submitDAG(controller, request)  [IPC hop]
    PB->>DH: submitDAG(dagPlan, additionalResources)
    DH->>AM: submitDAGToAppMaster(dagPlan, additionalResources)
    AM->>AM: waitToBeReady(); createDAG(dagPlan) -> DAGImpl
    AM->>DAG: dagEventDispatcher.handle(DAGEvent{DAG_INIT})
    Note over DAG: InitTransition.initializeDAG():<br/>create vertices, edges, DAGScheduler -> INITED
    AM->>DAG: sendEvent(DAGEventStartDag{DAG_START})
    DAG->>DAG: StartTransition -> initializeVerticesAndStart()
    DAG->>V: VertexEvent{V_INIT} then V_START (root vertices)
    V->>V: InitTransition -> setupVertex(); propagate V_INIT to targets
    V->>TA: schedule task attempts
    TA->>CL: request + launch container (main class = TezChild)
    CL->>Child: JVM start: TezChild.main(host, port, containerId, ...)
    Child->>Child: umbilical.getTask() -> ContainerTask{TaskSpec}
    Child->>RT: TezTaskRunner2.run() -> runtimeTask.run()
    RT->>RT: processor.run(runInputMap, runOutputMap)

Keep this diagram open in one pane while you verify each hop in the source in the other. Every arrow below is checkable with grep.

Step-by-Step Tasks

Step 1 — The client entry point: TezClient.submitDAG

Open tez-api/src/main/java/org/apache/tez/client/TezClient.java and find submitDAG(DAG dag). It is a two-line fork:

public synchronized DAGClient submitDAG(DAG dag) throws TezException, IOException {
    DAGClient result = isSession ? submitDAGSession(dag) : submitDAGApplication(dag);
    ...
}

That isSession boolean is the single most important branch in the whole path, so learn both sides.

grep -n "submitDAGSession\|submitDAGApplication\|prepareAndCreateDAGPlan\|SubmitDAGRequestProto" \
  tez-api/src/main/java/org/apache/tez/client/TezClient.java | head

Read submitDAGSession(dag). Confirm these facts in the code:

1. It calls TezClientUtils.prepareAndCreateDAGPlan(dag, ...) — this is where the DAG object becomes a DAGPlan protobuf. Note the type: DAGPlan.
2. It wraps the plan: SubmitDAGRequestProto.Builder requestBuilder = SubmitDAGRequestProto.newBuilder(); requestBuilder.setDAGPlan(dagPlan);.
3. There is a size guard. If request.getSerializedSize() > maxSubmitDAGRequestSizeThroughIPC, the plan is written to the staging dir on the filesystem (as tez-dag.pb plus a counter) and the request carries only a serializedRequestPath instead of the inline bytes. This exists because Hadoop IPC caps message size (IPC_MAXIMUM_DATA_LENGTH).
4. Finally: return frameworkClient.submitDag(dag, request, clientName, sessionAppId, clientTimeout, getUgi(), ...).

Now read submitDAGApplication(appId, dag) (non-session). It does not send an IPC request — there is no AM yet. Instead it calls TezClientUtils.createApplicationSubmissionContext(...) and frameworkClient.submitApplication(appContext). The DAG plan travels a different way here — as a YARN local resource. Find it in tez-api/.../client/TezClientUtils.java:

grep -n "TEZ_PB_PLAN_BINARY_NAME\|writeTo\|amLocalResources.put" \
  tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java

You will see the plan serialized to a binary path and registered under TezConstants.TEZ_PB_PLAN_BINARY_NAME — which is the literal string "tez-dag.pb":

grep -n "TEZ_PB_PLAN_BINARY_NAME" tez-api/src/main/java/org/apache/tez/dag/api/TezConstants.java

Log it in ~/tez-notes/trace-3.1.md: the two ways a DAGPlan reaches the AM — inline in an IPC SubmitDAGRequestProto (session), or as the tez-dag.pb local resource in the AM launch context (non-session).

Step 2 — The IPC boundary (and how local mode skips it)

In session mode, frameworkClient is a YARN-backed client and submitDag ends at a proxy call:

grep -n "submitDag\|waitForProxy\|proxy.submitDAG" \
  tez-api/src/main/java/org/apache/tez/client/FrameworkClient.java

You will find SubmitDAGResponseProto response = proxy.submitDAG(null, request); where proxy is a DAGClientAMProtocolBlockingPB. That is the IPC hop — the request leaves the client JVM here.

The AM-side implementation of that protocol is tez-dag/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientAMProtocolBlockingPBServerImpl.java. Read its submitDAG(RpcController, SubmitDAGRequestProto):

• It first enforces ACLs: real.getACLManager().checkAMModifyAccess(user).
• If request.hasSerializedRequestPath(), it reads the plan back off the filesystem (the mirror of the Step 1 size guard).
• Then DAGPlan dagPlan = request.getDAGPlan(); String dagId = real.submitDAG(dagPlan, additionalResources); where real is a DAGClientHandler.

Local mode is the important exception. When tez.local.mode=true, frameworkClient is a LocalClient, and there is no IPC at all. Read tez-dag/src/main/java/org/apache/tez/client/LocalClient.java submitDag(...):

String dagId = dagAppMaster.submitDAGToAppMaster(request.getDAGPlan(), additionalResources);

It calls the AM method directly, in-process. This is exactly why local mode is the right place to set breakpoints — the whole client→AM path collapses into one call stack you can step through.

Step 3 — DAGClientHandler → DAGAppMaster.submitDAGToAppMaster

DAGClientHandler.submitDAG(dagPlan, additionalAmResources) is a thin forwarder:

grep -n "submitDAG\|submitDAGToAppMaster" \
  tez-dag/src/main/java/org/apache/tez/dag/api/client/DAGClientHandler.java

It returns dagAppMaster.submitDAGToAppMaster(dagPlan, additionalAmResources). Open DAGAppMaster.java and read submitDAGToAppMaster. Confirm the guards, in order:

1. appMasterReadinessService.waitToBeReady() — blocks until the AM finished booting. A DAG submitted too early waits here, not forever.
2. SessionNotRunning if the session is stopping.
3. TezException("App master already running a DAG") if currentDAG != null && !currentDAG.isComplete() — Tez runs one DAG at a time per AM.
4. LOG.info("Starting DAG submitted via RPC: " + dagPlan.getName());
5. startDAG(dagPlan, additionalResources); return currentDAG.getID().toString();

That returned string — e.g. dag_1699...0001_1 — is the DAG id the client will poll on.

Step 4 — startDAG and startDAGExecution: where the events are born

Read DAGAppMaster.startDAG(dagPlan, additionalAMResources):

• final DAG newDAG = createDAG(dagPlan); — turns the DAGPlan protobuf back into a live DAGImpl.
• LOG.info("Running DAG: " + dagPlan.getName() + callerContextStr);
• It prints to stdout: System.out.println(timeStamp + " Running Dag: " + newDAG.getID()); — a line you will grep for in Step 8.
• It records a DAGSubmittedEvent in history, then calls startDAGExecution(newDAG, lrDiff) and sets this.state = DAGAppMasterState.RUNNING.

Now the crucial method — startDAGExecution. This is where the state-machine cascade is kicked off. Read it closely; the ordering is deliberate and load-bearing:

sendEvent(new DAGAppMasterEvent(DAGAppMasterEventType.NEW_DAG_SUBMITTED));
DAGEvent initDagEvent = new DAGEvent(currentDAG.getID(), DAGEventType.DAG_INIT);
// This is a synchronous call, not an event through dispatcher. We want
// job-init to be done completely here.
dagEventDispatcher.handle(initDagEvent);
dag.onStart();
DAGEvent startDagEvent = new DAGEventStartDag(currentDAG.getID(), additionalUrlsForClasspath);
sendEvent(startDagEvent);

Three things worth writing down:

• DAG_INIT is dispatched synchronously (dagEventDispatcher.handle(...)), on purpose — the AM wants initialization fully done before it declares the DAG running.
• DAG_START is sent asynchronously (sendEvent(...)), through the AsyncDispatcher. This is what actually triggers execution.
• So the vertex-init cascade does not begin at DAG_INIT. It begins at DAG_START. Verify that next.

Step 5 — DAGImpl: the InitTransition vs StartTransition split

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

grep -n "DAGEventType.DAG_INIT\|DAGEventType.DAG_START\|class InitTransition\|class StartTransition\|initializeVerticesAndStart" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java

Read InitTransition.transition(...). It calls dag.initializeDAG(). Inside initializeDAG():

• it creates a VertexImpl for each vertex in the plan (createVertex(...), addVertex(v));
• it wires edges (createDAGEdges(this), parseVertexEdges(...), then e.initialize() per edge);
• it assigns the DAGScheduler; and
• it returns DAGState.INITED.

Note what InitTransition does not do: it does not post V_INIT to every vertex. The vertices exist as objects but are still in state NEW.

Now read StartTransition.transition(...). Its body ends with dag.initializeVerticesAndStart();. Read that method — it is short and decisive:

protected void initializeVerticesAndStart() {
  for (Vertex v : vertices.values()) {
    if (v.getInputVerticesCount() == 0) {
      eventHandler.handle(new VertexEvent(v.getVertexId(), VertexEventType.V_INIT));
    }
  }
  for (Vertex v : vertices.values()) {
    if (v.getInputVerticesCount() == 0) {
      eventHandler.handle(new VertexEvent(v.getVertexId(), VertexEventType.V_START));
    }
  }
}

Only root vertices (those with no incoming edges) get V_INIT/V_START directly from the DAG. Everything downstream is initialized by the cascade you trace in Step 6. Write this down — it is a common source of confusion: "why didn't my middle vertex get V_INIT when the DAG started?" Because its upstream vertices init it, not the DAG.

Step 6 — VertexImpl.InitTransition: the cascade

Open tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java and find the public InitTransition:

grep -n "class InitTransition\|setupVertex\|numInitedSourceVertices\|V_INIT" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java | head

Read InitTransition.transition(...). The logic that propagates the cascade is:

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));
      }
    }
  }
}

Two facts to record:

1. A vertex only initializes once all its source vertices have initialized (the numInitedSourceVertices == sourceVertices.size() gate). This is how a join/union vertex waits for both inputs — you will rely on this in Lab 3.3.
2. handleInitEvent calls vertex.setupVertex(), which determines the vertex's task parallelism, constructs its VertexManager (e.g. ShuffleVertexManager), and runs any root-input initializers. Then the vertex fans V_INIT out to its targets — the recursion that walks the DAG top to bottom.

Step 7 — Task → container → TezChild

Once a vertex is RUNNING, its TaskImpl/TaskAttemptImpl request containers and the container launcher builds the JVM command. The main class of that JVM is the thing to confirm:

grep -rn "TezChild.class.getName()" tez-dag/src/main/java | grep -v test

You will land in tez-dag/.../dag/utils/TezRuntimeChildJVM.java, where the command vector is built:

vargs.add(TezChild.class.getName());  // main of Child
vargs.add(taskAttemptListenerAddr.getAddress().getHostName());
vargs.add(Integer.toString(taskAttemptListenerAddr.getPort()));
vargs.add(containerIdentifier);
vargs.add(tokenIdentifier);
vargs.add(Integer.toString(applicationAttemptNumber));

So TezChild learns where the AM's umbilical is and who it is purely from main(String[] args) — host, port, container id, token id, attempt number. Confirm by opening tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TezChild.java and reading main:

grep -n "public static void main\|args\[" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TezChild.java | head

TezChild does not receive its task in the launch command. It asks for one. Read its run() loop: a ContainerReporter calls the umbilical (TezTaskUmbilicalProtocol) to get a ContainerTask; if shouldDie it exits, otherwise it builds a TezTaskRunner2 around containerTask.getTaskSpec() and calls taskRunner.run(). This second IPC — container asking the AM "what should I run?" — is the umbilical handshake.

Step 8 — LogicalIOProcessorRuntimeTask.run(): the finish line

TezTaskRunner2 ultimately runs a LogicalIOProcessorRuntimeTask. Open tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java and read run():

public void run() throws Exception {
  Preconditions.checkState(this.state.get() == State.INITED, ...);
  this.state.set(State.RUNNING);
  processor.run(runInputMap, runOutputMap);
}

That processor.run(...) is the call your AbstractLogicalIOProcessor finally receives. Note the map keys — you will need this in Lab 3.2 and Lab 3.3: the input map is keyed by source vertex name, the output map by destination vertex name:

grep -n "runInputMap.put\|runOutputMap.put" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java

You will see runInputMap.put(inputSpec.getSourceVertexName(), input) and runOutputMap.put(outputSpec.getDestinationVertexName(), output). There is no separate "edge name" — the vertex names are the keys.

Recipe (a) — IDE breakpoint session (local mode)

Run any local-mode Tez program (the Lab 3.3 jar is ideal) under your debugger with tez.local.mode=true and tez.local.mode.without.network=true. Because local mode collapses the client, AM, and task into one process, you can watch the entire cascade in a single call stack. Set breakpoints in this exact order and step through:

When you hit breakpoint 8 on the union/middle vertex, inspect numInitedSourceVertices and sourceVertices.size() — you will see the vertex refuse to init until both sources have reported. That is the multi-input gate, live.

Local-mode caveat. In local mode the IPC hops (DAGClientAMProtocolBlockingPBServerImpl, TezTaskUmbilicalProtocol over the wire) are short-circuited, so those two frames will not appear. That is the trade-off: local mode gives you one clean stack at the cost of hiding the real network boundaries. To see those, you need a real (or MiniTez) cluster and remote debug — a Stretch Goal below.

Recipe (b) — Log-based trace

Every hop above emits a real log line. These are the actual strings from the source; grep the AM container's syslog/stdout (or your local-mode console) for them, in order.

# Confirm each string exists in the source that produces it:
grep -rn "Submitting dag to TezSession\|Submitting DAG to YARN" \
  tez-api/src/main/java/org/apache/tez/client/TezClient.java
grep -rn "Starting DAG submitted via RPC\|Running DAG:" \
  tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java
grep -rn "Starting to run new task attempt" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TezChild.java
grep -rn "AutoStartComplete\|Waiting for .* IOs to start" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java

On a real run, tail the AM log and watch them appear in that sequence. If a line is missing, you have found your failure boundary — e.g. Running DAG: present but no task-attempt line means the DAG initialized but never scheduled a task (look at the vertex/scheduler).

Complete the Trace Table

Fill this in from the code, not from this page. Each ? is a real method name.

Deliverables

• ~/tez-notes/trace-3.1.md with a file path noted for every hop in the table above.
• Both submission paths documented: inline IPC DAGPlan (session) vs tez-dag.pb local resource (non-session).
• The completed Trace Table with all ? methods filled from source.
• A note recording the observed values of numInitedSourceVertices / sourceVertices.size() at the middle vertex during your breakpoint session (Recipe a).
• The ordered list of real log strings you confirmed exist in the source (Recipe b).

Troubleshooting

Stretch Goals

1. The AsyncDispatcher queue. DAG_START goes through the AsyncDispatcher. Find where it is created and started in DAGAppMaster, and read what happens when the event queue backs up.

   grep -n "AsyncDispatcher\|dispatcher" tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java | head
   

2. The reverse path (TA_DONE). When TezChild finishes a task it reports up the umbilical. Trace the reverse chain: umbilical → AM posts a TaskAttemptEvent → TaskAttemptImpl → TaskImpl → VertexImpl → DAGImpl completion. Identify every class and event.
3. Non-session AM boot. Read DAGAppMaster.serviceStart and readDAGPlanFile. In non-session mode the AM reads tez-dag.pb from its working directory and calls startDAG itself — no client IPC. Confirm the branch (if (!isSession) { ... dagPlan = readDAGPlanFile(); ... }).
4. Remote debug the real IPC. Launch a MiniTez or single-node cluster, attach a remote debugger to the AM, and set the breakpoint on DAGClientAMProtocolBlockingPBServerImpl.submitDAG that local mode hid. Compare the call stack to your local-mode stack from Recipe (a).

Validation / Self-check

Answer without looking back at this lab:

1. In TezClient.submitDAG, which boolean decides between an IPC submission and a fresh YARN application submission, and which method name does each branch call?
2. Name the two distinct ways a DAGPlan can travel from client to AM, and the condition that selects between inline-IPC and filesystem transfer within session mode.
3. In startDAGExecution, DAG_INIT is dispatched one way and DAG_START another. Which is synchronous, which is asynchronous, and why does the ordering matter?
4. After DAGImpl.InitTransition runs, what state is each vertex in, and which event actually triggers the vertex-init cascade?
5. In VertexImpl.InitTransition, what exact condition must hold before a vertex with multiple upstream sources will initialize? Name the two fields compared.
6. What information does TezChild receive from its launch command versus from the AM over the umbilical? Which of the two carries the TaskSpec?
7. In LogicalIOProcessorRuntimeTask.run, the processor receives two maps. What is each map keyed by — and where in the code is that key chosen?

When you can answer all seven and your trace notes carry a file path for every hop, you have completed Lab 3.1. Continue to Lab 3.2: Understand the IPO Abstraction.