TezClient

TezClient is the client-side API: the class your driver code instantiates to launch a Tez ApplicationMaster, submit DAGs to it, and — in session mode — keep that AM alive across many DAGs. It is a ~1,300-line orchestrator in tez-api that hides three genuinely different submission paths behind one façade: session-over-RPC, non-session-new-application, and local (in-process). This chapter walks the bring-up, the mode split, the AM launch context, prewarm, and the RPC that carries a DAGPlan to a running AM.

After this chapter you should be able to point at every method that runs between TezClient.create(...) and the moment a DAGClient handle comes back, name the config key behind each timeout, and explain why the same submitDAG call reaches YARN differently depending on mode.

Prerequisite: dag-model.md (what a DAG is). Sequel: dag-client.md (the handle you get back) and dag-app-master.md (what receives the plan). Hands-on: Lab 3.1: trace a DAG submission.

Files to open

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

tez-api/src/main/java/org/apache/tez/client/
  TezClient.java            (the façade + session/non-session logic)
  TezClientUtils.java       (~46 KB: builds the ApplicationSubmissionContext)
  FrameworkClient.java      (abstract: submitApplication + submitDag + getProxy)
  TezYarnClient.java        (YARN-backed FrameworkClient, wraps YarnClient)
  AMConfiguration.java      (AM resources, credentials, tez conf)
tez-dag/src/main/java/org/apache/tez/client/
  LocalClient.java          (in-process FrameworkClient for local mode)

Plus the AM protocol and the proto:

tez-api/src/main/java/org/apache/tez/dag/api/client/rpc/DAGClientAMProtocolBlockingPB.java
tez-api/src/main/proto/DAGClientAMProtocol.proto

Two modes: session and non-session

The mode is fixed at construction and stored as isSession:

grep -n "public static TezClient create\|isSession\|TEZ_AM_SESSION_MODE" \
  tez-api/src/main/java/org/apache/tez/client/TezClient.java | head

// Explicit:
TezClient client = TezClient.create("MyApp", tezConf, /* isSession */ true);
// Or driven by config, TezConfiguration:
//   tez.am.mode.session  (TEZ_AM_SESSION_MODE), default false

Property	Non-session	Session
AM lifetime	One DAG, then AM exits	Many DAGs across the AM's life
`start()`	Effectively deferred; app is submitted at `submitDAG`	Submits the YARN application immediately, launching the AM
DAGs in flight	1 (the app is the DAG)	1 at a time; the AM rejects a second concurrent DAG
Idle shutdown	n/a	`tez.session.am.dag.submit.timeout.secs`, default 300
Use case	One-shot batch, CLI tools	Latency-sensitive engines (Hive, Pig) that reuse containers

The idle-shutdown timer is the single most surprising session behavior. After a DAG completes, the AM waits up to TEZ_SESSION_AM_DAG_SUBMIT_TIMEOUT_SECS for the next DAG; if none arrives it shuts down to release YARN resources. The enforcement is in the AM, not the client:

grep -n "checkAndHandleSessionTimeout\|sessionTimeoutInterval\|getDAGSessionTimeout" \
  tez-dag/src/main/java/org/apache/tez/dag/app/DAGAppMaster.java

// tez-dag: DAGAppMaster.checkAndHandleSessionTimeout (trimmed)
if (EnumSet.of(DAGAppMasterState.RUNNING, DAGAppMasterState.RECOVERING).contains(this.state)
    || sessionStopped.get()) {
  return;                                   // a DAG is running -> cannot time out
}
if (currentTime < (lastDAGCompletionTime + sessionTimeoutInterval)) {
  return;                                   // still within the grace window
}
String message = "Session timed out"
    + ", lastDAGCompletionTime=" + lastDAGCompletionTime + " ms"
    + ", sessionTimeoutInterval=" + sessionTimeoutInterval + " ms";
addDiagnostic(message);
shutdownTezAM(message);

A negative value (TezCommonUtils.getDAGSessionTimeout returns -1) disables the check and lets the AM idle forever — useful for a long-lived Hive HiveServer2 session, dangerous on a shared queue.

`start()` — what actually happens

grep -n "public synchronized void start()" -A 25 \
  tez-api/src/main/java/org/apache/tez/client/TezClient.java

