Event Routing

Events are the only sanctioned way to mutate an AM-side entity. A VertexImpl never reaches into a TaskImpl and flips a field; it constructs a TaskEvent and hands it to the dispatcher, which serializes it onto a queue, pulls it off on a single thread, resolves the target object, and invokes handle(). This chapter catalogs the real event-type enums, shows how they are wired to handlers in DAGAppMaster.serviceInit, explains the central AsyncDispatcher and the optional concurrent dispatchers, and walks a task completion percolating up to the DAG.

This is the companion to state-machines.md: that chapter told you what happens inside handle(); this one tells you how an event reaches handle() in the first place.

After this chapter you can: name the six AM event-type enums and their producers, trace any transition back to the event that caused it, explain why the AM uses one central dispatcher plus a handful of specialized ones, and recognize the failure modes of a backed-up event queue.

The event-type enums

Every entity has a matching *EventType enum in tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/. List them:

ls tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/*EventType.java

The enums are small and self-documenting — the source even annotates each constant with its producer. VertexEventType (verify with sed -n '/enum/,/}/p' on the file):

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/VertexEventType.java
public enum VertexEventType {
  //Producer:Client, Job
  V_TERMINATE,
  //Producer:Job
  V_INIT,
  //Producer:Vertex
  V_COMPLETED,
  V_START,
  V_SOURCE_TASK_ATTEMPT_COMPLETED,
  V_SOURCE_VERTEX_STARTED,
  V_DELETE_SHUFFLE_DATA,
  //Producer:Task
  V_TASK_COMPLETED,
  V_TASK_RESCHEDULED,
  V_TASK_ATTEMPT_COMPLETED,
  //Producer:Any component
  V_INTERNAL_ERROR,
  V_MANAGER_USER_CODE_ERROR,
  V_ROUTE_EVENT,
  //Producer: VertexInputInitializer
  V_ROOT_INPUT_INITIALIZED,
  V_ROOT_INPUT_FAILED,
  V_INPUT_DATA_INFORMATION,
  // Recover Event, Producer:DAG
  V_RECOVER,
  // Producer: Vertex
  V_READY_TO_INIT,
  // Producer: Edge
  V_NULL_EDGE_INITIALIZED,
  // Committer
  V_COMMIT_COMPLETED,
}

The complete AM-side set, with the producer relationships that matter:

Alongside these live the scheduler and infrastructure enums — AMSchedulerEventType, AMContainerEventType, AMNodeEventType, ContainerLauncherEventType, CallableEventType. They follow the same pattern; the six above are the ones you touch when working on DAG execution logic.

The concrete event classes (one per payload shape) sit in the same package. TaskAttemptEventType.TA_FAILED, for instance, is carried by TaskAttemptEventAttemptFailed, which holds the TaskFailureType and a TaskAttemptTerminationCause:

ls tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/TaskAttemptEvent*.java

The "events are the only mutation API" rule

Any change to the state of a DAGImpl, VertexImpl, TaskImpl, or TaskAttemptImpl must occur inside a state-machine transition hook, triggered by an event that flowed through the dispatcher.

Why this is non-negotiable:

1. Concurrency safety. The dispatcher serializes everything onto one thread; combined with the per-entity write lock from state-machines.md, that makes transition hooks single-threaded by construction. A direct cross-thread setter reintroduces races that no unit test reliably catches.
2. Auditability. Events and their resulting transitions are logged; silent field writes are not. The transitioned from X to Y due to event Z lines are the AM's flight recorder.
3. Recoverability. RecoveryService persists a subset of events as history events; on AM restart they are replayed through the same machines. State produced by a side-door mutation is invisible to recovery and will not come back.

The one sanctioned side door is read-only getters, which take the read lock and tolerate slight staleness. Everything that changes state goes through an event.

Bubble-up: a task completion reaches the DAG

Nothing illustrates the model better than following a single successful task attempt all the way to the DAG counter. Every arrow below is either an eventHandler.handle(...) emit or a doTransition inside a hook:

sequenceDiagram
    participant TCM as TaskCommunicatorManager
    participant TA as TaskAttemptImpl
    participant T as TaskImpl
    participant V as VertexImpl
    participant D as DAGImpl
    participant DI as AsyncDispatcher

    Note over TCM: umbilical heartbeat carries TASK_ATTEMPT_COMPLETED
    TCM->>DI: TA_DONE
    DI->>TA: handle(TA_DONE)
    TA->>DI: T_ATTEMPT_SUCCEEDED
    DI->>T: handle(T_ATTEMPT_SUCCEEDED)
    T->>T: pick winner; kill other attempts
    T->>DI: VertexEventTaskCompleted(SUCCEEDED)
    DI->>V: handle(V_TASK_COMPLETED)
    V->>V: checkTasksForCompletion(); bump succeededTaskCount
    alt all tasks done, no committer
        V->>DI: DAGEventVertexCompleted
        DI->>D: handle(DAG_VERTEX_COMPLETED)
        D->>D: bump succeededVertexCount
    end

The umbilical entry point is real and worth reading: the AM receives task progress and completion through TaskCommunicatorManager, which translates runtime TezEvents into AM TaskAttemptEvents. Note the ordering comment in the source — status must precede completion:

// tez-dag/src/main/java/org/apache/tez/dag/app/TaskCommunicatorManager.java
if (eventType == EventType.TASK_STATUS_UPDATE_EVENT) {
  // send TA_STATUS_UPDATE before TA_DONE/TA_FAILED/TA_KILLED otherwise Status may be missed
  taskAttemptEvent = new TaskAttemptEventStatusUpdate(taskAttemptID,
      (TaskStatusUpdateEvent) tezEvent.getEvent());
} else if (eventType == EventType.TASK_ATTEMPT_COMPLETED_EVENT
   || eventType == EventType.TASK_ATTEMPT_FAILED_EVENT
   || eventType == EventType.TASK_ATTEMPT_KILLED_EVENT) {
  taFinishedEvents.add(tezEvent);
}

Find the emit sites for yourself:

grep -n "eventHandler.handle\|sendEvent(" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskImpl.java | head
grep -n "eventHandler.handle" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java | wc -l

The winner-selection step inside TaskImpl (AttemptSucceededTransition) and the completion check inside VertexImpl (checkTasksForCompletion) are detailed in task-lifecycle.md and vertex-lifecycle.md respectively.

Crossing subsystems: the scheduler and the CallableEvent

Not every event stays within the DAG-execution machines. A TaskAttemptImpl that needs a container emits an AMSchedulerEvent into the scheduler subsystem, whose enum is a different world entirely:

// tez-dag/src/main/java/org/apache/tez/dag/app/rm/AMSchedulerEventType.java
public enum AMSchedulerEventType {
  //Producer: TaskAttempt
  S_TA_LAUNCH_REQUEST,
  S_TA_STATE_UPDATED,
  S_TA_ENDED, // Annotated with FAILED/KILLED/SUCCEEDED.
  //Producer: Node
  S_NODE_BLACKLISTED,
  S_NODE_UNBLACKLISTED,
  S_NODE_UNHEALTHY,
  S_NODE_HEALTHY,
  // Producer : AMContainer
  S_CONTAINER_DEALLOCATE
}

So a single attempt scheduling request (TA_SCHEDULE → attempt → S_TA_LAUNCH_REQUEST → TaskSchedulerManager) crosses two enums and two handlers, all on the same central dispatcher. This is the pattern to internalize: subsystems communicate only through the dispatcher, never by direct method call, so that the same serialization and ordering guarantees hold across the whole AM, not just within one machine.

The one deliberate exception is heavy or blocking work, which is wrapped in a CallableEvent and submitted to the AM's execService rather than enqueued on the dispatcher. Its type enum has a single value:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/CallableEventType.java
public enum CallableEventType {
  CALLABLE,
}

VertexImpl uses this to run OutputCommitter.commitOutput() off the dispatch thread (see vertex-lifecycle.md); when the callable finishes, its Guava FutureCallback emits a normal dispatcher event (V_COMMIT_COMPLETED) to fold the result back into the state machine. This is the sanctioned bridge between "async work on a thread pool" and "single-threaded state machine": do the work on execService, report the result as an event.

Where events are registered

Registration happens in one place: DAGAppMaster.serviceInit. The dispatcher is created first, then every event-type enum is mapped to a handler:

grep -n "dispatcher.register\|registerAndCreateDispatcher\|createDispatcher" \
  tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java

// tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java
dispatcher.register(AMContainerEventType.class, containers);
// ...
dispatcher.register(AMNodeEventType.class, nodes);

this.dagEventDispatcher = new DagEventDispatcher();
this.vertexEventDispatcher = new VertexEventDispatcher();

dispatcher.register(DAGAppMasterEventType.class, new DAGAppMasterEventHandler());
dispatcher.register(DAGEventType.class, dagEventDispatcher);
dispatcher.register(VertexEventType.class, vertexEventDispatcher);
boolean useConcurrentDispatcher =
    conf.getBoolean(TezConfiguration.TEZ_AM_USE_CONCURRENT_DISPATCHER,
        TezConfiguration.TEZ_AM_USE_CONCURRENT_DISPATCHER_DEFAULT);
if (!useConcurrentDispatcher) {
  dispatcher.register(TaskEventType.class, new TaskEventDispatcher());
  dispatcher.register(TaskAttemptEventType.class, new TaskAttemptEventDispatcher());
} else {
  int concurrency = conf.getInt(TezConfiguration.TEZ_AM_CONCURRENT_DISPATCHER_CONCURRENCY,
      TezConfiguration.TEZ_AM_CONCURRENT_DISPATCHER_CONCURRENCY_DEFAULT);
  AsyncDispatcherConcurrent sharedDispatcher = dispatcher.registerAndCreateDispatcher(
      TaskEventType.class, new TaskEventDispatcher(), "TaskAndAttemptEventThread", concurrency);
  dispatcher.registerWithExistingDispatcher(TaskAttemptEventType.class,
      new TaskAttemptEventDispatcher(), sharedDispatcher);
}

// register other delegating dispatchers
dispatcher.registerAndCreateDispatcher(SpeculatorEventType.class, new SpeculatorEventHandler(),
    "Speculator");

Each registered handler is a small inner class that does one thing: resolve the target entity from IDs carried in the event, then forward. TaskEventDispatcher is representative:

// tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java
private class TaskEventDispatcher implements EventHandler<TaskEvent> {
  @Override
  public void handle(TaskEvent event) {
    DAG dag = context.getCurrentDAG();
    int eventDagIndex = event.getDAGID().getId();
    if (dag == null || eventDagIndex != dag.getID().getId()) {
      return; // event not relevant anymore
    }
    Task task = dag.getVertex(event.getVertexID()).getTask(event.getTaskID());
    ((EventHandler<TaskEvent>)task).handle(event);
  }
}

Two design points fall out of this:

• Events carry IDs, not object references. TezTaskID, TezVertexID, TezTaskAttemptID — the dispatcher resolves them against the current DAG. This is what makes the eventDagIndex != dag.getID().getId() guard possible: in session mode a stale event from a previous DAG is simply dropped rather than misapplied to the new DAG.
• The dispatcher handler is the resolve step. The state machine never sees an ID it has to look up; it always gets a live object.

One central dispatcher, several specialized ones

The AM's backbone is a single central AsyncDispatcher named "Central":

// tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java
protected AsyncDispatcher createDispatcher() {
  return new AsyncDispatcher("Central");
}

Tez uses its own org.apache.tez.common.AsyncDispatcher (not YARN's) plus org.apache.tez.common.AsyncDispatcherConcurrent. The central dispatcher owns a single LinkedBlockingQueue drained by one thread. But it can delegate certain event types to nested dispatchers with their own threads. The GenericEventHandler that every emit lands in checks for a delegate first:

// tez-common/src/main/java/org/apache/tez/common/AsyncDispatcher.java
class GenericEventHandler implements EventHandler<Event> {
  public void handle(Event event) {
    if (stopped) { return; }
    if (blockNewEvents) { return; }
    drained = false;
    // offload to specific dispatcher if one exists
    Class<? extends Enum> type = event.getType().getDeclaringClass();
    AsyncDispatcher registeredDispatcher = eventDispatchers.get(type);
    if (registeredDispatcher != null) {
      registeredDispatcher.getEventHandler().handle(event);
      return;
    }
    AsyncDispatcherConcurrent concurrentDispatcher = concurrentEventDispatchers.get(type);
    if (concurrentDispatcher != null) {
      concurrentDispatcher.getEventHandler().handle(event);
      return;
    }
    // no registered dispatcher. use internal dispatcher.
    int qSize = eventQueue.size();
    if (qSize != 0 && qSize % 1000 == 0) {
      LOG.info("Size of event-queue is " + qSize);
    }
    int remCapacity = eventQueue.remainingCapacity();
    if (remCapacity < 1000) {
      LOG.warn("Very low remaining capacity in the event-queue: " + remCapacity);
    }
    // ... eventQueue.put(event);
  }
}

So the topology on master is:

The speculator gets its own thread because status updates are high-volume and must not compete with lifecycle events on the central queue — S_TASK_ATTEMPT_STATUS_UPDATE fires on every heartbeat of every attempt.

Note: The concurrent dispatcher (tez.am.use.concurrent-dispatcher, default false) exists for very large DAGs where a single thread can't keep up with task/attempt event volume. It is safe because events are routed to queues by getSerializingHash() — all events for a given task land on the same queue and thread, preserving per-entity ordering:

// tez-common/src/main/java/org/apache/tez/common/AsyncDispatcherConcurrent.java
int index = numThreads > 1 ? event.getSerializingHash() % numThreads : 0;

TaskEvent.getSerializingHash() and TaskAttemptEvent.getSerializingHash() both derive from the task ID, so a task and its attempts share a thread.

Ordering guarantees

The invariant that keeps state machines correct:

All events destined for a single entity are processed in the order they were enqueued, by exactly one thread.

On the central dispatcher this is trivially true — one queue, one thread. On the concurrent dispatcher it holds because of the serializing-hash routing above. What Tez does not guarantee is global ordering across different entities: if TaskImpl A emits V_TASK_COMPLETED and TaskImpl B emits V_TASK_COMPLETED at nearly the same time, the vertex may see them in either order. Transition hooks must therefore be commutative with respect to independent siblings — which is why completion logic is written as "increment a counter and re-check totals" (checkTasksForCompletion) rather than "assume this is the last one."

What happens when the queue backs up

The central queue is an unbounded LinkedBlockingQueue, so it will not reject events — it grows. The symptoms of a backup, straight from the code above:

• Size of event-queue is N logged every 1000 events.
• Very low remaining capacity in the event-queue when remainingCapacity() drops below 1000 (only meaningful if a bounded queue was configured).
• On DAG completion, DAGAppMaster logs the residual queue depth: Central Dispatcher queue size after DAG completion, before cleanup: N.

A growing queue almost always means a transition hook is doing blocking work on the dispatch thread (see the warning in state-machines.md). The classic culprits: a synchronous HDFS listing inside an InputInitializer-adjacent transition, or an OutputCommitter.commitOutput() run inline instead of on the execService. The fix is never "make the queue bigger"; it is "get the slow work off the dispatch thread."

On shutdown, the dispatcher can be told to drain:

// tez-common/src/main/java/org/apache/tez/common/AsyncDispatcher.java
if (drainEventsOnStop) {
  blockNewEvents = true;   // GenericEventHandler now drops new events
  // ... waits until drained == true (queue empty) before stopping the thread
}

blockNewEvents makes GenericEventHandler.handle silently drop anything arriving after stop begins, while the dispatch thread finishes what is already queued. This is why late TaskKilled/ContainerCompleted events after a DAG finishes do not crash anything — they are either dropped by blockNewEvents or absorbed by ignorable transitions.

Warning: If enableExitOnDispatchException() was called (it is, for non-local AMs), an uncaught exception on the dispatch thread will call System.exit. That is deliberate: a bug in a transition hook that escapes as a RuntimeException is considered unrecoverable, and a fast crash produces a cleaner failure (and recovery on the next AM attempt) than a zombie AM whose dispatch thread has died silently.

Common deadlock and starvation patterns

Reading exercise

cd /path/to/tez
# 1. All AM event-type enums
ls tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/*EventType.java

# 2. Every registration, in order
grep -n "\.register(\|registerAndCreateDispatcher\|registerWithExistingDispatcher" \
  tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java

# 3. A cross-subsystem emit: TaskAttemptImpl asking the scheduler for a container
grep -n "AMSchedulerEvent" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java | head

# 4. The serializing-hash routing that makes the concurrent dispatcher safe
grep -n "getSerializingHash" \
  tez-common/src/main/java/org/apache/tez/common/*.java \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/event/TaskEvent.java

Then answer:

1. Why do events carry TezTaskID rather than a Task reference? Give two consequences (hint: session mode, recovery).
2. The TaskEventDispatcher drops events whose eventDagIndex doesn't match the current DAG. In what real scenario does that guard fire?
3. Which two event types share the concurrent dispatcher when it is enabled, and what property of getSerializingHash keeps per-task ordering intact?
4. Trace a TA_TIMED_OUT: which class issues it, what does it carry, and which handler receives it? (Start from TaskHeartbeatHandler.)
5. What does enableExitOnDispatchException change, and why is it disabled in local mode? See local-mode.md.
6. Find one VertexEventType constant produced by Edge and explain the cross-entity flow that emits it.

Common bugs and symptoms

Validation: prove you understand this

1. From the source, list every dispatcher.register* call in DAGAppMaster.serviceInit in order, and note which dispatcher (central, speculator, concurrent) each ends up on.
2. Pick one transition in TaskAttemptImpl and enumerate every event it emits; for each, name the receiving entity and how the dispatcher resolves it.
3. Walk a V_TERMINATE (DAG kill) from DAGImpl down to a single TaskAttemptImpl issuing a TA_KILL_REQUEST and eventually a container stop. Cross-reference vertex-lifecycle.md and task-attempt-lifecycle.md.
4. Enable tez.am.use.concurrent-dispatcher in a MiniTezCluster job, then explain — using getSerializingHash — why a task's attempts never race each other despite running on a 10-thread pool.
5. Write a unit test that sends a malformed event (wrong payload cast) into an entity and confirm the AM logs the error and routes to INTERNAL_ERROR rather than corrupting state.