Scheduler

Every task attempt in a Tez DAG needs a container, and the scheduler is the AM-side subsystem that gets it one. It sits between two worlds that know nothing about each other: the DAG engine (state machines for vertices, tasks, and attempts — see task-attempt-lifecycle.md) and the resource source (YARN's ResourceManager, a thread pool in local mode, or LLAP daemons under Hive). Get this layer wrong as a contributor and you produce jobs that hang at 0% with a healthy cluster, priority inversions where a join starves the map feeding it, or preemption storms that kill an hour of work to service a request the RM would have satisfied in ten seconds.

This chapter dissects TaskSchedulerManager (the mediator), the TaskScheduler service-plugin API, the two YARN scheduler implementations that ship on master — YarnTaskSchedulerService and DagAwareYarnTaskScheduler — plus the DAG-topology-to-YARN-priority arithmetic, locality-delay scheduling, and Tez-internal preemption. It runs inside the AM described in dag-app-master.md and talks YARN protocols described in yarn-integration.md. The container-reuse decision path it feeds is detailed separately in container-reuse.md.

After this chapter you can:

• Trace an AMSchedulerEventTALaunchRequest from TaskAttemptImpl to a YARN addContainerRequest and back to a TAEventContainerAssigned equivalent.
• Quote the TaskScheduler abstract API and say which upcalls a plugin owes the AM via TaskSchedulerContext.
• Explain why DagAwareYarnTaskScheduler is the default on master and what it does that YarnTaskSchedulerService cannot.
• Compute the exact YARN priority a task attempt gets from its vertex's distance-from-root, and predict who preempts whom.
• Tune the locality-delay and preemption knobs from real defaults, not folklore.

All commands below run from a Tez checkout root. Code moves between branches — trust the greps, not remembered line numbers.

The cast

ls tez-dag/src/main/java/org/apache/tez/dag/app/rm/

Two files dominate: YarnTaskSchedulerService.java (~94K) and DagAwareYarnTaskScheduler.java (~74K). Between them sits ~47K of TaskSchedulerManager.java. That is the whole subsystem.

                 DAG engine (tez-dag)                     resource world
  ┌────────────────────────────────────────┐
  │ VertexImpl → TaskImpl → TaskAttemptImpl│
  │        │  TaskAttemptEventSchedule     │
  │        ▼                               │
  │  AMSchedulerEventTALaunchRequest       │
  └───────────────────┬────────────────────┘
                      ▼
  ┌────────────────────────────────────────┐
  │        TaskSchedulerManager            │  one per AM
  │  handleEvent() switch on event type    │
  │  taskSchedulers[schedulerId] ──────────┼──► TaskScheduler plugin
  │  ◄── upcalls via TaskSchedulerContext  │      ├─ DagAwareYarnTaskScheduler ─► YARN RM
  │      (taskAllocated, containerCompleted│      ├─ YarnTaskSchedulerService  ─► YARN RM
  │       preemptContainer, ...)           │      ├─ LocalTaskSchedulerService ─► threads
  └────────────────────────────────────────┘      └─ custom (e.g. LLAP)       ─► daemons

TaskSchedulerManager: the mediator

grep -n "S_TA_LAUNCH_REQUEST\|S_TA_ENDED\|S_CONTAINER_DEALLOCATE\|S_NODE" \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/AMSchedulerEventType.java
grep -n "handleTaLaunchRequest\|handleTASucceeded\|handleTAUnsuccessfulEnd\|handleContainerDeallocate" \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/TaskSchedulerManager.java

TaskSchedulerManager is an AbstractService and EventHandler wired into the AM's central AsyncDispatcher. Everything the DAG engine wants from the resource world arrives as an AMSchedulerEvent. The full event vocabulary is one small enum — read it before anything else:

// 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
}

The concrete event classes are AMSchedulerEventTALaunchRequest, AMSchedulerEventTAStateUpdated, AMSchedulerEventTAEnded, AMSchedulerEventDeallocateContainer, and AMSchedulerEventNodeBlacklistUpdate — nothing else. If you remember a AMSchedulerEventContainerCompleted from old blog posts, it does not exist on master; container completion flows back through the scheduler's RM callback and the containerCompleted upcall, not through this enum.