// tez-api: TezClient.start (trimmed)
public synchronized void start() throws TezException, IOException {
  startFrameworkClient();       // init+start the FrameworkClient (YARN or Local)
  setupJavaOptsChecker();
  if (isSession) {
    LOG.info("Session mode. Starting session.");
    // ...
    clientTimeout = amConfig.getTezConfiguration().getInt(
        TezConfiguration.TEZ_SESSION_CLIENT_TIMEOUT_SECS,        // default 120
        TezConfiguration.TEZ_SESSION_CLIENT_TIMEOUT_SECS_DEFAULT);
    if (sessionAppId == null) {
      sessionAppId = createApplication();                       // RM: getNewApplication
    }
    ApplicationSubmissionContext appContext = setupApplicationContext();
    frameworkClient.submitApplication(appContext);              // RM: submitApplication -> AM launches
    ApplicationReport appReport = frameworkClient.getApplicationReport(sessionAppId);
    LOG.info("The url to track the Tez Session: " + appReport.getTrackingUrl());
    sessionStarted.set(true);
    startClientHeartbeat();                                     // AM keep-alive pinger
    this.stagingFs = FileSystem.get(amConfig.getTezConfiguration());
  }
}

The asymmetry is the whole point: in session mode start() launches the AM container; in non-session mode start() only spins up the framework client, and the YARN application is not created until submitDAG builds it around a specific DAG. If session bring-up throws, cleanStagingDir() removes the staged jars/conf so a retry starts clean.

startClientHeartbeat() deserves attention. When tez.am.client.heartbeat.timeout.secs (TEZ_AM_CLIENT_HEARTBEAT_TIMEOUT_SECS, default -1 = disabled) is positive, the client schedules a daemon AMKeepAliveThread that pings the AM. This is the reverse liveness channel: it lets the AM notice a dead client and shut itself down rather than idling on the queue. It is skipped entirely in local mode.

Waiting for readiness

Two waits exist, and confusing them causes real hangs:

grep -n "waitTillReady\|waitNonSessionTillReady\|TezAppMasterStatus" \
  tez-api/src/main/java/org/apache/tez/client/TezClient.java | head

waitTillReady() (session only; no-op non-session) polls getAppMasterStatus() until it returns TezAppMasterStatus.READY, throwing SessionNotRunning if the AM reports SHUTDOWN. Statuses are INITIALIZING, READY, RUNNING, SHUTDOWN (org.apache.tez.client.TezAppMasterStatus).
waitNonSessionTillReady() (private, non-session) blocks after submitDAGApplication until the AM is RUNNING or SHUTDOWN, so that a subsequent getDAGStatus has something to talk to.

Both sleep in SLEEP_FOR_READY increments. waitTillReady(timeout, unit) returns false on expiry rather than throwing, which is the API a scheduler should use to bound its own startup.

Submission: three paths, one method

grep -n "public synchronized DAGClient submitDAG\|submitDAGSession\|submitDAGApplication" \
  tez-api/src/main/java/org/apache/tez/client/TezClient.java

// tez-api: TezClient.submitDAG
public synchronized DAGClient submitDAG(DAG dag) throws TezException, IOException {
  DAGClient result = isSession ? submitDAGSession(dag) : submitDAGApplication(dag);
  if (result != null) {
    closePrewarmDagClient();   // the real DAG supersedes any prewarm DAG
  }
  return result;
}

Session path — DAG over RPC

submitDAGSession is the hot path for Hive/Pig. It:

verifySessionStateForSubmission() — throws SessionNotRunning if the session was never started or was stopped.
Builds the DAGPlan via TezClientUtils.prepareAndCreateDAGPlan(...) (which calls DAG.createDag from dag-model.md).
Wraps it in a SubmitDAGRequestProto and hands off to frameworkClient.submitDag(...).

The subtle production detail is the IPC size guard:

