Tez Runtime Internals

This chapter is about the code that runs inside the container, not inside the AM. Its job is narrow and unglamorous: boot a JVM, accept tasks from the AM over an umbilical RPC, run each one to completion, report status, and — critically — loop to run the next task in the same JVM so container and JVM startup costs amortize across many tasks.

Three modules collaborate:

• tez-runtime-internals — process boot (TezChild), the per-attempt driver (TezTaskRunner2), the task orchestrator (LogicalIOProcessorRuntimeTask), the memory broker (MemoryDistributor), and the umbilical client (TaskReporter).
• tez-runtime-library — the concrete IO/processor implementations (shuffle, sort, KV), covered in shuffle-sort.md and ipo-abstractions.md.
• tez-api — the SPI users implement (AbstractLogicalInput, etc.).

After this chapter you can trace one task attempt from TezChild.main to processor.run, name the interruption model that lets the AM kill a running task, explain how memory is brokered across IOs, and describe how the umbilical heartbeat carries events in both directions with no separate event bus.

ls tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/
ls tez-runtime-internals/src/main/java/org/apache/tez/runtime/

The container process: TezChild

TezChild.main() is the JVM entry point for every Tez task container.

grep -n "public static void main\|public static TezChild newTezChild\|public ContainerExecutionResult run()\|getTask\|shouldDie\|cleanupOnTaskChanged" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TezChild.java

Boot sequence (paraphrased from TezChild.java):

1. main parses positional args: AM host, AM port, container id, application attempt number, and the JVM/PID identifiers, then builds a TezChild via newTezChild(...).
2. Read security tokens from $HADOOP_TOKEN_FILE_LOCATION; construct the TezTaskUmbilicalProtocol RPC proxy pointing at the AM's task-attempt listener.
3. Enter run(), an infinite loop: a. getTask(...) blocks (via a ContainerReporter future) until the AM hands over a ContainerTask. b. If containerTask.shouldDie(), return ExitStatus.SUCCESS and shut down. c. Otherwise cleanupOnTaskChanged(...), build a TezTaskRunner2 for the attempt, run it, and loop.

The loop is container reuse: same JVM, next task. The exit statuses are a real enum:

// tez-runtime-internals, TezChild.ContainerExecutionResult.ExitStatus
SUCCESS(0), EXECUTION_FAILURE(1), INTERRUPTED(2), ASKED_TO_DIE(3);

flowchart TD
  S[JVM start: main] --> P[parse args + tokens]
  P --> R[RPC connect to AM umbilical]
  R --> L{getTask}
  L -- "shouldDie()" --> X["exit SUCCESS(0)"]
  L -- "new ContainerTask" --> C[cleanupOnTaskChanged]
  C --> T["TezTaskRunner2.run()"]
  T --> L

Why container reuse needs this loop

Allocating a YARN container costs hundreds of milliseconds; starting a JVM costs seconds. Tez amortizes both by running many tasks in one TezChild process. The AM-side reuse decision (which container gets which next task) is in container-reuse.md; the JVM side is just this getTask loop.

The getTask handoff

getTask does not return immediately — it blocks, polling the AM until work is available, bounded by a max sleep. TezChild drives it through a ContainerReporter future so the poll can be interrupted:

grep -n "getTaskMaxSleepTime\|getTaskFuture\|mergeTaskSpecConfToConf\|handleNewTaskLocalResources" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TezChild.java

The ContainerTask the AM returns wraps the TaskSpec (see logical-physical.md), the task's credentials, and any additional local resources. On each new task TezChild merges the per-task configuration into a copy of the base configuration (mergeTaskSpecConfToConf), refreshes credentials and localized resources if they changed, then hands the spec to a fresh TezTaskRunner2. The base JVM configuration is never mutated in place — each task runs against its own merged copy, so a reused container cannot leak one task's config into the next.

Note: getTask and heartbeat are two of the three methods on the umbilical; the third, canCommit, is how a task that produces committable output (an MROutput writing to HDFS) asks the AM for permission to commit, ensuring only one attempt of a task commits when speculation is on.

ObjectRegistry: caching across reused tasks

Because the JVM persists, Tez offers a per-JVM cache scoped to vertex / DAG / session lifetimes. On each task change, TezChild invalidates the caches whose scope was left behind:

grep -n "cleanupOnTaskChanged\|clearCache\|lastVertexID\|ObjectLifeCycle" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TezChild.java

// tez-runtime-internals, TezChild.cleanupOnTaskChanged()
if (lastVertexID != null) {
  if (!lastVertexID.equals(newVertexID)) {
    objectRegistry.clearCache(ObjectRegistryImpl.ObjectLifeCycle.VERTEX);
  }
  if (!lastVertexID.getDAGID().equals(newVertexID.getDAGID())) {
    objectRegistry.clearCache(ObjectRegistryImpl.ObjectLifeCycle.DAG);
  }
}
lastVertexID = newVertexID;

