TaskAttemptImpl Lifecycle

TaskAttemptImpl is the AM-side representation of a single execution attempt of a task. It is where the abstract DAG meets physical reality: it requests a container from the scheduler, tracks the umbilical heartbeat, records diagnostics, and — most consequentially — stamps the TaskAttemptTerminationCause that tells TaskImpl whether this attempt's death counts against the task's failure budget (task-lifecycle.md). It has around 48 transitions, the most of any DAG-execution machine after VertexImpl.

Read state-machines.md and task-lifecycle.md first. This chapter closes the loop: TaskImpl decides policy (retry, winner, budget); TaskAttemptImpl produces the facts that policy consumes.

After this chapter you can look at any attempt state in an AM log and explain which container holds it, which umbilical calls have landed, and what its termination cause will be if it dies right now.

The file, the tests, and the two state enums

tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestTaskAttempt.java

The internal state machine state — quote it exactly:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/TaskAttemptStateInternal.java
public enum TaskAttemptStateInternal {
  NEW,
  START_WAIT,
  SUBMITTED,
  RUNNING,
  KILL_IN_PROGRESS,
  FAIL_IN_PROGRESS,
  KILLED,
  FAILED,
  SUCCEEDED
}

Nine internal states. As with tasks, there is a coarser external TaskAttemptState (what ATS/history and clients see), and the machine maps between them:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
private static TaskAttemptState getExternalState(TaskAttemptStateInternal smState) {
  switch (smState) {
  case NEW:
  case START_WAIT:
  case SUBMITTED:
    return TaskAttemptState.STARTING;
  case RUNNING:
    return TaskAttemptState.RUNNING;
  case FAILED:
  case FAIL_IN_PROGRESS:
    return TaskAttemptState.FAILED;
  case KILLED:
  case KILL_IN_PROGRESS:
    return TaskAttemptState.KILLED;
  case SUCCEEDED:
    return TaskAttemptState.SUCCEEDED;
  default:
    throw new TezUncheckedException(/* ... */);
  }
}

The *_IN_PROGRESS states exist for the same reason the task has KILL_WAIT and the vertex has TERMINATING: an attempt cannot become terminal while its container is still alive. It marks intent, waits for the container-terminated event, then finalizes.

The transitions and the handle() pattern

grep -n "Transitions from\|addTransition" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java | head -50

The machine opens from NEW, and the very first arc is another recovery-aware multiple-arc:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
= new StateMachineFactory
    <TaskAttemptImpl, TaskAttemptStateInternal, TaskAttemptEventType, TaskAttemptEvent>
    (TaskAttemptStateInternal.NEW)

  .addTransition(TaskAttemptStateInternal.NEW,
      EnumSet.of(TaskAttemptStateInternal.NEW, TaskAttemptStateInternal.START_WAIT,
          TaskAttemptStateInternal.FAILED),
      TaskAttemptEventType.TA_SCHEDULE, new ScheduleTaskattemptTransition())
  // NEW -> FAILED due to TA_FAILED happens in recovery
  .addTransition(TaskAttemptStateInternal.NEW, TaskAttemptStateInternal.FAILED,
      TaskAttemptEventType.TA_FAILED, new TerminateTransition(FAILED_HELPER))
  // NEW -> KILLED / SUCCEEDED also possible in recovery ...

Unlike VertexImpl/TaskImpl, TaskAttemptImpl uses the raw Hadoop machine (no StateMachineTez wrapper — nothing subscribes to attempt state entry), and its handle() routes invalid/uncaught events to the DAG as an internal error:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
public void handle(TaskAttemptEvent event) {
  writeLock.lock();
  try {
    final TaskAttemptStateInternal oldState = getInternalState();
    try {
      stateMachine.doTransition(event.getType(), event);
    } catch (InvalidStateTransitonException e) {
      LOG.error("Can't handle this event at current state for " + this.attemptId, e);
      eventHandler.handle(new DAGEventDiagnosticsUpdate(getDAGID(),
          "Invalid event " + event.getType() + " on TaskAttempt " + this.attemptId));
      eventHandler.handle(new DAGEvent(getDAGID(), DAGEventType.INTERNAL_ERROR));
    } catch (RuntimeException e) {
      // ... same: diagnostics update + DAGEventType.INTERNAL_ERROR
    }
    // ...
  } finally {
    writeLock.unlock();
  }
}