// tez-api: TezClient.submitDAGSession (trimmed)
SubmitDAGRequestProto request = requestBuilder.build();
if (request.getSerializedSize() > maxSubmitDAGRequestSizeThroughIPC) {
  Path dagPlanPath = new Path(TezCommonUtils.getTezSystemStagingPath(...),
      TezConstants.TEZ_PB_PLAN_BINARY_NAME + serializedSubmitDAGPlanRequestCounter.incrementAndGet());
  try (FSDataOutputStream out = fs.create(dagPlanPath, false)) {
    LOG.info("Send dag plan using YARN local resources since it's too large ...");
    request.writeTo(out);
  }
  request = requestBuilder.clear()
      .setSerializedRequestPath(fs.resolvePath(dagPlanPath).toString()).build();
}
return frameworkClient.submitDag(dag, request, clientName, sessionAppId, clientTimeout, ...);

If the serialized plan exceeds the Hadoop IPC max (ipc.maximum.data.length), the client writes the whole request to the staging dir and sends only the path in serializedRequestPath (proto field 3). This is why an enormous UserPayload (see dag-model.md) does not simply blow up submission — but it does turn every submit into an HDFS round-trip.

FrameworkClient.submitDag then acquires the AM proxy with a bounded wait and makes the actual call:

// tez-api: FrameworkClient.submitDag (trimmed)
proxy = waitForProxy(clientTimeout, tezConf, sessionAppId, ugi);
if (proxy == null) {
  stop();
  throw new DAGSubmissionTimedOut("Could not submit DAG to Tez Session"
      + ", timed out after " + clientTimeout + " seconds");
}
SubmitDAGResponseProto response = proxy.submitDAG(null, request);
dagId = response.getDagId();
return getDAGClient(sessionAppId, dagId, tezConf, ugi);

Non-session path — DAG baked into the application

submitDAGApplication builds a fresh YARN application whose DAGPlan is staged as a local resource, so the AM finds it on disk at startup and never needs a submitDAG RPC:

// tez-api: TezClient.submitDAGApplication (trimmed)
ApplicationSubmissionContext appContext = TezClientUtils.createApplicationSubmissionContext(
    appId, dag, dag.getName(), amConfig, tezJarResources, credentials,
    usingTezArchiveDeploy, apiVersionInfo, servicePluginsDescriptor, javaOptsChecker);
frameworkClient.submitApplication(appContext);
// ...
waitNonSessionTillReady();
return getDAGClient(appId, amConfig.getTezConfiguration(), frameworkClient, getUgi());

Inside createApplicationSubmissionContext, when dag != null, the plan is written to TezConstants.TEZ_PB_PLAN_BINARY_NAME (tez-dag.pb) and added to amLocalResources. On the AM side, DAGAppMaster in non-session mode reads exactly this file (readDAGPlanFile()), whereas a session AM logs "In Session mode. Waiting for DAG over RPC" and enters IDLE. That single if (dag == null) branch — which also appends the --session CLI option to the AM launch command — is the seam between the two worlds.

sequenceDiagram
    participant U as User code
    participant TC as TezClient
    participant TCU as TezClientUtils
    participant FC as FrameworkClient
    participant RM as YARN RM
    participant AM as DAGAppMaster

    Note over U,AM: SESSION MODE
    U->>TC: create(name, conf, isSession=true)
    U->>TC: addAppMasterLocalFiles(map)
    U->>TC: start()
    TC->>FC: createApplication() (RM getNewApplication)
    TC->>TCU: setupApplicationContext()
    TCU->>TCU: stage tez libs + conf.pb + amResources.pb to HDFS
    TC->>FC: submitApplication(appContext)
    FC->>RM: submitApplication
    RM-->>AM: launch AM container (--session)
    AM->>AM: serviceInit/serviceStart -> IDLE
    U->>TC: waitTillReady()  (poll until READY)
    U->>TC: submitDAG(dag)
    TC->>TC: DAG.createDag -> DAGPlan; wrap in SubmitDAGRequestProto
    TC->>AM: proxy.submitDAG(request)
    AM-->>TC: dagId
    TC-->>U: DAGClient

The AM launch context: what `TezClientUtils` stages

grep -n "createApplicationSubmissionContext\|TEZ_PB_BINARY_CONF_NAME\|TEZ_AM_LOCAL_RESOURCES_PB_FILE_NAME\|setupTezJarsLocalResources" \
  tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java | head