handleEvent is a plain switch that routes each event to a handler, and every handler ends in a call on taskSchedulers[event.getSchedulerId()] — the array of active scheduler plugins, indexed by scheduler ID (a DAG can name multiple schedulers; the vertex → scheduler binding rides on the event):

// tez-dag/src/main/java/org/apache/tez/dag/app/rm/TaskSchedulerManager.java
public synchronized void handleEvent(AMSchedulerEvent sEvent) {
  LOG.debug("Processing the event {}", sEvent);
  switch (sEvent.getType()) {
  case S_TA_LAUNCH_REQUEST:
    handleTaLaunchRequest((AMSchedulerEventTALaunchRequest) sEvent);
    break;
  // ...
  case S_TA_ENDED: // TaskAttempt considered complete.
    AMSchedulerEventTAEnded event = (AMSchedulerEventTAEnded)sEvent;
    switch(event.getState()) {
    case FAILED:
    case KILLED:
      handleTAUnsuccessfulEnd(event);
      break;
    case SUCCEEDED:
      handleTASucceeded(event);
      break;
    // ...

handleTaLaunchRequest unpacks the TaskLocationHint into hosts[] / racks[] arrays — or, when the hint carries a TaskBasedLocationAffinity, resolves the affinity target's assigned ContainerId and calls the container-affinity allocateTask overload instead — then invokes taskSchedulers[schedulerId].allocateTask(...) with Priority.newInstance(event.getPriority()). Note what this means: priority is already computed before the scheduler ever sees the request (next section).

The reverse direction — scheduler to AM — is the callback surface. When TaskSchedulerManager instantiates a plugin it builds a TaskSchedulerContextImpl and wraps it in TaskSchedulerContextImplWrapper, whose javadoc states the contract: "Makes use of an ExecutionService to invoke application callbacks" — a single-threaded TaskSchedulerAppCallbackExecutor, so plugin upcalls are serialized and never run AM logic on the RM heartbeat thread. The most important upcall:

// tez-dag/src/main/java/org/apache/tez/dag/app/rm/TaskSchedulerManager.java
public synchronized void taskAllocated(int schedulerId, Object task,
                                         Object appCookie,
                                         Container container) {
  AMSchedulerEventTALaunchRequest event =
      (AMSchedulerEventTALaunchRequest) appCookie;
  // ...
  if (amContainer.getState() == AMContainerState.ALLOCATED) {
    sendEvent(new AMContainerEventLaunchRequest(containerId, taskAttempt.getVertexID(),
        event.getContainerContext(), event.getLauncherId(), event.getTaskCommId()));
  }
  // ...
  sendEvent(new AMContainerEventAssignTA(containerId, taskAttempt.getTaskAttemptID(),
      event.getRemoteTaskSpec(), /* ... */ event.getPriority()));
}

The appCookie the plugin hands back is literally the original launch-request event — the scheduler treats it as opaque and returns it on assignment, which is how the manager reconnects a YARN Container to the attempt, launcher, and task-communicator that requested it. Other upcalls: containerCompleted (translates YARN exit status into AMContainerEventCompleted, mapping ContainerExitStatus.PREEMPTED to TaskAttemptTerminationCause.EXTERNAL_PREEMPTION), containerBeingReleased, nodesUpdated, appShutdownRequested, and preemptContainer — which, note, just turns around and calls deallocateContainer on the owning plugin.

The TaskScheduler plugin API

grep -n "public abstract" tez-api/src/main/java/org/apache/tez/serviceplugins/api/TaskScheduler.java
grep -n "void taskAllocated\|void containerCompleted\|void preemptContainer\|getVertexIndexForTask" \
  tez-api/src/main/java/org/apache/tez/serviceplugins/api/TaskSchedulerContext.java

Every scheduler — YARN, local, LLAP — extends one abstract class in tez-api. This is public plugin surface, so it changes slowly and is the right thing to memorize:

// tez-api/src/main/java/org/apache/tez/serviceplugins/api/TaskScheduler.java
public abstract class TaskScheduler implements ServicePluginLifecycle {
  // ...
  public abstract Resource getAvailableResources() throws ServicePluginException;
  public abstract Resource getTotalResources() throws ServicePluginException;
  public abstract int getClusterNodeCount() throws ServicePluginException;
  public abstract void blacklistNode(NodeId nodeId) throws ServicePluginException;
  public abstract void unblacklistNode(NodeId nodeId) throws ServicePluginException;
  public abstract void allocateTask(Object task, Resource capability,
                                    String[] hosts, String[] racks, Priority priority,
                                    Object containerSignature, Object clientCookie)
      throws ServicePluginException;
  public abstract void allocateTask(Object task, Resource capability,
                                    ContainerId containerId, Priority priority,
                                    Object containerSignature, Object clientCookie)
      throws ServicePluginException;
  public abstract boolean deallocateTask(Object task, boolean taskSucceeded,
                                         TaskAttemptEndReason endReason,
                                         @Nullable String diagnostics)
      throws ServicePluginException;
  public abstract Object deallocateContainer(ContainerId containerId)
      throws ServicePluginException;
  public abstract void setShouldUnregister() throws ServicePluginException;
  public abstract boolean hasUnregistered() throws ServicePluginException;
  public abstract void dagComplete() throws ServicePluginException;
}

Two allocateTask overloads: one by hosts/racks (fresh placement), one by ContainerId (task-affinity — "put me where that other attempt ran"). The task is an opaque Object (in practice a TaskAttempt), and the API-doc on the hosts/racks overload nails the priority convention: "A lower value implies a higher priority." deallocateTask fires on every attempt end; deallocateContainer force-releases a specific container and is exactly what Tez-internal preemption uses.

The plugin's window back into the AM is TaskSchedulerContext (tez-api/.../serviceplugins/api/TaskSchedulerContext.java): taskAllocated, containerCompleted, containerBeingReleased, nodesUpdated, preemptContainer, setApplicationRegistrationData, appShutdownRequested, plus queries like getAMState(), isSession(), getCurrentDagInfo(), and getVertexIndexForTask(Object) — the last two exist specifically so a scheduler can be DAG-aware without depending on tez-dag internals.

Note: There is no tez.am.task.scheduler.classes config key on master. Custom schedulers are registered programmatically via ServicePluginsDescriptor (same package), which carries TaskSchedulerDescriptor[]; TaskSchedulerManager.createTaskScheduler matches each descriptor's entity name against TezConstants.getTezYarnServicePluginName() / getTezUberServicePluginName() and reflects up anything else by class name. Hive LLAP plugs in through this path.

Two YARN schedulers, one knob

grep -n "TEZ_AM_YARN_SCHEDULER_CLASS" \
  tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/TaskSchedulerManager.java

Both implementations live side by side in tez-dag/.../dag/app/rm/, and the selector is a real, verifiable config key:

// tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java
public static final String TEZ_AM_YARN_SCHEDULER_CLASS = TEZ_AM_PREFIX + "yarn.scheduler.class";
public static final String TEZ_AM_YARN_SCHEDULER_CLASS_DEFAULT =
    "org.apache.tez.dag.app.rm.DagAwareYarnTaskScheduler";

So on master, tez.am.yarn.scheduler.class defaults to DagAwareYarnTaskScheduler; set it to org.apache.tez.dag.app.rm.YarnTaskSchedulerService to fall back to the classic scheduler. TaskSchedulerManager.createYarnTaskScheduler reflects the configured class whenever a descriptor names the YARN plugin.

Warning: Do not confuse tez.am.yarn.scheduler.class (which TaskScheduler plugin talks to YARN) with tez.am.dag.scheduler.class (which DAGScheduler assigns priorities — next section). They are unrelated knobs with confusingly similar names.

YarnTaskSchedulerService: the delayed-container loop

grep -n "class DelayedContainerManager\|assignDelayedContainer\|LocalityMatchLevel" \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/YarnTaskSchedulerService.java | head -20

allocateTask wraps the request in a CookieContainerRequest (the cookie carries the task, the app cookie, and the container signature) and hands it to AMRMClientAsync.addContainerRequest. When containers arrive or fall idle, assignment runs through three ContainerAssigner singletons — NodeLocalContainerAssigner, RackLocalContainerAssigner, NonLocalContainerAssigner — each calling getMatchingRequestWithPriority(container, location) for new containers or getMatchingRequestWithoutPriority(...) for reused ones. That asymmetry is deliberate: a new container was granted by the RM for a specific priority, so it must serve that priority; a reused container is a sunk cost, so any pending request at the right location may take it, top priority first.

One guard in the reuse path is worth quoting because it encodes a subtle deadlock: a new container must not be stolen by a lower-priority task while a higher-priority request is still pending, or the RM — which believes it already satisfied you — will never send a replacement:

// tez-dag/src/main/java/org/apache/tez/dag/app/rm/YarnTaskSchedulerService.java
if (topPendingTaskPriority.compareTo(containerPriority) > 0 &&
    heldContainers.get(container.getId()).isNew()) {
  // if the next task to assign is higher priority than the container then
  // dont assign this container to that task.
  // ...
  return false;
}

Idle containers park in the DelayedContainerManager thread's queue; its mainLoop() wakes when the head's getNextScheduleTime() expires and calls assignDelayedContainer, which escalates the container's LocalityMatchLevel one step per pass (NEW → NODE → RACK → NON_LOCAL, gated by the reuse-fallback knobs) and finally releases containers nobody wants. The full reuse decision tree is in container-reuse.md.

DagAwareYarnTaskScheduler: request matching with topology

grep -n "vertexDescendants\|createVertexBlockedSet\|allowedVertices\|tryAssignReuseContainer" \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/DagAwareYarnTaskScheduler.java | head -20

The class javadoc says it plainly: "A YARN task scheduler that is aware of the dependencies between vertices in the DAG and takes them into account when deciding how to schedule and preempt tasks." On DAG start it caches DagInfo.getVertexDescendants(i) as an ArrayList<BitSet>; the RequestTracker then maintains, per priority, which vertices are requesting and — crucially — allowedVertices: vertices not downstream of any higher-priority pending request. Reuse matching consults it:

// tez-dag/src/main/java/org/apache/tez/dag/app/rm/DagAwareYarnTaskScheduler.java
for (Entry<Priority,RequestPriorityStats> entry : requestTracker.getStatsEntries()) {
  Priority priority = entry.getKey();
  RequestPriorityStats stats = entry.getValue();
  if (!stats.allowedVertices.intersects(stats.vertices)) {
    LOG.debug("Skipping requests at priority {} because all requesting vertices are blocked"
        + " by higher priority requests", priority);
    continue;
  }
  // ...
}

This kills a classic failure of the priority-only scheduler: a reducer request grabbing a freed container while the mapper vertex feeding it still has pending work. New containers are assigned in three straight passes — exact host, then rack, then ResourceRequest.ANY — mirroring the classic assigners but against the tracker instead of the AMRMClient's matching maps:

// tez-dag/src/main/java/org/apache/tez/dag/app/rm/DagAwareYarnTaskScheduler.java
for (Container c : newContainers) {
  HeldContainer hc = new HeldContainer(c);
  heldContainers.put(hc.getId(), hc);
  Resources.addTo(allocatedResources, c.getResource());
  tryAssignNewContainer(hc, hc.getHost(), assignments, unassigned);
}
// ... then hc.getRack(), then ResourceRequest.ANY

The allocation round, end to end

sequenceDiagram
  participant TA as TaskAttemptImpl
  participant DS as DAGScheduler (NaturalOrder)
  participant TSM as TaskSchedulerManager
  participant S as TaskScheduler plugin
  participant RM as YARN RM
  TA->>DS: DAGEventSchedulerUpdate
  DS->>TA: TaskAttemptEventSchedule (priorityLow/HighLimit)
  TA->>TSM: AMSchedulerEventTALaunchRequest (priority, hint, capability)
  TSM->>S: allocateTask(task, capability, hosts, racks, priority, sig, cookie)
  S->>RM: addContainerRequest (heartbeat)
  RM-->>S: onContainersAllocated([Container])
  S->>S: match host → rack → ANY at container priority
  S->>TSM: ctx.taskAllocated(task, cookie, container)
  TSM->>TSM: AMContainerEventLaunchRequest + AMContainerEventAssignTA
  Note over TSM,RM: container launched, TezChild runs the attempt

For tez.local.mode=true the same contract is implemented by LocalTaskSchedulerService with an in-JVM queue instead of an RM — see local-mode.md.

Priority: DAG topology becomes a YARN number

grep -n "getPriorityLowLimit\|getPriorityHighLimit\|getDistanceFromRoot" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/DAGScheduler.java
grep -n "TEZ_AM_DAG_SCHEDULER_CLASS" \
  tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java

Priorities are not set by the YARN scheduler plugins at all — they arrive on the launch request, computed by the DAG scheduler, selected by:

// tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java
public static final String TEZ_AM_DAG_SCHEDULER_CLASS = TEZ_AM_PREFIX + "dag.scheduler.class";
public static final String TEZ_AM_DAG_SCHEDULER_CLASS_DEFAULT =
    "org.apache.tez.dag.app.dag.impl.DAGSchedulerNaturalOrder";

The arithmetic lives in the abstract base and is short enough to memorize:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/DAGScheduler.java
public int getPriorityLowLimit(final DAG dag, final Vertex vertex) {
  final int vertexDistanceFromRoot = vertex.getDistanceFromRoot();
  return ((vertexDistanceFromRoot + 1) * dag.getTotalVertices() * 3)
      + (vertex.getVertexId().getId() * 3);
}

public int getPriorityHighLimit(final DAG dag, final Vertex vertex) {
  return  getPriorityLowLimit(dag, vertex) - 2;
}

Each vertex gets a three-wide priority window keyed on its distance from the DAG roots, tie-broken by vertex ID. Deeper vertices get numerically larger — i.e. worse — priorities, so upstream producers always outrank downstream consumers at the RM and inside every matching loop. TaskAttemptImpl then picks the point inside the window:

// tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/TaskAttemptImpl.java
int priority;
if (ta.isRescheduled  && ta.getVertex().getVertexConfig().getTaskRescheduleHigherPriority()) {
  // higher priority for rescheduled attempts
  priority = scheduleEvent.getPriorityHighLimit();
} else {
  priority = (scheduleEvent.getPriorityHighLimit() + scheduleEvent.getPriorityLowLimit()) / 2;
}

Worked example — a 3-vertex chain A → B → C (IDs 0, 1, 2; distances 0, 1, 2; getTotalVertices()==3):

A rescheduled (retried) attempt gets the top of its window so it jumps ahead of its siblings — a failed map re-run beats queued first-run maps of the same vertex, but never beats a different vertex's window.

DAGSchedulerNaturalOrder.scheduleTaskEx simply computes the window and fires TaskAttemptEventSchedule. Its sibling DAGSchedulerNaturalOrderControlled (same package, same arithmetic) adds gating: per its javadoc it "schedules task attempts belonging to downstream vertices only after all attempts belonging to upstream vertices have been scheduled" — it parks TaskAttemptEventSchedule events in a pendingEvents multimap until trySchedulingVertex clears the vertex, which prevents a slow-started upstream from being fenced out by a flood of already-submitted downstream requests. Natural order remains the default; Controlled is opt-in via the config key above.

Locality and delayed scheduling

grep -n "localitySchedulingDelay" \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/YarnTaskSchedulerService.java \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/DagAwareYarnTaskScheduler.java | head
grep -n "container.reuse.locality\|container.idle.release\|session.min.held" \
  tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java

All knobs quoted from TezConfiguration:

The 250 ms delay is the heart of delay scheduling: when a container frees up, the scheduler first offers it only to node-local requests; if nothing matches within the delay it re-queues the container one level looser. In YarnTaskSchedulerService this is the DelayedContainerManager + LocalityMatchLevel escalation described above; in DagAwareYarnTaskScheduler the held container re-schedules itself on the reuse executor:

// tez-dag/src/main/java/org/apache/tez/dag/app/rm/DagAwareYarnTaskScheduler.java
assignedRequest = tryAssignReuseContainerAppRunning(hc);
if (assignedRequest == null) {
  if (hc.atMaxMatchLevel()) {
    LOG.info("Releasing idle container {} due to pending requests", hc.getId());
    releaseContainer(hc);
  } else {
    hc.scheduleForReuse(localitySchedulingDelay);
  }
}

Tip: For scan-heavy DAGs reading HDFS, leaving non-local-fallback off and tuning the delay upward trades a little idle time for a lot of data-local reads. For shuffle-dominated stages locality barely matters — see shuffle-sort.md — and aggressive reuse wins. Where the hints come from (MRInput splits, VertexLocationHint) is covered in vertex-lifecycle.md.

Preemption: making room for starved priorities

grep -rn "preemptIfNeeded\|maybePreempt" tez-dag/src/main/java/org/apache/tez/dag/app/rm/
grep -n "TEZ_AM_PREEMPTION" tez-api/src/main/java/org/apache/tez/dag/api/TezConfiguration.java

Tez-internal preemption (distinct from YARN preempting the application) fires when a high-priority request cannot be satisfied while lower-priority work holds resources. The knobs, quoted from TezConfiguration:

Both schedulers check on every RM heartbeat. The classic preemptIfNeeded() finds the highest-priority pending request; if it fits in free headroom (and the deadline hasn't passed), do nothing. Otherwise it first releases new unassigned containers at lower priorities, and only then hunts running tasks — requiring genuine priority inversion:

// tez-dag/src/main/java/org/apache/tez/dag/app/rm/YarnTaskSchedulerService.java
for(Map.Entry<Object, Container> entry : taskAllocations.entrySet()) {
  HeldContainer heldContainer = heldContainers.get(entry.getValue().getId());
  CookieContainerRequest lastTaskInfo = heldContainer.getLastTaskInfo();
  Priority taskPriority = lastTaskInfo.getPriority();
  Object signature = lastTaskInfo.getCookie().getContainerSignature();
  if(!isHigherPriority(highestPriRequest.getPriority(), taskPriority)) {
    // higher or same priority
    continue;
  }
  if (containerSignatureMatcher.isExactMatch(
      highestPriRequest.getCookie().getContainerSignature(), signature)) {
    // exact match with different priorities
    continue;
  }
  // ... track the lowest running priority as the victim class
}

Note the signature-match escape: if the running lower-priority container is compatible with the starved request, killing it is pointless — the request can reuse it when the task finishes. Victims are then the newest containers at the lowest running priority, count scaled by scaleDownByPreemptionPercentage(...), and each is preempted via getContext().preemptContainer(cId) — which routes through TaskSchedulerManager.preemptContainer back into deallocateTask/deallocateContainer and surfaces to the attempt as a KILLED end state (see failure-handling.md). Releasing new containers instead of reusing them has a subtle RM-accounting consequence the code documents with a reference to TEZ-915 (real JIRA — the fix that requeues requests after such a release).

The DAG-aware version replaces "lowest priority loses" with "descendants lose". Its guard clauses are a checklist of reasons not to kill:

// tez-dag/src/main/java/org/apache/tez/dag/app/rm/DagAwareYarnTaskScheduler.java
private Collection<ContainerId> maybePreempt(Resource freeResources) {
  if (preemptionPercentage == 0 ||
      numHeartbeats - lastPreemptionHeartbeat < numHeartbeatsBetweenPreemptions) {
    return null;
  }
  if (!requestTracker.isPreemptionDeadlineExpired() &&
      requestTracker.fitsHighestPriorityRequest(freeResources)) {
    return null;      // RM should satisfy us; be patient
  }
  int numIdleContainers = idleTracker.getNumContainers();
  if (numIdleContainers > 0) {
    return null;      // reuse loop will find these first
  }
  BitSet blocked = requestTracker.createVertexBlockedSet();
  if (!blocked.intersects(assignedVertices)) {
    return null;      // nothing running is a descendant of the starved vertices
  }
  Resource preemptLeft = requestTracker.getAmountToPreempt(preemptionPercentage);
  // ... poll candidates ordered by PREEMPT_ORDER_COMPARATOR, newest first
}

Only containers assigned to vertices that are DAG-descendants of the starved requests are candidates — preempting anything else could never unblock the requester and would just burn work.

flowchart TD
  HB[RM heartbeat] --> P0{preemption.percentage == 0<br/>or too few heartbeats since last round?}
  P0 -- yes --> DONE[no preemption]
  P0 -- no --> FIT{highest-priority pending request<br/>fits free resources<br/>and deadline not expired?}
  FIT -- yes --> DONE
  FIT -- no --> IDLE{idle containers held?}
  IDLE -- yes --> DONE
  IDLE -- no --> DESC{descendants of starved vertices<br/>currently running?}
  DESC -- no --> DONE
  DESC -- yes --> AMT[compute amount to preempt<br/>scaled by preemption.percentage]
  AMT --> KILL[kill newest matching containers<br/>via ctx.preemptContainer]
  KILL --> WAIT[wait heartbeats-between-preemptions]
  WAIT --> HB

Warning: tez.am.preemption.percentage: 0 silently disables internal preemption in both schedulers. In a session AM running DAGs with wide priority spreads, that can convert a transient resource shortage into a permanent hang: descendants hold every container while ancestors starve. The 60-second preemption.max.wait-time-ms deadline exists because the RM sometimes reports headroom it cannot actually place on any single node.

Reading exercise

Run each command from the Tez checkout root and answer from the code:

# 1. The mediator's full inbound vocabulary and outbound handlers
grep -n "case S_" tez-dag/src/main/java/org/apache/tez/dag/app/rm/TaskSchedulerManager.java

# 2. Where the appCookie round-trips: launch request in, taskAllocated out
grep -n "appCookie\|clientCookie" \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/TaskSchedulerManager.java | head

# 3. The classic scheduler's three assigners and both matching modes
grep -n "class NodeLocalContainerAssigner\|class RackLocalContainerAssigner\|class NonLocalContainerAssigner\|getMatchingRequestWithPriority\|getMatchingRequestWithoutPriority" \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/YarnTaskSchedulerService.java

# 4. The DAG-aware tracker's blocking machinery
grep -n "class RequestTracker\|allowedVertices\|createVertexBlockedSet\|getAmountToPreempt" \
  tez-dag/src/main/java/org/apache/tez/dag/app/rm/DagAwareYarnTaskScheduler.java

# 5. Priority windows in both DAG schedulers
grep -n "getPriorityLowLimit\|getPriorityHighLimit" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGSchedulerNaturalOrder.java \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGSchedulerNaturalOrderControlled.java

Then: sketch, on paper, what happens when a running DAG has vertex M (maps, priority window ~8) fully allocated and vertex R (reducers, window ~20) submits 500 requests — which guard in each scheduler stops R's requests from consuming M's freed containers while M still has retries pending? Which BitSet answers that question in DagAwareYarnTaskScheduler, and what answers it (or fails to) in YarnTaskSchedulerService?

Common bugs and symptoms

Validation: prove you understand this

1. Walk an attempt from TaskAttemptEventSchedule to a running container: name the event, the manager method, the plugin method, the RM callback, and the upcall — with the file for each.
2. tez.am.yarn.scheduler.class vs tez.am.dag.scheduler.class: what does each select, what are the exact defaults on master, and which one changes who gets preempted?
3. For a diamond DAG A → {B, C} → D with vertex IDs 0–3 in that order, compute each vertex's priority window and the priority of a normal and a rescheduled attempt of C.
4. Why does YarnTaskSchedulerService refuse to give a new container to a task whose priority is lower than the container's, while allowing it for a reused container? What deadlock does the refusal prevent?
5. List the four guard conditions that make DagAwareYarnTaskScheduler.maybePreempt return null without killing anything, in order.
6. A container finishes a task and no node-local request exists. Describe the next 750 ms of its life under default configuration in each scheduler.
7. Your team's plugin scheduler must report cluster size and support blacklisting. Which abstract methods must it implement, and through which object does it deliver a container back to the AM?