Container assignment

An attempt is not born with a container — that is the entire reason START_WAIT exists. The ScheduleTaskattemptTransition hook (past its recovery preamble) builds a launch request from the task's location hints and hands it to the scheduler as an AMSchedulerEventTALaunchRequest:

grep -n "AMSchedulerEventTALaunchRequest\|allocateTask\|AMSchedulerEvent" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java | head

The flow, end to end:

sequenceDiagram
    participant T as TaskImpl
    participant TA as TaskAttemptImpl
    participant SCH as TaskSchedulerManager
    participant CL as ContainerLauncherManager
    participant TCM as TaskCommunicatorManager (umbilical)

    T->>TA: TA_SCHEDULE
    TA->>TA: NEW -> START_WAIT
    TA->>SCH: AMSchedulerEventTALaunchRequest
    Note over SCH: match a granted/reused container
    SCH->>TA: TA_SUBMITTED
    TA->>TA: START_WAIT -> SUBMITTED
    CL->>TA: TA_STARTED_REMOTELY
    TA->>TA: SUBMITTED -> RUNNING
    loop while running
        TCM->>TA: TA_STATUS_UPDATE (heartbeat)
    end
    TCM->>TA: TA_DONE
    TA->>TA: RUNNING -> SUCCEEDED
    TA->>T: T_ATTEMPT_SUCCEEDED

The scheduler side (TaskSchedulerManager, YarnTaskSchedulerService) and container reuse are covered in scheduler.md and container-reuse.md. What matters here: with reuse enabled, START_WAIT can be nearly instantaneous because the scheduler hands back an already-running idle container; without it, the attempt waits for YARN to grant a fresh one, which is where "stuck in START_WAIT" symptoms come from.

SUBMITTED vs RUNNING: two distinct milestones

The two-step START_WAIT → SUBMITTED → RUNNING sequence is not redundant. TA_SUBMITTED fires when a container has been chosen and the launch request handed off; that is where the attempt first binds to a concrete container and records where it lives:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java (SubmittedTransition)
AMContainer amContainer = ta.appContext.getAllContainers().get(event.getContainerId());
TezContainer container = new TezContainer(amContainer.getContainer());
ta.allocationTime = amContainer.getCurrentTaskAttemptAllocationTime();
ta.container = new TezContainer(container);
ta.setLaunchTime();
// resolve NM http address for the UI, record trackerName / httpPort
ta.sendEvent(createDAGCounterUpdateEventTALaunched(ta));

TA_STARTED_REMOTELY then fires when the task JVM has actually started executing the processor and the umbilical is live; StartedTransition merely notifies the scheduler that the attempt is now STARTED:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java (StartedTransition)
ta.sendEvent(new AMSchedulerEventTAStateUpdated(ta,
    TaskScheduler.SchedulerTaskState.STARTED, ta.getVertex().getTaskSchedulerIdentifier()));

The gap between the two states is where "container launched but processor never came up" failures live — an attempt stuck in SUBMITTED means the JVM was launched but never heartbeated, distinct from START_WAIT (no container yet) and RUNNING (executing). When you read a hung attempt in a log, the exact internal state tells you which of the three provisioning steps stalled.

The umbilical heartbeat

Once running, the task JVM talks to the AM over the umbilical protocol (TezTaskUmbilicalProtocol, served by TaskCommunicatorManager and the per-plugin TezTaskCommunicatorImpl). Each heartbeat carries runtime TezEvents that TaskCommunicatorManager translates into TaskAttemptEvents. The AM tracks liveness with TaskHeartbeatHandler (a HeartbeatHandlerBase<TezTaskAttemptID>); if an attempt misses heartbeats past the timeout, the handler synthesizes a failure:

// tez-dag/src/main/java/org/apache/tez/dag/app/TaskHeartbeatHandler.java
@Override
protected void handleTimeOut(TezTaskAttemptID attemptId) {
  eventHandler.handle(new TaskAttemptEventAttemptFailed(attemptId,
      TaskAttemptEventType.TA_TIMED_OUT, TaskFailureType.NON_FATAL, "AttemptID:" + attemptId.toString()
      + " Timed out after " + timeOut / 1000 + " secs", TaskAttemptTerminationCause.TASK_HEARTBEAT_ERROR));
}