A YARN container starts with an empty working directory plus whatever the ApplicationSubmissionContext localizes. For a Tez AM, TezClientUtils stages:

Tez framework jars — resolved from tez.lib.uris (TezConfiguration.TEZ_LIB_URIS) by setupTezJarsLocalResources, which returns a boolean telling the AM whether the libs are an archive (tarball) or a directory of jars. That flag becomes usingTezArchiveDeploy and rides into the plan.
Binary configuration — the client's TezConfiguration serialized to TezConstants.TEZ_PB_BINARY_CONF_NAME (tez-conf.pb) and localized so the AM boots with the same config the client saw.
The AM resource manifest — TEZ_AM_LOCAL_RESOURCES_PB_FILE_NAME, the PlanLocalResourcesProto listing what the AM already has, so session DAGs can send additional resources incrementally.
User AM jars — anything added via addAppMasterLocalFiles(Map<String, LocalResource>).
The DAGPlan — only in non-session mode, as tez-dag.pb. In session mode the plan travels in the submitDAG RPC instead.

// tez-api: TezClientUtils.createApplicationSubmissionContext (trimmed)
vargs.add(TezConstants.TEZ_APPLICATION_MASTER_CLASS);   // the AM main class
if (dag == null) {
  vargs.add("--" + TezConstants.TEZ_SESSION_MODE_CLI_OPTION);   // "--session"
}
// AM memory/vcores from TEZ_AM_RESOURCE_MEMORY_MB / TEZ_AM_RESOURCE_CPU_VCORES
// AM env from TEZ_AM_LAUNCH_ENV (+ cluster default env)
// credentials/tokens written into securityTokens ByteBuffer

Note: The AMRM token that lets the AM talk to the RM is injected by YARN when the container starts; Tez does not manage it. Delegation tokens for HDFS and for shuffle are Tez's responsibility and are gathered into the launch credentials here (createSessionToken, processTezLocalCredentialsFile).

Why addAppMasterLocalFiles is a Map<String, LocalResource> and not a List<Path>: YARN localization is keyed by the symlink name the container will see, and each resource needs a type/visibility/timestamp/size, all of which LocalResource carries and a bare Path does not.

Prewarm

grep -n "public synchronized void preWarm\|prewarmDagClient\|TEZ_PREWARM_DAG_NAME_PREFIX" \
  tez-api/src/main/java/org/apache/tez/client/TezClient.java

preWarm(PreWarmVertex) is a session-only optimization: it submits a synthetic one-vertex DAG whose only job is to make YARN allocate containers that later DAGs can reuse (container reuse is container-reuse.md).

// tez-api: TezClient.preWarm (trimmed)
if (!isSession) {
  LOG.warn("preWarm is not supported in non-session mode, please use session-mode of TezClient");
  return;
}
DAG dag = DAG.create(TezConstants.TEZ_PREWARM_DAG_NAME_PREFIX + "_" + preWarmDAGCounter++);
dag.addVertex(preWarmVertex);
if (waitTillReady(timeout, unit)) {
  prewarmDagClient = submitDAG(dag);
} else {
  throw new SessionNotReady("Tez AM not ready, could not submit DAG");
}

The prewarm vertex must be configured exactly like the real vertices (resources, environment) or its containers will not match the reuse signature. When the next real DAG is submitted, submitDAG calls closePrewarmDagClient (and stop() calls killAndClosePrewarmDagClient) so the throwaway DAG does not leak. The AM special-cases prewarm names so they do not count toward submittedDAGs.

FrameworkClient: YARN vs Local

FrameworkClient is the abstraction that lets the same TezClient code run against a real cluster or in-process:

// tez-api: FrameworkClient.createFrameworkClient (trimmed)
boolean isLocal = tezConf.getBoolean(TezConfiguration.TEZ_LOCAL_MODE,
    TezConfiguration.TEZ_LOCAL_MODE_DEFAULT);        // default false
if (isLocal) {
  return ReflectionUtils.createClazzInstance("org.apache.tez.client.LocalClient");
} else {
  ClientFrameworkService svc = FrameworkUtils.get(ClientFrameworkService.class, tezConf,
      YarnClientFrameworkService.class);
  return svc.newFrameworkClient();                   // -> TezYarnClient wrapping a YarnClient
}

