Failure Handling

A Tez DAG fails for many reasons: a corrupted input split, a flaky NM, an OOM in the user processor, a Kerberos token expiry, an RM connectivity blip, an AM crash. Tez has a layered escalation model: small failures are absorbed, big ones propagate, and the AM persists enough state to recover from its own death.

This chapter walks the escalation, the failure taxonomy, and the recovery machinery.

Escalation: attempt → task → vertex → DAG

flowchart TD
  TA[TaskAttempt fails] -->|retry budget| TA2[New TaskAttempt]
  TA -->|exhausted| T[Task fails]
  T -->|failure policy| V[Vertex fails]
  V -->|fail-on-vertex-failure| D[DAG fails]

Default behavior:

grep -n "MAX_FAILED_ATTEMPTS\|MAX_TASK_ATTEMPTS\|TEZ_AM_TASK_MAX" \
  tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java

TaskAttemptTerminationCause

find tez-dag/src/main/java -name "TaskAttemptTerminationCause.java"
cat $(find tez-dag/src/main/java -name "TaskAttemptTerminationCause.java")

The enum names every reason a TaskAttempt can end up non-SUCCEEDED. A selection:

This enum is the most important debugging signal — every failed attempt in ATS / AM log surfaces a cause from this list.

TaskAttempt failure retries

grep -n "max.failed.attempts\|maxFailedAttempts\|attemptFailed" \
  $(find tez-dag/src/main/java -name "TaskImpl.java")

On a TA failure:

1. TaskAttemptImpl transitions to FAILED (or KILLED if the cause is in the "killed" subset).
2. TaskImpl increments its failed-attempt counter.
3. If counter < tez.am.task.max.failed.attempts, TaskImpl schedules a new TaskAttemptImpl (incremented attempt index).
4. Otherwise TaskImpl transitions to FAILED and reports up to VertexImpl.

Some causes are not counted against the budget (e.g. OUTPUT_LOST, NODE_FAILED) — these are infrastructure failures, not user-code failures.

grep -n "isFatalFailure\|isExternalError\|countAsFailure" \
  $(find tez-dag/src/main/java -name "*.java" | head -50)

Node blacklisting

grep -rn "NodeTracker\|blacklist\|BLACKLISTED" \
  tez-dag/src/main/java/org/apache/tez/dag/app | head

Per-node failure accounting:

A blacklisted node:

• No new container requests go to it.
• Held containers on it are released.
• Existing attempts already running on it are allowed to finish (not preemptively killed).

Output loss

A common late-stage failure: a downstream task reads from a shuffle source and finds the source's output is gone (the NM died, the disk was wiped, etc).

grep -rn "OUTPUT_LOST\|reportSourceTaskAttemptFailed\|inputFailedEvent" \
  tez-runtime-library/src/main/java tez-dag/src/main/java | head

Flow:

1. Destination TA's Fetcher fails permanently on source S.a.0.
2. Destination TA sends InputReadErrorEvent via umbilical heartbeat.
3. AM's VertexImpl receives the event, marks S.a.0 as OUTPUT_LOST.
4. TaskImpl for S.a schedules a new attempt S.a.1.
5. New attempt re-runs, produces fresh outputs, downstream resumes.

This is the cascading-rerun engine — and a source of pathological behavior when a single bad disk poisons many downstream tasks.

AM failover

find tez-dag/src/main/java -name "RecoveryService.java"
find tez-dag/src/main/java -name "RecoveryEventHandler.java"
wc -l $(find tez-dag/src/main/java -name "RecoveryService.java")

YARN keeps a small budget of AM restarts (yarn.resourcemanager.am.max-attempts, default 2). When the AM crashes:

1. RM allocates a fresh AM container, attempt index incremented.
2. New AM boots, sees attempt > 1, enters recovery mode.
3. RecoveryService reads the recovery log from HDFS (written by attempt 1).
4. Replays events to reconstruct DAG, Vertex, Task, TaskAttempt state.
5. Inherits any pre-existing containers via AMRMClient.getContainersFromPreviousAttempts.
6. Resumes scheduling from the last consistent state.

RecoveryService

grep -n "writeEvent\|flush\|recover\|RecoveryEventType" \
  $(find tez-dag/src/main/java -name "RecoveryService.java" \
                                -o -name "RecoveryEventHandler.java")

Append-only event log on HDFS, one file per app attempt:

hdfs:///tmp/staging/<user>/.staging/application_<id>/appattempt_<id>_NNNNNN/
  recovery/
    summary.dag_1.recovery
    dag_1.recovery

Event kinds:

find tez-dag/src/main/java -name "*Event*.java" -path "*recovery*"

Configuration

What recovery can and cannot recover

A VertexManagerPlugin that holds in-memory state across events must override getState() / setState() to participate in recovery — otherwise it starts fresh on AM attempt 2.

DAG-level termination causes

find tez-dag/src/main/java -name "DAGTerminationCause.java"
cat $(find tez-dag/src/main/java -name "DAGTerminationCause.java")

Reading exercise

1. grep -n "transition\|FAILED\|KILLED" \ $(find tez-dag/src/main/java -name "TaskAttemptImpl.java") | head -40 — count terminal transitions.
2. grep -rn "OUTPUT_LOST" tez-dag/src/main/java tez-runtime-library/src/main/java — what triggers this cause?
3. cat $(find tez-dag/src/main/java -name "RecoveryService.java") | head -200 — read the writer loop.
4. grep -n "RecoveryEvent" $(find tez-dag/src/main/java -name "*.java" | head -50) — list all recovery event classes.
5. wc -l $(find tez-dag/src/main/java -name "TaskAttemptTerminationCause.java" \ -o -name "DAGTerminationCause.java" \ -o -name "VertexTerminationCause.java")
6. grep -rn "node-blacklisting\|blacklistNode" tez-dag/src/main/java | head — where is blacklist enforcement implemented?

Common bugs and symptoms

Validation: prove you understand this

1. List five TaskAttemptTerminationCause values that do not count against the attempt budget. Cite where the predicate lives.
2. Explain in two sentences how an OUTPUT_LOST on source S.a.0 triggers a re-run of S.a, not just S.a.0's downstream consumers.
3. Identify the HDFS path pattern under which recovery events are written. Give the exact path components.
4. Describe what happens to in-flight DataMovementEvents when the AM crashes mid-DAG and AM attempt 2 takes over.
5. Given tez.am.maxtaskfailures.per.node=3 and an 8-node cluster, what is the smallest sequence of task failures that triggers blacklisting? Show the math.