So a heartbeat timeout arrives at the attempt as TA_TIMED_OUT, carrying TaskFailureType.NON_FATAL and cause TASK_HEARTBEAT_ERROR — a failure that counts against the budget. The timeout is:

// tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java
public static final String TASK_HEARTBEAT_TIMEOUT_MS = TEZ_TASK_PREFIX + "timeout-ms";
public static final int TASK_HEARTBEAT_TIMEOUT_MS_DEFAULT = 5 * 60 * 1000;  // 300000
public static final String TASK_HEARTBEAT_TIMEOUT_CHECK_MS = TEZ_TASK_PREFIX + "heartbeat.timeout.check-ms";
public static final int TASK_HEARTBEAT_TIMEOUT_CHECK_MS_DEFAULT = 30 * 1000;

Warning: A TA_TIMED_OUT does not always mean the task hung. An AM GC pause, clock skew between AM and NM, or a saturated dispatcher (event-routing.md) can all delay heartbeat processing and trip the timeout on a task that was actually fine. Before blaming the task, check the AM's own health.

Distinguish the failure events:

The termination cause: the policy fact

Every terminal transition stamps a TaskAttemptTerminationCause. This is the single field TaskImpl reads to decide budget accounting. Quote the full enum — do not paraphrase, because the exact members matter:

// tez-common/src/main/java/org/apache/tez/dag/records/TaskAttemptTerminationCause.java
public enum TaskAttemptTerminationCause {
  UNKNOWN_ERROR, // The error cause is unknown. Usually means a gap in error propagation

  TERMINATED_BY_CLIENT, // Killed by client command
  TERMINATED_AT_SHUTDOWN, // Killed due execution shutdown
  TERMINATED_AT_RECOVERY, // Killed in recovery, due to can not recover running task attempt
  INTERNAL_PREEMPTION, // Killed by Tez to makes space for higher pri work
  EXTERNAL_PREEMPTION, // Killed by the cluster to make space for other work
  TERMINATED_INEFFECTIVE_SPECULATION, // Killed speculative attempt because original succeeded
  TERMINATED_EFFECTIVE_SPECULATION, // Killed original attempt because speculation succeeded
  TERMINATED_ORPHANED, // Attempt is no longer needed by the task

  APPLICATION_ERROR, // Failed due to application code error
  FRAMEWORK_ERROR, // Failed due to code error in Tez code
  INPUT_READ_ERROR, // Failed due to error in reading inputs
  OUTPUT_WRITE_ERROR, // Failed due to error in writing outputs
  OUTPUT_LOST, // Failed because attempts output were reported lost
  NO_PROGRESS, // Failed because no progress was being made
  TASK_HEARTBEAT_ERROR, // Failed because AM lost connection to the task

  CONTAINER_LAUNCH_FAILED, // Failed to launch container
  CONTAINER_EXITED, // Container exited. Indicates gap in specific error propagation from the cluster
  CONTAINER_STOPPED, // Container stopped or released by Tez
  NODE_FAILED, // Node for the container failed
  NODE_DISK_ERROR, // Disk failed on the node running the task

  COMMUNICATION_ERROR, // Equivalent to a launch failure
  SERVICE_BUSY, // Service rejected the task
  INTERRUPTED_BY_SYSTEM, // Interrupted by the system. e.g. Pre-emption
  INTERRUPTED_BY_USER, // Interrupted by the user

}

The line that separates the two blank-line groups is meaningful: the first group (through TERMINATED_ORPHANED) are kills — the attempt was terminated for reasons that are not its fault, and they arrive at TaskImpl as T_ATTEMPT_KILLED (no budget cost). The second and third groups are failures that reach TaskImpl as T_ATTEMPT_FAILED and cost a budget slot. The mapping from cause to event is done by the terminating transitions (TerminateTransition and helpers) via the TaskFailureType and the specific TaskAttemptEvent subclass constructed. Cross-check the budget consequences in task-lifecycle.md.

Output-failure: re-running a source attempt