TezYarnClient (tez-api) wraps a YARN YarnClient; submitApplication, getApplicationReport, and killApplication are thin delegations, and getProxy builds a protobuf proxy to the AM's DAGClientServer.
LocalClient (tez-dag, because it needs DAGAppMaster) starts the AM in the same JVM on a DAGAppMaster Thread (startDAGAppMaster/createDAGAppMaster). With tez.local.mode.without.network=true (TEZ_LOCAL_MODE_WITHOUT_NETWORK), it bypasses RPC and calls the AM's methods directly. Local mode is the subject of local-mode.md.

Because the choice is reflective and config-driven, your driver code is identical across all three deployment shapes; only tez.local.mode changes.

`stop()` — graceful, then forceful

grep -n "public synchronized void stop()\|shutdownSession\|killApplication\|TEZ_CLIENT_ASYNCHRONOUS_STOP\|TEZ_CLIENT_HARD_KILL_TIMEOUT_MS" \
  tez-api/src/main/java/org/apache/tez/client/TezClient.java

stop() is a two-tier shutdown. First it kills any prewarm DAG and the keep-alive thread, then — in session mode — it asks the AM to shut itself down politely via the shutdownSession RPC; only if that fails does it fall back to YARN's killApplication:

// tez-api: TezClient.stop (trimmed)
sessionStopped.set(true);
boolean sessionShutdownSuccessful = frameworkClient
    .shutdownSession(amConfig.getTezConfiguration(), sessionAppId, getUgi());
boolean asynchronousStop = conf.getBoolean(TezConfiguration.TEZ_CLIENT_ASYNCHRONOUS_STOP,
    TezConfiguration.TEZ_CLIENT_ASYNCHRONOUS_STOP_DEFAULT);      // default true
if (!asynchronousStop && sessionShutdownSuccessful) {
  // poll getApplicationReport until terminal, up to TEZ_CLIENT_HARD_KILL_TIMEOUT_MS
  // if still not terminal -> frameworkClient.killApplication(sessionAppId)
}
if (!sessionShutdownSuccessful) {
  LOG.info("Could not connect to AM, killing session via YARN ...");
  frameworkClient.killApplication(sessionAppId);
}

The tez.client.asynchronous.stop flag (default true) is why a normal stop() returns quickly: it fires the shutdown RPC and does not block waiting for the app to reach a terminal YARN state. Set it false (e.g. in tests, or a scheduler that must confirm teardown) and stop() will poll the RM up to tez.client.hard.kill.timeout.ms, then hard-kill via YARN. In non-session mode stop() is nearly a no-op — the AM already exits when its single DAG finishes — but it still closes the FrameworkClient in the finally block.

Reattaching to a running session

A session AM outlives the JVM that launched it, so a new process can reattach with TezClient.getClient(appIdStr) instead of start(). This is how a long-lived Hive session survives a client restart: the app id is persisted, and the fresh TezClient picks up the existing AM's staging dir and resource fingerprints rather than staging fresh jars. Grep "public synchronized TezClient getClient" in TezClient.java.

The submission RPC

grep -n "rpc " tez-api/src/main/proto/DAGClientAMProtocol.proto

// tez-api: DAGClientAMProtocol.proto
service DAGClientAMProtocol {
  rpc getAllDAGs        (GetAllDAGsRequestProto)        returns (GetAllDAGsResponseProto);
  rpc getDAGStatus      (GetDAGStatusRequestProto)      returns (GetDAGStatusResponseProto);
  rpc getVertexStatus   (GetVertexStatusRequestProto)   returns (GetVertexStatusResponseProto);
  rpc tryKillDAG        (TryKillDAGRequestProto)        returns (TryKillDAGResponseProto);
  rpc submitDAG         (SubmitDAGRequestProto)         returns (SubmitDAGResponseProto);
  rpc shutdownSession   (ShutdownSessionRequestProto)   returns (ShutdownSessionResponseProto);
  rpc getAMStatus       (GetAMStatusRequestProto)       returns (GetAMStatusResponseProto);
  rpc getWebUIAddress   (GetWebUIAddressRequestProto)   returns (GetWebUIAddressResponseProto);
}