The user-facing API (tez-api, org.apache.tez.runtime.api.ObjectRegistry) has cacheForVertex, cacheForDAG, cacheForSession, get, delete. A broadcast hash-join builds its hash table once with cacheForDAG and every subsequent task in the reused container reads it back — this is the single biggest win of container reuse for Hive.

Warning: Anything you stash in the session-scoped registry lives as long as the container. A leak there is a slow container OOM that survives task boundaries and is invisible in any single task's counters.

TezTaskRunner2 — the per-attempt driver and interruption model

grep -n "public TaskRunner2Result run()\|killTaskRequested\|firstEndReason\|taskKillStartTime\|TaskRunner2Callable\|EndReason" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TezTaskRunner2.java

TezTaskRunner2 owns one attempt. It submits the actual work to a TaskRunner2Callable on an executor and waits on the future, so the calling thread stays free to receive a kill. The key state:

// tez-runtime-internals, TezTaskRunner2 (fields)
private final AtomicBoolean killTaskRequested = new AtomicBoolean(false);
private volatile EndReason firstEndReason = null;
private volatile long taskKillStartTime = 0;
private volatile TaskRunner2Callable taskRunnerCallable;

The interruption model is cooperative. TaskRunner2Callable checks a stopRequested flag at each phase boundary and only proceeds if it is clear:

// tez-runtime-internals, TaskRunner2Callable.callInternal()  (abbreviated)
task.initialize();
if (!stopRequested.get() && !Thread.currentThread().isInterrupted()) {
  task.run();                    // processor.run(...)
} else { return new TaskRunner2CallableResult(null); }
if (!stopRequested.get() && !Thread.currentThread().isInterrupted()) {
  task.close();                  // flush outputs, emit events
} else { return new TaskRunner2CallableResult(null); }

A kill from the AM sets stopRequested and interrupts the worker thread. That is why processor code should be interrupt-aware: a KeyValuesReader loop that never checks Thread.interrupted() and never blocks on an interruptible call cannot be killed promptly, and the AM eventually declares the attempt lost. The possible outcomes are an EndReason (SUCCESS, TASK_ERROR, CONTAINER_STOP_REQUESTED, COMMUNICATION_FAILURE, KILL_REQUESTED, …), which run() maps to a TaskRunner2Result.

sequenceDiagram
  participant TC as TezChild
  participant TR as TezTaskRunner2
  participant TK as TaskRunner2Callable (worker thread)
  participant T as LogicalIOProcessorRuntimeTask
  TC->>TR: run()
  TR->>TK: submit to executor
  TK->>T: initialize()
  Note over TK: check stopRequested at each boundary
  TK->>T: run() => processor.run(inputs, outputs)
  TK->>T: close()
  TR-->>TC: TaskRunner2Result (EndReason)
  Note over TR,TK: AM kill -> stopRequested=true + interrupt

LogicalIOProcessorRuntimeTask — the orchestrator

This is the class that instantiates your IPO triple and enforces the lifecycle that ipo-abstractions.md depends on.

grep -n "public void initialize()\|public void run()\|public void close()\|InitializeInputCallable\|InitializeOutputCallable\|StartInputCallable\|makeInitialAllocations\|eventRouterThread\|numInitializers" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java

initialize() runs, in order:

1. Guard state == NEW, transition to INITED.
2. Create the ProcessorContext and the processor.
3. Submit one InitializeInputCallable per input and one InitializeOutputCallable per output to a thread pool sized numInputs + numOutputs; initialize the processor (order controlled by initializeProcessorFirst / initializeProcessorIOSerially).
4. Block on the completion service until all IO initialize() calls return.
5. initialMemoryDistributor.makeInitialAllocations() — the memory callbacks fire here (next section).
6. Auto-start each not-yet-started input via StartInputCallable.

// tez-runtime-internals, LogicalIOProcessorRuntimeTask.initialize()  (abbreviated)
int inputIndex = 0;
for (InputSpec inputSpec : taskSpec.getInputs()) {
  this.initializerCompletionService.submit(new InitializeInputCallable(inputSpec, inputIndex++));
}
for (OutputSpec outputSpec : taskSpec.getOutputs()) {
  this.initializerCompletionService.submit(new InitializeOutputCallable(outputSpec, outputIndex++));
}
// ... block until all complete ...
initialMemoryDistributor.makeInitialAllocations();

run() is trivially thin — it just drives the processor on the current thread:

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

close() closes inputs, then outputs, then the processor, collecting the events each close() returns and shipping them:

// tez-runtime-internals, LogicalIOProcessorRuntimeTask.close()  (abbreviated)
for (InputSpec inputSpec : inputSpecs) {
  closeInputEvents.add(inputsMap.get(srcVertexName).close());
}
for (OutputSpec outputSpec : outputSpecs) {
  closeOutputEvents.add(outputsMap.get(destVertexName).close());
}
processor.close();
// then sendTaskGeneratedEvents(...) for all collected input/output events