The subtlest flow in the whole AM is what happens when a downstream consumer cannot fetch a completed upstream attempt's output. This is how Tez recovers from lost intermediate data (a dead node, a wiped local disk) without failing the whole DAG.

When a consumer task hits a fetch failure, it emits a runtime InputReadErrorEvent. The AM's Edge routes it back to the source attempt that produced the data, as a TaskAttemptEventOutputFailed:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java
TezTaskAttemptID srcTaskAttemptId = TezTaskAttemptID.getInstance(srcTaskId, srcTaskAttemptIndex);
sendEvent(new TaskAttemptEventOutputFailed(srcTaskAttemptId, tezEvent, numConsumers));

That lands on the source attempt (which may be in SUCCEEDED!) at OutputReportedFailedTransition, a multiple-arc that decides whether this report is serious enough to re-run the source. It does not re-run on the first complaint — it tracks how many distinct consumers blamed this attempt, over what time window, against configurable thresholds:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java (OutputReportedFailedTransition)
int maxAllowedOutputFailures = sourceAttempt.getVertex().getVertexConfig()
    .getMaxAllowedOutputFailures();          // tez.task.max.allowed.output.failures, default 10
int maxAllowedTimeForTaskReadErrorSec = sourceAttempt.getVertex()
    .getVertexConfig().getMaxAllowedTimeForTaskReadErrorSec();  // default 300
double maxAllowedOutputFailuresFraction = sourceAttempt.getVertex()
    .getVertexConfig().getMaxAllowedOutputFailuresFraction();   // default 0.1
// ...
boolean withinOutputFailureLimits =
    (sourceAttempt.uniquefailedOutputReports.size() < maxAllowedOutputFailures);
// ... if within all limits and not a local/disk error and no host-blame threshold crossed:
if (!crossTimeDeadline && withinFailureFractionLimits && withinOutputFailureLimits
    && !(readErrorEvent.isLocalFetch() || readErrorEvent.isDiskErrorAtSource())
    && !tooManyDownstreamHostsBlamedTheSameUpstreamHost) {
  return sourceAttempt.getInternalState();   // ignore this report for now
}
// otherwise: fail (and thus re-run) the source attempt
sourceAttempt.sendInputFailedToConsumers();
if (sourceAttempt.getInternalState() == TaskAttemptStateInternal.SUCCEEDED) {
  (new TerminatedAfterSuccessHelper(FAILED_HELPER)).transition(sourceAttempt, event);
  return TaskAttemptStateInternal.FAILED;
} else {
  (new TerminatedWhileRunningTransition(FAILED_HELPER)).transition(sourceAttempt, event);
  return TaskAttemptStateInternal.FAIL_IN_PROGRESS;
}

The governing configs, verified:

When the source attempt is re-run, its output is regenerated, and the earlier SUCCEEDED → SCHEDULED/FAILED retroactive arc in TaskImpl (task-lifecycle.md) is exactly what receives the resulting T_ATTEMPT_FAILED. This is the mechanism behind "a map task that finished an hour ago suddenly re-runs" — a reducer somewhere couldn't fetch its shuffle data. See shuffle-sort.md for the fetch side.

Output commit, per attempt

The attempt is one half of the commit protocol (TaskImpl.canCommit is the other — task-lifecycle.md). Whether a task-level committer runs inside the task JVM or the AM commits at vertex/DAG granularity is set by the same flag that governs vertex-lifecycle.md:

// tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java
public static final String TEZ_AM_COMMIT_ALL_OUTPUTS_ON_DAG_SUCCESS =
    TEZ_AM_PREFIX + "commit-all-outputs-on-dag-success";
public static final boolean TEZ_AM_COMMIT_ALL_OUTPUTS_ON_DAG_SUCCESS_DEFAULT = true;

Under the default (true), individual attempts never commit — the DAG commits after all vertices succeed, so a losing speculative attempt physically cannot publish output. Only when the flag is false does a task attempt run its own commit(), and then exactly one attempt per task is granted permission through TaskImpl.canCommit, which hands the commit slot to the first RUNNING attempt that asks and refuses everyone else. That single-writer guarantee is what stops two speculative copies of the same task from both writing the final output — the data-corruption bug that the commit protocol exists to prevent.