message SubmitDAGRequestProto {
  optional DAGPlan d_a_g_plan = 1;
  optional PlanLocalResourcesProto additional_am_resources = 2;
  optional string serializedRequestPath = 3;      // used when the request is too big for IPC
}

The client-side Java interface is DAGClientAMProtocolBlockingPB; the AM-side server is DAGClientServer (a Hadoop RPC Server running DAGClientAMProtocol.newReflectiveBlockingService), which forwards to DAGClientHandler and thence to the DAGAppMaster. Both the reader (dag-client.md) and the writer (this chapter) speak the same protocol; only the direction of the interesting call differs.

Reading exercise

sed -n '1,120p' tez-api/src/main/java/org/apache/tez/client/TezClient.java
grep -n "submitDAG\b\|submitDAGSession\|submitDAGApplication" \
  tez-api/src/main/java/org/apache/tez/client/TezClient.java
grep -n "stop\b\|stopSession\|shutdownSession" \
  tez-api/src/main/java/org/apache/tez/client/TezClient.java
grep -n "createApplicationSubmissionContext" -A 5 \
  tez-api/src/main/java/org/apache/tez/client/TezClientUtils.java

Answer, with citations:

In session vs non-session mode, what does stop() do differently? (Trace killAndClosePrewarmDagClient and the shutdownSession RPC.)
When submitDAG is called while a DAG is already running in the session, which side rejects it and with what message? (Hint: it is the AM — submitDAGToAppMaster — not the client.)
Which config key bounds the wait for the AM proxy during submitDag, and what exception fires on expiry?
Trace addAppMasterLocalFiles(...) to where the files are localized. Which staging path holds them?
Why is tez.lib.uris sometimes a directory and sometimes a tarball, and what does the boolean return of setupTezJarsLocalResources change downstream?
In non-session mode, where does the AM read the DAG plan from, and why is there no submitDAG RPC on that path?

Common bugs and symptoms

Symptom	Root cause	Fix
Client hangs in `start()`; AM never registers	`tez.lib.uris` points to a path the NodeManager cannot read, or the archive is corrupt	Check NM container log; verify HDFS perms and the tarball
`SessionNotRunning` on `submitDAG`	AM died (idle timeout, crash, or client heartbeat lapse)	Catch it, recreate `TezClient`, resubmit; consider raising `tez.session.am.dag.submit.timeout.secs`
`DAGSubmissionTimedOut: Could not submit DAG ... timed out after N seconds`	AM proxy unreachable within `tez.session.client.timeout.secs` (default 120)	Verify AM is `READY`; check network to the AM host/port
`TezException: App master already running a DAG`	Second concurrent submit into one session	Serialize DAGs, or use separate sessions for parallelism
Submission slow, HDFS write per submit	`DAGPlan` exceeds `ipc.maximum.data.length`, triggering the `serializedRequestPath` fallback	Shrink `UserPayload`; move data to `LocalResource`s
Prewarm containers never reused	Prewarm vertex resources/env differ from real vertices	Match the reuse signature exactly; see container-reuse.md
Tasks fail `ClassNotFoundException` for user code	Jar added only to the AM, not to tasks	Add via both `addAppMasterLocalFiles` and `addTaskLocalFiles` (per-vertex `addTaskLocalFiles` also works)

Validation: prove you understand this

Write a ~30-line driver that creates a session TezClient, waitTillReady, submits two DAGs back-to-back, prints each DAGClient.getDAGStatus(), and stop()s cleanly. Then flip isSession to false and explain which lines change behavior.
From TezClient.java, list every method that ultimately reaches the AM proxy (DAGClientAMProtocolBlockingPB), directly or through FrameworkClient.
Set tez.local.mode=true and run the same driver in local mode; confirm from logs that no YARN application is created and the AM runs on a DAGAppMaster Thread.
Reproduce the idle-timeout path: submit one DAG in a session, sleep past tez.session.am.dag.submit.timeout.secs, submit a second DAG, and record the exact exception class and message.
Force the IPC-size fallback: build a DAG with a large UserPayload, submit it, and find the "Send dag plan using YARN local resources" log plus the staged tez-dag.pb... file.

Open-Source Engineer & Contributor