Note: the parallel IO init (the initializerCompletionService) is what makes Tez fast for processors with many inputs, e.g. multi-way joins — all inputs initialize concurrently, not one after another. There is no ordering guarantee among them; the framework only guarantees all are done before run(). See ipo-abstractions.md.

Inbound event delivery: the event-router thread

Inbound events arrive on heartbeat replies (next section), but they are not dispatched to inputs on the heartbeat thread — that would couple event delivery latency to RPC. Instead LogicalIOProcessorRuntimeTask runs a dedicated eventRouterThread that pulls from an internal queue and calls each IO's handleEvents(...):

grep -n "eventRouterThread\|handleEvents\|inputReadyTracker\|InputReadyTracker" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java

The router thread is why an input's handleEvents can run concurrently with the processor's run — a DataMovementEvent for a late upstream completion is delivered while the processor is already draining earlier inputs. close() interrupts and joins this thread so no event dispatch outlives the task:

// tez-runtime-internals, LogicalIOProcessorRuntimeTask.close()  (finally block)
if (eventRouterThread != null) {
  eventRouterThread.interrupt();
  eventRouterThread.join();     // no dispatch outlives the task
  eventRouterThread = null;
}

An InputReadyTracker coordinates the other direction: it lets the processor block on "any input ready" or "all inputs ready" so a processor with many inputs can begin work on whichever data lands first rather than waiting for the slowest upstream. This is the runtime primitive behind Tez's pipelined execution.

The memory broker: MemoryDistributor

Every buffer-heavy IO asks this broker for memory instead of allocating directly. It is the runtime side of the handshake described in ipo-abstractions.md.

find tez-runtime-internals/src/main/java -name "MemoryDistributor.java"
grep -n "requestInitialMemory\|makeInitialAllocations\|InitialMemoryAllocator\|reserveMemory" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/common/resources/MemoryDistributor.java

Flow:

1. During init, each IO calls context.requestInitialMemory(size, callback). The distributor records the (requestedSize, callback, componentType) tuple.
2. makeInitialAllocations() runs an InitialMemoryAllocator plugin (default WeightedScalingMemoryDistributor in tez-runtime-library) to scale the sum of requests down to what the container actually has.
3. Each callback receives its scaled budget via MemoryUpdateCallback.memoryAssigned(long).
4. IOs size their buffers from the assigned value.

The weighting is per-component-type; the plugin reserves a JVM fraction first, then distributes the rest by weight:

grep -n "TEZ_TASK_SCALE_MEMORY_WEIGHTED_RATIOS\|TEZ_TASK_SCALE_MEMORY_RESERVE_FRACTION" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/resources/WeightedScalingMemoryDistributor.java

Worked example. Container heap 1 GB, one OrderedPartitionedKVOutput requesting 512 MB and two OrderedGroupedKVInputs requesting 256 MB each — sum 1024 MB against a heap that must also reserve ~0.3 for the JVM. The distributor cannot honor the raw requests; it reserves the JVM fraction, then splits the remainder by the sorted-output vs sorted-merged-input weights. Every IO gets less than it asked for, and — because each sized its buffer from memoryAssigned, not the request — none of them OOMs the container. Disable scaling (tez.task.scale.memory.enabled=false) and the same three requests sum past the heap and the container dies at init.

Requesting memory after makeInitialAllocations() has run throws — the allocation phase is one-shot.

The umbilical and TaskReporter

Tez has no side-channel event bus between AM and containers. Everything — counters, status, inbound DataMovementEvents, kill signals — rides the umbilical heartbeat.

grep -n "getTask\|canCommit\|heartbeat\|versionID" \
  tez-runtime-internals/src/main/java/org/apache/tez/common/TezTaskUmbilicalProtocol.java

// tez-runtime-internals, org.apache.tez.common.TezTaskUmbilicalProtocol
ContainerTask         getTask(ContainerContext containerContext) throws IOException;
boolean               canCommit(TezTaskAttemptID taskid) throws IOException;
TezHeartbeatResponse  heartbeat(TezHeartbeatRequest request) throws IOException, TezException;

TaskReporter runs a HeartbeatCallable per attempt. Read the loop:

grep -n "class HeartbeatCallable\|response.shouldDie\|nonOobHeartbeatCounter\|pollInterval\|sendCounterInterval\|maxEventsToGet\|eventsToSend" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TaskReporter.java