Node blacklisting

Repeated failures on the same node get that node blacklisted so the scheduler stops placing attempts there. AMNodeTracker / AMNodeImpl own an AMNodeState machine (ACTIVE, FORCED_ACTIVE, BLACKLISTED, UNHEALTHY). The relevant knobs:

// tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java
public static final int TEZ_AM_MAX_TASK_FAILURES_PER_NODE_DEFAULT = 10;
public static final boolean TEZ_AM_NODE_BLACKLISTING_ENABLED_DEFAULT = true;   // node-blacklisting.enabled
public static final int TEZ_AM_NODE_BLACKLISTING_IGNORE_THRESHOLD_DEFAULT = 33; // ignore-threshold-node-percent

The ignore-threshold (33% by default) caps blacklisting: if blacklisting more than a third of the cluster's nodes, Tez ignores blacklisting (moves nodes to FORCED_ACTIVE) rather than starve itself of capacity. This is the N_IGNORE_BLACKLISTING_ENABLED event in AMNodeEventType.

Diagnostics accumulation

Every failure, kill, and notable transition appends to the attempt's diagnostics list, which is what surfaces in the UI and in the DAG's final diagnostics string:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
private final List<String> diagnostics = new ArrayList<String>();
// ...
private void addDiagnosticInfo(String diag) { /* appends under lock */ }

When you debug a failed DAG, the attempt-level diagnostics are the leaves of the tree: task diagnostics quote attempt diagnostics, vertex diagnostics quote task diagnostics, DAG diagnostics quote vertex diagnostics. Read from the leaf up.

Reading exercise

cd /path/to/tez
# 1. The full machine
sed -n '/= new StateMachineFactory/,/installTopology()/p' \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java | less

# 2. The termination cause enum in full
sed -n '/public enum TaskAttemptTerminationCause/,/^}/p' \
  tez-common/src/main/java/org/apache/tez/dag/records/TaskAttemptTerminationCause.java

# 3. The heartbeat timeout path
grep -n "TA_TIMED_OUT\|handleTimeOut\|TASK_HEARTBEAT_ERROR" \
  tez-dag/src/main/java/org/apache/tez/dag/app/TaskHeartbeatHandler.java

# 4. The output-failure re-run path
grep -n "OutputReportedFailedTransition\|uniquefailedOutputReports\|maxAllowedOutputFailures" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
grep -n "TaskAttemptEventOutputFailed" tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/Edge.java

Then answer:

1. Given an attempt in RUNNING and a TA_CONTAINER_TERMINATED_BY_SYSTEM (preemption), what is the next internal state and does the failure counter increment? Trace the cause to the event TaskImpl receives.
2. What is the difference between TA_FAILED and TA_TIMED_OUT? Which class issues each and with what TaskAttemptTerminationCause?
3. From the enum, which causes reach TaskImpl as T_ATTEMPT_KILLED? Why is TASK_HEARTBEAT_ERROR not among them?
4. In OutputReportedFailedTransition, name the four thresholds that gate a source re-run and give the default of each.
5. Why do FAIL_IN_PROGRESS and KILL_IN_PROGRESS exist instead of jumping straight to FAILED/KILLED?
6. Walk the path from a reducer's InputReadErrorEvent to the source map attempt transitioning to FAILED, naming every class the event passes through.

Common bugs and symptoms

Validation: prove you understand this

1. List all nine TaskAttemptStateInternal values and, for each, the external TaskAttemptState it maps to. Verify against getExternalState.
2. For every TaskAttemptTerminationCause, tag it "kill (no budget)" or "failure (budget)" and confirm against the terminating transitions.
3. On MiniTezCluster, suspend a task JVM (kill -STOP) and find the exact log line where TaskHeartbeatHandler issues TA_TIMED_OUT; note the elapsed time and compare with tez.task.timeout-ms.
4. Trace TaskCommunicatorManager heartbeat handling from an incoming TASK_ATTEMPT_COMPLETED_EVENT to the TA_DONE the attempt receives.
5. Read OutputReportedFailedTransition and prove that a speculative-loser attempt cannot corrupt the failure counter — follow the cause it is killed with and the event TaskImpl gets.