TaskImpl Lifecycle
TaskImpl is the AM-side representation of one logical task within a vertex.
It is the smallest of the DAG-execution state machines — around 26 transitions
across seven internal states — but it owns one critical piece of policy:
which attempt of this task is the winner, and when repeated attempt
failures escalate to failing the whole task (and therefore the vertex, and
therefore the DAG). This chapter walks the real states, the failed-vs-killed
accounting that decides whether a failure counts against the budget, the
speculation hook, and winner selection.
Read state-machines.md and task-attempt-lifecycle.md alongside this — the task delegates all container/execution mechanics to its attempts and only tracks outcomes.
After this chapter you can explain why a task with three failed attempts may
still be RUNNING while another with a single failure is already FAILED, and
you can cite the exact code that makes that decision.
The file, the tests, and the internal/external state split
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
tez-dag/src/test/java/org/apache/tez/dag/app/dag/impl/TestTaskImpl.java
Tez maintains two task-state enums, and confusing them is a classic mistake. The internal state machine state:
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/TaskStateInternal.java
public enum TaskStateInternal {
NEW, SCHEDULED, RUNNING, SUCCEEDED, FAILED, KILL_WAIT, KILLED
}
The external state (TaskState in the old-records API, what the DAG,
history logging, and clients see) does not have KILL_WAIT. The machine maps
between them:
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
private static TaskState getExternalState(TaskStateInternal smState) {
if (smState == TaskStateInternal.KILL_WAIT) {
return TaskState.KILLED;
} else {
return TaskState.valueOf(smState.name());
}
}
So KILL_WAIT is an internal-only draining state — the task is being killed
and is waiting for its outstanding attempts to acknowledge, but to the outside
world it already reads as KILLED. This is the task-level analogue of the
vertex's TERMINATING.
| Internal state | External | Meaning |
|---|---|---|
NEW | NEW | Constructed; no attempts yet. |
SCHEDULED | SCHEDULED | First attempt requested; none running yet. |
RUNNING | RUNNING | At least one attempt has launched. |
KILL_WAIT | KILLED | Kill issued; draining outstanding attempts. |
SUCCEEDED | SUCCEEDED | Terminal: an attempt succeeded and was chosen winner. |
FAILED | FAILED | Terminal: failure budget exhausted or a fatal failure. |
KILLED | KILLED | Terminal: killed and drained. |
TaskImpl has no INITIALIZING or TERMINATING of its own — those are the
vertex's concerns.
The transitions
grep -n "Transitions from\|addTransition" \
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java | head -40
The NEW group shows the recovery-aware multiple-arc that opens the machine:
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
// Transitions from NEW state
// Stay in NEW in recovery when Task is killed in the previous AM
.addTransition(TaskStateInternal.NEW,
EnumSet.of(TaskStateInternal.NEW, TaskStateInternal.SCHEDULED),
TaskEventType.T_SCHEDULE, new InitialScheduleTransition())
.addTransition(TaskStateInternal.NEW, TaskStateInternal.KILLED,
TaskEventType.T_TERMINATE,
new KillNewTransition())
The RUNNING group is where the interesting policy lives:
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
.addTransition(TaskStateInternal.RUNNING, TaskStateInternal.RUNNING,
TaskEventType.T_ADD_SPEC_ATTEMPT, new RedundantScheduleTransition())
.addTransition(TaskStateInternal.RUNNING,
EnumSet.of(TaskStateInternal.SUCCEEDED),
TaskEventType.T_ATTEMPT_SUCCEEDED,
new AttemptSucceededTransition())
.addTransition(TaskStateInternal.RUNNING, EnumSet.of(TaskStateInternal.RUNNING, TaskStateInternal.FAILED),
TaskEventType.T_ATTEMPT_KILLED,
ATTEMPT_KILLED_TRANSITION)
.addTransition(TaskStateInternal.RUNNING,
EnumSet.of(TaskStateInternal.RUNNING, TaskStateInternal.FAILED),
TaskEventType.T_ATTEMPT_FAILED,
new AttemptFailedTransition())
.addTransition(TaskStateInternal.RUNNING, TaskStateInternal.KILL_WAIT,
TaskEventType.T_TERMINATE,
KILL_TRANSITION)
Note the two arcs that can go RUNNING → FAILED: both T_ATTEMPT_FAILED and
T_ATTEMPT_KILLED may fail the task, because if killing/failing an attempt
leaves no path to schedule a replacement, the task cannot make progress. There
is also a subtle SUCCEEDED-state arc — a retroactive failure. A map task
that already succeeded can be failed after the fact if a downstream consumer
reports its output was lost:
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
// Transitions from SUCCEEDED state
.addTransition(TaskStateInternal.SUCCEEDED, //only possible for map tasks
EnumSet.of(TaskStateInternal.SCHEDULED, TaskStateInternal.SUCCEEDED, TaskStateInternal.FAILED),
TaskEventType.T_ATTEMPT_FAILED, new TaskRetroactiveFailureTransition())
That is the AM side of the fetch-failure / output-lost story detailed in task-attempt-lifecycle.md.
stateDiagram-v2
[*] --> NEW
NEW --> SCHEDULED: T_SCHEDULE
NEW --> KILLED: T_TERMINATE
SCHEDULED --> RUNNING: T_ATTEMPT_LAUNCHED
SCHEDULED --> KILL_WAIT: T_TERMINATE
RUNNING --> RUNNING: T_ATTEMPT_FAILED (budget left)
RUNNING --> FAILED: T_ATTEMPT_FAILED (budget exhausted / FATAL)
RUNNING --> SUCCEEDED: T_ATTEMPT_SUCCEEDED
RUNNING --> KILL_WAIT: T_TERMINATE
RUNNING --> RUNNING: T_ADD_SPEC_ATTEMPT
KILL_WAIT --> KILLED: last attempt drained
SUCCEEDED --> SCHEDULED: T_ATTEMPT_FAILED (output lost, rerun)
SUCCEEDED --> FAILED: T_ATTEMPT_FAILED (retroactive, budget exhausted)
SUCCEEDED --> [*]
FAILED --> [*]
KILLED --> [*]
The failure budget: AttemptFailedTransition
This is the single most important piece of logic in the file. A task does not fail on the first attempt failure; it retries up to a budget. The budget and the decision:
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
public TaskStateInternal transition(TaskImpl task, TaskEvent event) {
task.failedAttempts++;
task.getVertex().incrementFailedTaskAttemptCount();
TaskEventTAFailed castEvent = (TaskEventTAFailed) event;
// ...
if (task.failedAttempts < task.maxFailedAttempts &&
castEvent.getTaskFailureType() == TaskFailureType.NON_FATAL) {
task.handleTaskAttemptCompletion(
((TaskEventTAUpdate) event).getTaskAttemptID(),
TaskAttemptStateInternal.FAILED);
if (task.shouldScheduleNewAttempt()) {
LOG.info("Scheduling new attempt for task: " + task.getTaskID()
+ ", currentFailedAttempts: " + task.failedAttempts + ", maxFailedAttempts: "
+ task.maxFailedAttempts + ", maxAttempts: " + task.maxAttempts);
if (!task.addAndScheduleAttempt(getSchedulingCausalTA())){
return task.finished(TaskStateInternal.FAILED);
}
}
} else {
// too many failures, or a FATAL failure -> fail the task
task.handleTaskAttemptCompletion(/* ... */ TaskAttemptStateInternal.FAILED);
task.logJobHistoryTaskFailedEvent(TaskState.FAILED);
task.eventHandler.handle(
new VertexEventTaskCompleted(task.taskId, TaskState.FAILED));
return task.finished(TaskStateInternal.FAILED);
}
return getDefaultState(task);
}
Two things gate escalation to FAILED:
-
failedAttempts >= maxFailedAttempts— the budget is exhausted. -
TaskFailureType.FATAL— a fatal failure fails the task immediately regardless of budget.TaskFailureTypehas exactly two values:// tez-api/src/main/java/org/apache/tez/runtime/api/TaskFailureType.java public enum TaskFailureType { NON_FATAL, // may recover on another attempt FATAL, // no more attempts }
The two budget configs
There are two distinct limits, and they mean different things:
// tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java
public static final String TEZ_AM_TASK_MAX_FAILED_ATTEMPTS =
TEZ_AM_PREFIX + "task.max.failed.attempts";
public static final int TEZ_AM_TASK_MAX_FAILED_ATTEMPTS_DEFAULT = 4;
public static final String TEZ_AM_TASK_MAX_ATTEMPTS = TEZ_AM_PREFIX + "task.max.attempts";
public static final int TEZ_AM_TASK_MAX_ATTEMPTS_DEFAULT = 0;
| Config | Default | Counts | Meaning |
|---|---|---|---|
tez.am.task.max.failed.attempts | 4 | failed attempts only | Fail the task once this many attempts have failed. |
tez.am.task.max.attempts | 0 (disabled) | every attempt (failed, killed, all) | Hard cap on total attempts regardless of outcome; 0 disables it. |
VertexImpl.VertexConfigImpl reads both and hands them to each TaskImpl:
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
this.maxFailedTaskAttempts = conf.getInt(TezConfiguration.TEZ_AM_TASK_MAX_FAILED_ATTEMPTS,
TezConfiguration.TEZ_AM_TASK_MAX_FAILED_ATTEMPTS_DEFAULT);
this.maxTaskAttempts = conf.getInt(TezConfiguration.TEZ_AM_TASK_MAX_ATTEMPTS,
TezConfiguration.TEZ_AM_TASK_MAX_ATTEMPTS_DEFAULT);
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
maxFailedAttempts = vertex.getVertexConfig().getMaxFailedTaskAttempts();
maxAttempts = vertex.getVertexConfig().getMaxTaskAttempts();
Killed attempts do not consume the budget
The decisive detail: failedAttempts++ only happens in AttemptFailedTransition
(on T_ATTEMPT_FAILED). A killed attempt goes through
AttemptKilledTransition, which increments a different counter and, if
possible, reschedules — but does not touch failedAttempts:
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java (AttemptKilledTransition)
if (isRejection) { // TODO: remove as part of TEZ-3881.
task.getVertex().incrementRejectedTaskAttemptCount();
} else {
task.getVertex().incrementKilledTaskAttemptCount();
}
if (task.shouldScheduleNewAttempt()) {
if (!task.addAndScheduleAttempt(castEvent.getTaskAttemptID())) {
return task.finished(TaskStateInternal.FAILED);
}
}
return originalState;
This is the answer to the chapter's opening question. Whether a terminated
attempt counts against the task budget is decided entirely by whether it
arrives as T_ATTEMPT_FAILED or T_ATTEMPT_KILLED — and that classification is
owned upstream by the TaskAttemptTerminationCause
(task-attempt-lifecycle.md). A speculative loser, a
preemption, or a vertex-kill all arrive as killed and are free; a processor
exception or a container crash arrive as failed and cost a budget slot.
| Attempt outcome | Event to task | Consumes max.failed.attempts? |
|---|---|---|
Processor threw (NON_FATAL) | T_ATTEMPT_FAILED | Yes |
Processor threw (FATAL) | T_ATTEMPT_FAILED | Fails task immediately |
| Container crashed | T_ATTEMPT_FAILED | Yes |
| Lost speculation race | T_ATTEMPT_KILLED | No |
| Killed by vertex termination | T_ATTEMPT_KILLED | No |
| Preempted (internal/external) | T_ATTEMPT_KILLED | No |
Speculation
Speculation launches a second copy of a slow-running task and races it against
the first. On master the implementation is the legacy speculator (there is no
SimpleSpeculator class — verify with the grep below):
ls tez-dag/src/main/java/org/apache/tez/dag/app/dag/speculation/legacy/
DataStatistics.java LegacySpeculator.java LegacyTaskRuntimeEstimator.java
SimpleExponentialTaskRuntimeEstimator.java StartEndTimesBase.java
TaskRuntimeEstimator.java forecast/
LegacySpeculator is an AbstractService owned per-vertex. It estimates task
runtimes (via a pluggable TaskRuntimeEstimator — LegacyTaskRuntimeEstimator
or SimpleExponentialTaskRuntimeEstimator) and, when a task looks like an
outlier, requests a redundant attempt. The path is indirect and worth tracing:
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/speculation/legacy/LegacySpeculator.java
protected void addSpeculativeAttempt(TezTaskID taskID) {
LOG.info("DefaultSpeculator.addSpeculativeAttempt -- we are speculating " + taskID);
vertex.scheduleSpeculativeTask(taskID);
mayHaveSpeculated.add(taskID);
}
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
public void scheduleSpeculativeTask(TezTaskID taskId) {
readLock.lock();
try {
Preconditions.checkState(taskId.getId() < numTasks);
eventHandler.handle(new TaskEvent(taskId, TaskEventType.T_ADD_SPEC_ATTEMPT));
} finally {
readLock.unlock();
}
}
So the speculator never talks to TaskImpl directly; it asks the vertex, which
emits T_ADD_SPEC_ATTEMPT. In RUNNING, that hits RedundantScheduleTransition
and spawns an extra attempt.
Speculation is off by default:
// tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java
public static final String TEZ_AM_SPECULATION_ENABLED = TEZ_AM_PREFIX + "speculation.enabled";
public static final boolean TEZ_AM_SPECULATION_ENABLED_DEFAULT = false;
Because a task can legitimately have several attempts alive at once (a retry
plus a speculative copy, or two speculative copies), RUNNING carries a
self-loop on T_ATTEMPT_LAUNCHED — the source even comments it
//more attempts may start later. The reschedule logic also avoids waste:
AttemptFailedTransition only spawns a replacement when
shouldScheduleNewAttempt() is true, so a task that already has a spare attempt
in flight does not pile on a redundant one. Reading these self-loops and guards
is how you convince yourself the task never leaks or double-counts attempts.
Winner selection: AttemptSucceededTransition
When an attempt succeeds, TaskImpl records it as the canonical winner and
kills every other live attempt. The kill cause depends on launch order — this
is where the effective vs ineffective speculation distinction is stamped:
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java (AttemptSucceededTransition)
task.successfulAttempt = successTaId;
task.eventHandler.handle(new VertexEventTaskCompleted(task.taskId, TaskState.SUCCEEDED));
LOG.info("Task succeeded with attempt " + task.successfulAttempt);
task.logJobHistoryTaskFinishedEvent();
TaskAttempt successfulAttempt = task.attempts.get(successTaId);
// issue kill to all other attempts
for (TaskAttempt attempt : task.attempts.values()) {
if (!attempt.getTaskAttemptID().equals(task.successfulAttempt) && !attempt.isFinished()) {
String diagnostics; TaskAttemptTerminationCause errCause;
if (attempt.getLaunchTime() < successfulAttempt.getLaunchTime()) {
diagnostics = "Killed this attempt as other speculative attempt : " + successTaId + " succeeded";
errCause = TaskAttemptTerminationCause.TERMINATED_EFFECTIVE_SPECULATION;
} else {
diagnostics = "Killed this speculative attempt as original attempt: " + successTaId + " succeeded";
errCause = TaskAttemptTerminationCause.TERMINATED_INEFFECTIVE_SPECULATION;
}
task.eventHandler.handle(new TaskAttemptEventKillRequest(attempt
.getTaskAttemptID(), diagnostics, errCause));
}
}
return task.finished(TaskStateInternal.SUCCEEDED);
Read that carefully: if the later-launched (speculative) attempt won, the
original loser is killed with TERMINATED_EFFECTIVE_SPECULATION — speculation
paid off. If the original won, the speculative copy is killed with
TERMINATED_INEFFECTIVE_SPECULATION — speculation wasted resources. Neither
cause counts against the failure budget, because both arrive at the loser's task
as T_ATTEMPT_KILLED. The winner (successfulAttempt) is what downstream
consumers fetch from; the losers' partial outputs are discarded.
There is also a guard against a subtle bug: the succeeded attempt must be the one the task selected to commit, if any:
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
if (task.commitAttempt != null && !task.commitAttempt.equals(successTaId)) {
// The succeeded attempt is not the one that was selected to commit
// This is impossible and has to be a bug
throw new TezUncheckedException("TA: " + successTaId
+ " succeeded but TA: " + task.commitAttempt + " was expected to commit and succeed");
}
The commit go/no-go itself is arbitrated by TaskImpl.canCommit(...), which
ensures at most one attempt is granted commit permission:
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
public boolean canCommit(TezTaskAttemptID taskAttemptID) {
writeLock.lock();
try {
TaskState state = getState();
if (state == TaskState.SCHEDULED) {
// Event processing delay: the attempt is asking to commit before the
// task's state machine caught up to RUNNING. Make it retry.
return false;
}
if (state != TaskState.RUNNING) {
eventHandler.handle(new TaskAttemptEventKillRequest(taskAttemptID,
"Task not running. Bad attempt.", TaskAttemptTerminationCause.TERMINATED_ORPHANED));
return false;
}
if (commitAttempt == null) {
TaskAttempt ta = getAttempt(taskAttemptID);
TaskAttemptState taState = ta.getStateNoLock();
if (taState == TaskAttemptState.RUNNING) {
commitAttempt = taskAttemptID; // first RUNNING attempt to ask wins the commit slot
return true;
}
return false;
} else {
return commitAttempt.equals(taskAttemptID); // only the chosen attempt may commit
}
} finally { writeLock.unlock(); }
}
Read the branches: an attempt that asks to commit before the task's own state
machine has reached RUNNING is told to retry (a SCHEDULED snapshot means the
event queue is behind); an attempt asking while the task is not running at all
gets a TERMINATED_ORPHANED kill; and the first RUNNING attempt to ask
claims the commitAttempt slot, after which only it may commit. This is what
makes the commitAttempt != successTaId guard in AttemptSucceededTransition
provably unreachable in correct operation — the same lock and the same slot
govern both call sites.
Note:
TaskImpldoes not schedule its own attempts against YARN. It constructs aTaskAttemptImpland sendsTA_SCHEDULE; the attempt then deals with the scheduler.TaskImplonly decides whether to create an attempt and which attempt won. Container allocation lives in task-attempt-lifecycle.md and scheduler.md.
The KILL_WAIT drain
When a task is terminated while attempts are live (T_TERMINATE in RUNNING or
SCHEDULED), it does not go straight to KILLED; it enters KILL_WAIT and
issues kills to its attempts, then waits for each to acknowledge. The
KillWaitAttemptCompletedTransition is a multiple-arc that only finalizes once
every attempt has finished:
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
private static class KillWaitAttemptCompletedTransition implements
MultipleArcTransition<TaskImpl, TaskEvent, TaskStateInternal> {
@Override
public TaskStateInternal transition(TaskImpl task, TaskEvent event) {
TaskEventTAUpdate castEvent = (TaskEventTAUpdate)event;
task.handleTaskAttemptCompletion(castEvent.getTaskAttemptID(),
TaskAttemptStateInternal.KILLED);
task.taskAttemptStatus.put(castEvent.getTaskAttemptID().getId(), true);
// check whether all attempts are finished
if (task.getFinishedAttemptsCount() == task.attempts.size()) {
task.logJobHistoryTaskFailedEvent(getExternalState(TaskStateInternal.KILLED));
task.eventHandler.handle(new VertexEventTaskCompleted(
task.taskId, getExternalState(TaskStateInternal.KILLED)));
return TaskStateInternal.KILLED;
}
return task.getInternalState();
}
}
It is registered for T_ATTEMPT_KILLED, T_ATTEMPT_FAILED, and
T_ATTEMPT_SUCCEEDED — a task being killed must absorb an attempt that
succeeds mid-kill without blowing up. KILL_WAIT also declares an ignorable
EnumSet for T_TERMINATE, T_ATTEMPT_LAUNCHED, and T_ADD_SPEC_ATTEMPT,
because all three can legitimately arrive at a task that is already draining.
This drain is why external observers see KILLED (via getExternalState) the
moment the kill is issued, even though internally the task lingers in
KILL_WAIT until the last attempt reports back.
Recovering a succeeded attempt
The recovery branch inside AttemptSucceededTransition is worth calling out
because it is where output committers and recovery intersect. On AM restart, a
task whose attempt previously succeeded tries to recover that success rather
than re-run it — but only if the vertex's OutputCommitter supports task
recovery:
// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
private String recoverSuccessTaskAttempt(TaskImpl task) {
String errorMsg = null;
if (task.getVertex().getOutputCommitters() != null
&& !task.getVertex().getOutputCommitters().isEmpty()) {
for (Entry<String, OutputCommitter> entry : task.getVertex().getOutputCommitters().entrySet()) {
OutputCommitter committer = entry.getValue();
if (!committer.isTaskRecoverySupported()) {
errorMsg = "Task recovery not supported by committer, failing task attempt";
break;
}
committer.recoverTask(task.getTaskID().getId(),
task.appContext.getApplicationAttemptId().getAttemptId() - 1);
}
}
return errorMsg;
}
If the committer cannot recover the task, the transition throws away the
recovered success, nulls successfulAttempt, and schedules a fresh attempt
(counting against the budget only if that reschedule fails). This is the state
machine reused verbatim for recovery, exactly as promised in
state-machines.md: the same AttemptSucceededTransition
handles both a live TA_DONE and a replayed one, forking on
task.recoveryData != null.
Reading exercise
cd /path/to/tez
# 1. The full machine
sed -n '/= new StateMachineFactory<TaskImpl/,/installTopology()/p' \
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java | less
# 2. The failure-budget decision
grep -n "failedAttempts\|maxFailedAttempts\|maxAttempts\|shouldScheduleNewAttempt" \
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
# 3. Winner selection and speculation causes
grep -n "successfulAttempt\|TERMINATED_EFFECTIVE_SPECULATION\|TERMINATED_INEFFECTIVE_SPECULATION" \
tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java
# 4. The speculator's request path
grep -rn "T_ADD_SPEC_ATTEMPT\|scheduleSpeculativeTask\|addSpeculativeAttempt" \
tez-dag/src/main/java/org/apache/tez/dag/app/
Then answer:
- Cite the exact condition in
AttemptFailedTransitionthat keeps a task inRUNNINGversus moving it toFAILED. What role doesTaskFailureTypeplay? - Why does a killed attempt (
AttemptKilledTransition) not incrementfailedAttempts? Which counter does it bump instead? - What is the difference between
tez.am.task.max.failed.attempts(default 4) andtez.am.task.max.attempts(default 0)? When would you set the latter? - In
AttemptSucceededTransition, what decides whether a killed loser getsTERMINATED_EFFECTIVE_SPECULATIONvsTERMINATED_INEFFECTIVE_SPECULATION? - Why is the
commitAttempt != successTaIdcheck described as "impossible and has to be a bug"? What invariant doescanCommitmaintain to make it so? - Explain the
SUCCEEDED → SCHEDULEDretroactive-failure arc. What upstream event causes it?
Common bugs and symptoms
| Symptom | Root cause | Where to look |
|---|---|---|
| Task retries forever, never fails | max.failed.attempts set very high, or every failure classified as killed not failed | Check config; verify the TaskAttemptTerminationCause for each terminated attempt |
| Task fails on the first failure unexpectedly | The failure arrived as TaskFailureType.FATAL | Inspect the processor's reported failure type in the attempt log |
| Speculation kills the original just as it succeeds (lost work) | Race between winner selection and speculative kill | Confirm the loser was killed with a *_SPECULATION cause, not counted as failure |
Task SUCCEEDED but a sibling attempt stays RUNNING for a long time | Container slow to acknowledge the kill | ContainerHeartbeatHandler; the TA_KILL_REQUEST path |
TezUncheckedException: ... was expected to commit and succeed | A non-commit attempt succeeded while another held commit | This is the guard firing on a real bug — inspect canCommit/commitAttempt history |
Task comes back RUNNING in recovery though it had finished | Missing TaskFinished history event in the recovery log | Investigate RecoveryService flush boundaries |
Validation: prove you understand this
- Draw the
TaskImplmachine from memory, includingKILL_WAIT, and note which internal state maps to which external state. - From
TestTaskImpl, find a test that drives a task toFAILEDand walk the exact event sequence (how manyT_ATTEMPT_FAILEDevents, and why that many). - List every
TaskAttemptTerminationCausethat arrives asT_ATTEMPT_KILLED(and therefore doesn't count against the budget), citing the enum in task-attempt-lifecycle.md. - Trace, statement by statement, what
AttemptSucceededTransitiondoes when a second concurrent attempt succeeds after the first was already the winner (hint: read the terminal-state ignorable arcs). - Enable
tez.am.speculation.enabled=trueon a skewed job inMiniTezCluster, then find theaddSpeculativeAttemptlog line and follow it to theT_ADD_SPEC_ATTEMPTthe task receives.