DAGClient

DAGClient is the read-only handle to a submitted DAG. TezClient.submitDAG returns one; your driver, the Tez CLI, and Hive's TezJobMonitor all poll it until the DAG reaches a terminal state. This chapter dissects the real implementation — DAGClientImpl and the two backends it delegates to (DAGClientRPCImpl for a live AM, DAGClientTimelineImpl for a dead one) — plus the long-poll getDAGStatus, the status caching layer, and the single mutating call, tryKillDAG.

After this chapter you should be able to tell, for any DAGClient instance, which backend answered a given call, which fields will be populated, and why a status can flip between sources (AM → RM → Timeline) as a job winds down.

Prerequisite: tez-client.md (how you get here) and dag-model.md (what state rolls up). See also dag-app-master.md for the server side of these RPCs and Lab 3.1.

Files to open

ls tez-api/src/main/java/org/apache/tez/dag/api/client/
ls tez-api/src/main/java/org/apache/tez/dag/api/client/rpc/

tez-api/src/main/java/org/apache/tez/dag/api/client/
  DAGClient.java              (abstract public API)
  DAGClientImpl.java          (~27 KB: the real orchestrator, backend switching, caching)
  DAGClientInternal.java      (the backend interface DAGClientImpl delegates to)
  DAGClientTimelineImpl.java  (ATS/Timeline backend, used after the AM is gone)
  DAGClientImplLocal.java     (local mode: reads status from in-JVM functions)
  DAGStatus.java              (snapshot type; State enum)
  VertexStatus.java           (per-vertex snapshot; State enum)
  Progress.java               (task counts)
  StatusGetOpts.java          (GET_COUNTERS, GET_MEMORY_USAGE)
  DagStatusSource.java        (AM, RM, TIMELINE)
  TimelineReaderFactory.java  (ATS REST client construction)
  rpc/
    DAGClientRPCImpl.java              (talks to the AM via DAGClientAMProtocol)
    DAGClientAMProtocolBlockingPB.java (the protobuf proxy interface)

Note: DAGClientTimelineImpl is alive and well on master — it did not get deleted the way some older ATS classes did. It is the realClient that DAGClientImpl.switchToTimelineClient() swaps in once the DAG has completed and ATS is enabled. There is no separate "timeline history with FS" module class for this; the backend lives in tez-api.

The public contract

grep -n "public abstract" tez-api/src/main/java/org/apache/tez/dag/api/client/DAGClient.java

// tez-api: org.apache.tez.dag.api.client.DAGClient
public abstract class DAGClient implements Closeable {
  public abstract String getExecutionContext();
  public abstract DAGStatus getDAGStatus(@Nullable Set<StatusGetOpts> statusOptions)
      throws IOException, TezException;
  public abstract DAGStatus getDAGStatus(@Nullable Set<StatusGetOpts> statusOptions,
      long timeout) throws IOException, TezException;
  public abstract VertexStatus getVertexStatus(String vertexName,
      Set<StatusGetOpts> statusOptions) throws IOException, TezException;
  public abstract String getDagIdentifierString();
  public abstract String getSessionIdentifierString();
  public abstract void tryKillDAG() throws IOException, TezException;
  public abstract DAGStatus waitForCompletion() throws IOException, TezException, InterruptedException;
  public abstract DAGStatus waitForCompletionWithStatusUpdates(
      @Nullable Set<StatusGetOpts> statusGetOpts) throws IOException, TezException, InterruptedException;
  public abstract String getWebUIAddress() throws IOException, TezException;
}

Two things to internalize: there is exactly one mutating method (tryKillDAG), and getDAGStatus comes in a fire-once form (timeout absent) and a long-poll form (timeout in millis). StatusGetOpts has exactly two values:

// tez-api: org.apache.tez.dag.api.client.StatusGetOpts
public enum StatusGetOpts {
  GET_COUNTERS,
  GET_MEMORY_USAGE
}

Both are opt-in because both are expensive: GET_COUNTERS makes the AM serialize the whole TezCounters tree into the response (see counters-diagnostics.md); GET_MEMORY_USAGE aggregates across containers. Never request either in a tight poll loop.

DAGStatus — what callers actually consume

grep -n "enum State" -A 10 tez-api/src/main/java/org/apache/tez/dag/api/client/DAGStatus.java