// tez-runtime-internals, TaskReporter.HeartbeatCallable.call()  (abbreviated)
while (!task.isTaskDone() && !task.wasErrorReported()) {
  ResponseWrapper response = heartbeat(null);
  if (response.shouldDie) { return false; }             // AM asked us to die
  if (response.numEvents < maxEventsToGet) {
    condition.await(pollInterval, TimeUnit.MILLISECONDS); // else it was an OOB heartbeat
    nonOobHeartbeatCounter.incrementAndGet();
  }
}

Each heartbeat call drains the outbound eventsToSend queue, appends a status update (with counters at most every sendCounterInterval), and sends a TezHeartbeatRequest. The reply is a TezHeartbeatResponse:

// tez-runtime-internals, TezHeartbeatResponse (fields)
private long lastRequestId;
private boolean shouldDie = false;
private List<TezEvent> events;      // inbound events, e.g. DataMovementEvents
private int nextFromEventId;
private int nextPreRoutedEventId;

The events list on the response is the only channel by which an upstream task's completion reaches a downstream input — the AM routes the source's CompositeDataMovementEvent (via the edge manager, see logical-physical.md) and returns per-destination DataMovementEvents on the next heartbeat. nextFromEventId / nextPreRoutedEventId are the cursor the task advances so it never re-receives an event — a form of at-least-once delivery with client-side dedup.

Heartbeat batching keeps this cheap. Note the OOB logic above: if a response returned a full batch (numEvents == maxEventsToGet), the callable skips the poll wait and immediately heartbeats again to drain the backlog, otherwise it sleeps pollInterval.

grep -n "TEZ_TASK_AM_HEARTBEAT_INTERVAL_MS\|TEZ_TASK_AM_HEARTBEAT_COUNTER_INTERVAL_MS\|TEZ_TASK_MAX_EVENTS_PER_HEARTBEAT" \
  tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java

Warning: event latency is bounded below by tez.task.am.heartbeat.interval-ms. Raise it to reduce AM load and you delay every DataMovementEvent — downstream tasks start their fetches later. The counter interval is a separate, coarser knob so high-frequency counter churn does not force a counters payload on every 100 ms tick.

A wedged umbilical (network partition, dead TaskReporter thread) blocks all task communication; the AM's heartbeat-timeout check eventually fires and the attempt is declared lost. The protocol itself carries a versionID (19L in the current source) so a container built against a mismatched Tez version fails the RPC handshake loudly at getTask rather than corrupting a heartbeat later. See failure-handling.md for the AM side and counters-diagnostics.md for how counters ride these heartbeats.

End-to-end task lifecycle inside the JVM

Reading exercise

# Every termination path out of the container loop
grep -n "shouldDie\|ExitStatus\|return new ContainerExecutionResult" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TezChild.java

# The exact lifecycle order
grep -n "state.set(State\|processor.run\|\.close()" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/LogicalIOProcessorRuntimeTask.java

# How the distributor handles over-subscription
grep -n "makeInitialAllocations\|InitialMemoryAllocator" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/common/resources/MemoryDistributor.java

# The heartbeat loop body
grep -n "class HeartbeatCallable\|shouldDie\|nonOobHeartbeatCounter" \
  tez-runtime-internals/src/main/java/org/apache/tez/runtime/task/TaskReporter.java

# The user-facing processor contract
sed -n '30,80p' tez-api/src/main/java/org/apache/tez/runtime/api/AbstractLogicalIOProcessor.java

Answer:

1. List every value of ContainerExecutionResult.ExitStatus and the condition that produces each.
2. In two sentences, why does LogicalIOProcessorRuntimeTask.initialize parallelize IO init? Name the field.
3. What single field on TezHeartbeatResponse delivers inbound events to a task, and what advances the cursor so events are not redelivered?
4. Explain the OOB heartbeat: when does the callable skip its poll wait, and why is that the right behavior under an event backlog?
5. Trace, with file:method references, the path from TezChild.main to processor.run for one attempt.

Common bugs and symptoms

Validation: prove you understand this

1. Trace, with file:method references, the full path from TezChild.main to processor.run(...) for a single attempt, naming every class in between.
2. Explain in two sentences why IO init is parallel and processor run is not. Cite the completion-service field and the State guard on run().
3. Container heap 1 GB; one OrderedPartitionedKVOutput requests 512 MB and two OrderedGroupedKVInputs request 256 MB each. Describe qualitatively what the default WeightedScalingMemoryDistributor assigns and why no single IO gets its full request. Cross-check the weight keys against shuffle-sort.md.
4. Identify the single umbilical method that delivers inbound TezEvents to the task, and the field on the response object that carries them. Cite the file.
5. Sketch the smallest AbstractLogicalIOProcessor that prints the class names of all configured inputs from run(inputs, outputs) and exits — include initialize, handleEvents, run, close, and the required constructor.
6. Given container reuse, explain what cleanupOnTaskChanged clears when the next task belongs to (a) the same vertex, (b) a different vertex in the same DAG, (c) a different DAG. Cite the ObjectLifeCycle values.