// tez-api: org.apache.tez.dag.api.client.DAGStatus
public enum State {
  SUBMITTED, // Returned from the RM only
  INITING,   // This is currently never returned. DAG_INITING is treated as RUNNING.
  RUNNING,
  SUCCEEDED,
  KILLED,
  FAILED,
  ERROR,
}

The terminal subset is {SUCCEEDED, KILLED, FAILED, ERROR} — isCompleted() returns true for exactly these. Two states carry a warning label in the source itself: SUBMITTED only ever comes from the RM (the AM hasn't answered yet), and INITING is never returned because the AM maps its internal DAG_INITING onto RUNNING. This is not trivia — the backend-switching logic below keys off both the state and the DagStatusSource.

Fields you will read in production triage:

DAGStatus.State is a roll-up; VertexStatus.State is richer — NEW, INITIALIZING, INITED, RUNNING, COMMITTING, SUCCEEDED, FAILED, KILLED, ERROR, TERMINATING (VertexStatus.State). A DAG is RUNNING while individual vertices are COMMITTING; see vertex-lifecycle.md.

Progress and per-vertex status

The progress field is a Progress object with counts, not a percentage:

grep -n "public int get" tez-api/src/main/java/org/apache/tez/dag/api/client/Progress.java

// tez-api: org.apache.tez.dag.api.client.Progress
public int getTotalTaskCount();
public int getSucceededTaskCount();
public int getRunningTaskCount();
public int getFailedTaskCount();
public int getKilledTaskCount();
public int getFailedTaskAttemptCount();     // attempts, not tasks
public int getKilledTaskAttemptCount();
public int getRejectedTaskAttemptCount();

Note the distinction between task counts (a task is done when one attempt succeeds) and task-attempt counts (retries, speculation, rejections) — the attempt counters are how you spot a vertex that is succeeding but thrashing. You build a percentage yourself: succeeded / total.

getVertexStatus(vertexName, opts) returns a VertexStatus with its own State, Progress, diagnostics, and (if GET_COUNTERS) vertexCounters. On the wire it is VertexStatusProto{ id, state, diagnostics, progress, vertexCounters }. The retrieval mirrors the DAG path exactly — getVertexStatusInternal runs the same AM→cache→Timeline waterfall via getVertexStatusViaAM, caching per-vertex results in cachedVertexStatus so a dead AM does not lose the last-known per-vertex progress. There is no long-poll form for vertex status; it is always a point read.

DAGClientImpl — the orchestrator

DAGClientImpl is what submitDAG actually returns. It holds a realClient of type DAGClientInternal — initially a DAGClientRPCImpl — and swaps it for a DAGClientTimelineImpl when the DAG finishes and ATS is enabled.

grep -n "isATSEnabled\|realClient\|switchToTimelineClient\|cachedDAGStatusRef\|dagCompleted" \
  tez-api/src/main/java/org/apache/tez/dag/api/client/DAGClientImpl.java | head

The constructor decides up front whether ATS fallback is even possible:

// tez-api: DAGClientImpl constructor (trimmed)
isATSEnabled = conf.get(TezConfiguration.TEZ_HISTORY_LOGGING_SERVICE_CLASS, "")
    .equals("org.apache.tez.dag.history.logging.ats.ATSHistoryLoggingService")
    && conf.getBoolean(TezConfiguration.TEZ_DAG_HISTORY_LOGGING_ENABLED,
        TezConfiguration.TEZ_DAG_HISTORY_LOGGING_ENABLED_DEFAULT)
    && conf.getBoolean(TezConfiguration.TEZ_AM_HISTORY_LOGGING_ENABLED,
        TezConfiguration.TEZ_AM_HISTORY_LOGGING_ENABLED_DEFAULT)
    && DAGClientTimelineImpl.isSupported();

realClient = new DAGClientRPCImpl(appId, dagId, conf, this.frameworkClient, ugi);
statusPollInterval = conf.getLong(
    TezConfiguration.TEZ_DAG_STATUS_POLLINTERVAL_MS,        // default 500
    TezConfiguration.TEZ_DAG_STATUS_POLLINTERVAL_MS_DEFAULT);
cachedDAGStatusRef = initCacheDAGRefFromConf(conf);        // TTL = TEZ_CLIENT_DAG_STATUS_CACHE_TIMEOUT_SECS, default 60

So isATSEnabled is true only when the history service is specifically ATSHistoryLoggingService, DAG-level and AM-level history logging are both on, and the Timeline client library is present. Miss any of those and a completed DAG's status falls back to the RM instead of ATS.

getDAGStatusInternal — the source waterfall

The single most important method to understand. Every status read funnels through it, and it implements a strict preference order: live AM → cached → Timeline (if completed) → RM.

// tez-api: DAGClientImpl.getDAGStatusInternal (trimmed)
protected DAGStatus getDAGStatusInternal(Set<StatusGetOpts> statusOptions, long timeout) {
  if (!dagCompleted) {
    // dagCompleted may be flipped inside getDAGStatusViaAM
    final DAGStatus dagStatus = getDAGStatusViaAM(statusOptions, timeout);
    if (!dagCompleted) {
      if (dagStatus != null) {
        cachedDAGStatusRef.setValue(dagStatus);   // refresh cache from the AM
        return dagStatus;
      }
      DAGStatus cachedDAG = cachedDAGStatusRef.getValue();
      if (cachedDAG != null) {
        return cachedDAG;                         // AM unreachable -> serve cache, don't reset progress
      }
    }
    if (isATSEnabled && dagCompleted) {
      switchToTimelineClient();                   // replace realClient with DAGClientTimelineImpl
    }
  }
  if (isATSEnabled && dagCompleted) {
    try {
      DAGStatus dagStatus = realClient.getDAGStatus(statusOptions);
      if (dagStatus.isCompleted()) { return dagStatus; }
    } catch (ApplicationNotFoundException e) { /* app aged out of YARN */ }
      catch (TezException e) { /* fall through */ }
  }
  if (dagCompleted) {
    DAGStatus cachedDag = cachedDAGStatusRef.getValue();
    if (cachedDag != null && cachedDag.isCompleted()) { return cachedDag; }
  }
  return getDAGStatusViaRM();   // last resort
}

getDAGStatusViaAM is where dagCompleted gets set: a DAGNotRunningException or ApplicationNotFoundException from the AM means "the DAG is over, stop asking the AM." A NoCurrentDAGException is handled specially — if DAG recovery is enabled it is tolerated (the AM may be restarting), otherwise the client gives up and returns a synthetic failed status via dagLost().

flowchart TD
    A["getDAGStatusInternal"] --> B{"dagCompleted?"}
    B -->|no| C["getDAGStatusViaAM(timeout)"]
    C --> D{"AM answered?"}
    D -->|yes| E["cache it, return"]
    D -->|"no, DAG still live"| F["return cached status\n(don't reset progress)"]
    D -->|"DAGNotRunning /\nAppNotFound"| G["dagCompleted = true"]
    G --> H{"isATSEnabled?"}
    B -->|yes| H
    H -->|yes| I["switchToTimelineClient\nquery ATS; if completed, return"]
    H -->|no| J["cached completed status?"]
    I --> J
    J -->|no| K["getDAGStatusViaRM\n(last resort)"]

The long-poll form

When you pass a positive timeout, DAGClientImpl.getDAGStatus(opts, timeout) does not simply sleep and re-ask; it drives a state-aware loop:

// tez-api: DAGClientImpl.getDAGStatus(opts, timeout) (trimmed)
if (dagStatus.getState() == DAGStatus.State.RUNNING) {
  if (dagStatus.getSource() == DagStatusSource.AM) {
    // RUNNING from the AM already reflects the long-poll wait server-side
    return dagStatus;
  }
  // RUNNING from the RM: fall through to a client-side sleep
} else if (dagStatus.getState() == DAGStatus.State.SUCCEEDED
    || dagStatus.getState() == DAGStatus.State.FAILED
    || dagStatus.getState() == DAGStatus.State.KILLED
    || dagStatus.getState() == DAGStatus.State.ERROR) {
  if (dagStatus.getSource() == DagStatusSource.RM) {
    return getDAGStatusInternal(statusOptions, 0);  // RM's terminal state is coarse; ask a better source
  }
  return dagStatus;
}
// sleep min(statusPollInterval, remainingTimeout) and retry

The client trusts an AM RUNNING immediately (the AM already blocked on its side), but treats an RM RUNNING or an RM terminal state as low quality and either sleeps or re-queries a better source. This is why terminal states from a crashed AM still resolve to something informative.

Server-side long-poll

The timeout is honored inside the AM, not just on the client. The chain is DAGClientRPCImpl.getDAGStatus(opts, timeout) → RPC GetDAGStatusRequestProto{ dagId, statusOptions, timeout } → DAGClientHandler.getDAGStatus(dagId, opts, timeout) → DAGImpl.getDAGStatus(opts, timeoutMillis):

// tez-dag: DAGImpl.getDAGStatus(statusOptions, timeoutMillis) (trimmed)
long timeoutNanos = timeoutMillis * 1000L * 1000L;
if (timeoutMillis < 0)  { timeoutNanos = Long.MAX_VALUE; }      // wait until a terminal/RUNNING change
if (timeoutMillis == 0 || isComplete()) { return getDAGStatus(statusOptions); }
while (true) {
  dagStatusLock.lock();
  try {
    if (isFinalState.get()) { break; }
    if (runningStatusYetToBeConsumed.compareAndSet(true, false)) { break; }  // just became RUNNING
    nanosLeft = dagStateChangedCondition.awaitNanos(timeoutNanos);
  } finally { dagStatusLock.unlock(); }
  if (nanosLeft <= 0) { break; }
  timeoutNanos = nanosLeft;
}
return getDAGStatus(statusOptions);

The AM parks the RPC handler thread on a Condition and wakes it only when the DAG changes state or the client's timeout elapses. That is what makes waitForCompletion() cheap: a timeout of -1 becomes Long.MAX_VALUE nanos, so the client blocks server-side until the DAG is actually done rather than busy-polling. Contrast this with an RM-sourced status, which has no such push and must be polled at tez.dag.status.pollinterval-ms (default 500).

waitForCompletion and friends

grep -n "waitForCompletion\|_waitForCompletionWithStatusUpdates" \
  tez-api/src/main/java/org/apache/tez/dag/api/client/DAGClientImpl.java

• waitForCompletion() → _waitForCompletionWithStatusUpdates(-1, false, none) — block until terminal, no per-vertex printing.
• waitForCompletionWithStatusUpdates(opts) → _waitForCompletionWithStatusUpdates(-1, true, opts) — same, but it prints progress lines (throttled by PRINT_STATUS_INTERVAL_MILLIS, 5000 ms) and attaches the requested counters. This is what the Tez CLI and TezExampleBase.runDag use so you see vertex progress on the console.

Both ultimately loop on the long-poll getDAGStatus, so they inherit the source waterfall and the cache.

tryKillDAG — the only mutation

grep -n "public void tryKillDAG" -A 8 tez-api/src/main/java/org/apache/tez/dag/api/client/DAGClientImpl.java

// tez-api: DAGClientImpl.tryKillDAG
public void tryKillDAG() throws IOException, TezException {
  if (!dagCompleted) {
    realClient.tryKillDAG();
  } else {
    LOG.info("TryKill for app: " + appId + " dag:" + dagId + " dag already completed.");
  }
}

It is fire-and-forget: the RPC (tryKillDAG → AM DAGAppMaster.tryKillDAG → DAGEventTerminateDag with cause DAG_KILL) only initiates the kill; tasks must drain. When the client is waiting for a terminal status to collect diagnostics, it bounds that wait with tez.client.diagnostics.wait.timeout-ms (TEZ_CLIENT_DIAGNOSTICS_WAIT_TIMEOUT_MS, default 3000 ms) so a kill that never produces a clean diagnostic string does not hang the caller forever. Always follow the kill with a wait:

client.tryKillDAG();
DAGStatus status = client.waitForCompletion();   // resolves to KILLED (or whatever it raced to)

If the DAG already finished, the client short-circuits and does not even make the RPC — killing a dead DAG is a no-op, not an error.

Local mode: DAGClientImplLocal

In local mode there is no RPC. DAGClientImplLocal extends DAGClientImpl but is constructed with two BiFunctions that read status straight out of the in-JVM AM:

// tez-api: DAGClientImplLocal (trimmed)
public DAGClientImplLocal(ApplicationId appId, String dagId, TezConfiguration conf,
    FrameworkClient frameworkClient, UserGroupInformation ugi,
    BiFunction<Set<StatusGetOpts>, Long, DAGStatus> dagStatusFunction,
    BiFunction<Set<StatusGetOpts>, String, VertexStatus> vertexStatusFunction) {
  super(appId, dagId, conf, frameworkClient, ugi);
  this.dagStatusFunction = dagStatusFunction;
  this.vertexStatusFunction = vertexStatusFunction;
}

Same public API, zero network, no ATS fallback — see local-mode.md.

How CLI, examples, and Hive consume it

The whole point of the source waterfall and the long-poll is that callers get a dead-simple loop. The tez-examples base class is the canonical pattern:

// tez-examples: TezExampleBase.runDag (trimmed)
DAGClient dagClient = tezClient.submitDAG(dag);
Set<StatusGetOpts> getOpts = EnumSet.noneOf(StatusGetOpts.class);
if (printCounters) { getOpts = EnumSet.of(StatusGetOpts.GET_COUNTERS); }
DAGStatus dagStatus = dagClient.waitForCompletionWithStatusUpdates(getOpts);
if (dagStatus.getState() != DAGStatus.State.SUCCEEDED) {
  // print dagStatus.getDiagnostics(), return failure
}

That single waitForCompletionWithStatusUpdates call blocks server-side (via the AM Condition), prints throttled progress lines, and — because getOpts was chosen once, up front — pays the counter-serialization cost exactly once at the end rather than on every poll. The Tez CLI (tez-tools) wraps the same call behind tez dag -status <dagId>.

Hive's TezJobMonitor is the more demanding consumer: it wants live per-vertex progress for its console bar, so instead of blocking on waitForCompletion it polls getDAGStatus on an interval and reads the Progress per vertex (getVertexStatus), rendering succeeded/total per stage. It requests GET_COUNTERS only at the end. When the AM finally exits, the same handle transparently starts answering from ATS (if isATSEnabled) so Hive's post-job summary still has counters. Neither consumer knows or cares which backend answered — that is the abstraction DAGClientImpl buys them.

Reading exercise

# Public surface + State enum
sed -n '1,120p' tez-api/src/main/java/org/apache/tez/dag/api/client/DAGClient.java
grep -n "enum State" -A 10 tez-api/src/main/java/org/apache/tez/dag/api/client/DAGStatus.java

# The source waterfall
grep -n "getDAGStatusInternal\|getDAGStatusViaAM\|getDAGStatusViaRM\|switchToTimelineClient" \
  tez-api/src/main/java/org/apache/tez/dag/api/client/DAGClientImpl.java

# Server-side long-poll
grep -n "getDAGStatus(Set<StatusGetOpts> statusOptions,\s*$\|awaitNanos\|runningStatusYetToBeConsumed" \
  tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/DAGImpl.java

Answer, with citations:

1. What is the difference between waitForCompletion() and waitForCompletionWithStatusUpdates(opts) — in RPC traffic and in console output?
2. Which exact conditions must all be true for isATSEnabled to be set, and what happens to completed-DAG status when one of them is false?
3. List the seven DAGStatus.State values and identify the terminal subset. Which two states have comments warning they are "special"?
4. In the long-poll loop, why does the client return immediately for a RUNNING+AM status but sleep for RUNNING+RM?
5. What does DAGImpl.getDAGStatus(opts, -1) do internally, and why does that make waitForCompletion() cheap on the cluster?
6. After tryKillDAG, what value does a subsequent getDAGStatus eventually report, and via which source once the AM exits?

Common bugs and symptoms

Validation: prove you understand this

1. Write a ~20-line program that polls getDAGStatus(EnumSet.of(GET_COUNTERS)) once a second and prints one framework counter per snapshot; note how the payload size grows.
2. Draw the construction/factory path from TezClient.submitDAG to a concrete DAGClient subclass in (a) YARN session, (b) YARN non-session, and (c) local mode.
3. From getDAGStatusInternal, enumerate every branch that can return, and for each name the DagStatusSource it came from.
4. Kill the AM mid-DAG (e.g. yarn application -kill on a MiniCluster) and confirm from logs whether the client falls back to ATS or RM, and cite the line that flips dagCompleted.
5. Explain why DAGStatus is a snapshot rather than an observable, and what a caller must do to build a live progress bar on top of it.