Stage 6 — Shuffle and Runtime

What this stage teaches

Stage 6 is the runtime stage. This is where Tez earns its performance reputation and where its subtlest bugs live: concurrent Fetcher threads, on-disk IFile segments, memory-vs-disk merge decisions, and event storms that back up the AM. The contracts here are enforced by threads and timers, not by types, so a wrong fix "passes the test" and then hangs a production DAG at 3am. You learn:

• The shuffle pipeline: how ShuffleScheduler/ShuffleManager schedule Fetcher threads against upstream task outputs, how MergeManager/TezMerger consolidate fetched segments, and how the result reaches the downstream processor.
• The on-disk IFile format and the off-by-one bugs that haunt every reader written against it.
• Fetch-failure handling: how a single bad NodeManager can flood the AM with read-error events, and how the runtime debounces and batches them.
• The two hardest bug classes in this module: concurrency (deadlocks and races between the scheduler, the penalty thread, and close()) and silent correctness (records dropped or duplicated with no exception).

Patches are 10–600 lines. The concurrency fixes are small diffs with large reasoning; the correctness fixes come with a targeted unit test that reproduces the exact record pattern that broke.

Prerequisite: Stage 5 plus the runtime deep dives: shuffle & sort, tez-runtime, and IPO abstractions. Read those — this stage assumes you know the Input/Processor/Output contract.

Reading order in the checkout

cd /Users/s0x/src/oss-repos/tez
wc -l tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/*.java
grep -rln "class ShuffleScheduler\|class MergeManager\|class TezMerger\|class IFile" \
  tez-runtime-library/src/main/java

Read, in order: ShuffleScheduler → FetcherOrderedGrouped → MergeManager → TezMerger → IFile. Then the deep dive.

Finding Stage 6 issues today

project = TEZ AND resolution = Unresolved
  AND component in ("Shuffle", "Fetcher", "tez-runtime-library")
  AND (summary ~ "shuffle" OR summary ~ "fetch" OR summary ~ "merge"
       OR summary ~ "sorter" OR summary ~ "IFile" OR summary ~ "spill")
ORDER BY priority DESC, updated DESC

Heuristics that beat JIRA search:

# concurrency hotspots: methods that lock 'this' in the scheduler
grep -n "synchronized" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/shuffle/orderedgrouped/ShuffleScheduler.java
# every place a fetch failure is constructed — candidate cause-loss sites
grep -rn "InputAttemptFetchFailure\|copyFailed\|reportException" \
  tez-runtime-library/src/main/java

The subsystem knowledge you need

The ordered shuffle pipeline, and where each case study lives in it:

flowchart LR
  UP[upstream task outputs<br/>IFile segments on disk] --> FET[FetcherOrderedGrouped<br/>TEZ-3761: NPE on bad header]
  FET --> SCHED[ShuffleScheduler<br/>TEZ-4334: close vs penalty deadlock<br/>TEZ-4336: preserve cause]
  SCHED --> MM[MergeManager / TezMerger<br/>memory-vs-disk, RLE]
  MM --> SORT[PipelinedSorter<br/>TEZ-3849: combiner drops records]
  SORT --> DOWN[downstream Processor<br/>KeyValuesReader]
  SCHED -.INPUT_READ_ERROR.-> AM[AppMaster<br/>TEZ-3976: batch + dedupe events]

Three facts underpin every fix in this stage:

• IFile is the on-disk record format. Each segment is a length-prefixed stream of key/value pairs terminated by an EOF marker, with a checksum. A Reader is bounded to a byte range. The recurring bug class is a boundary condition: a loop that tests bytesRead >= segmentLength after a read instead of before can emit one phantom record when a segment ends exactly on a boundary. Read the writer and the reader together — they share invariants and usually share bugs.

  cd /Users/s0x/src/oss-repos/tez
  grep -n "EOF_MARKER\|nextRawKey\|segmentLength\|bytesRead" \
    tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/IFile.java | head
  

• The memory-vs-disk decision. MergeManager decides whether a fetched output goes to memory or spills to disk, governed by tez.runtime.shuffle.memory.limit.percent and the fetch buffer percent. Relaxing a threshold to "fix" an OOM is almost always wrong — the thresholds are a contract with the AM's memory model. If the merge runs out of memory, the bug is usually an upstream Fetcher that put something in memory that should have gone to disk.

• Threads, not types, enforce the contracts. The ShuffleScheduler, the ShufflePenaltyReferee, the Fetcher pool, and close() all run concurrently. A fix that "works" in a single-threaded test can deadlock or race in production. This is why the deep dive shuffle & sort is a hard prerequisite, and why Stage 9 drills deterministic testing so hard.

Case study A — TEZ-4334: a deadlock between close() and the penalty thread

This is the canonical Stage 6 concurrency fix. Read it end to end:

git show 25a953677   # Fix deadlock in ShuffleScheduler between close() and ShufflePenaltyReferee

The symptom. A DAG hangs. Not fails — hangs. Thread dumps show ShuffleScheduler.close() waiting on the scheduler monitor while the ShufflePenaltyReferee thread holds it and is blocked reporting an exception. A classic lock-ordering deadlock.

The root cause. copyFailed(...) was declared synchronized on the whole method. Inside it, when shuffle went unhealthy, it called exceptionReporter.reportException(...) — while still holding the scheduler monitor. reportException can call back into scheduler shutdown, which needs the same monitor. Two threads, two acquisition orders, deadlock.

The fix — the shape to memorise. Bodor and Sungwoo Park shrank the critical section instead of widening it. copyFailed stopped being synchronized at the method level:

public void copyFailed(InputAttemptFetchFailure fetchFailure, MapHost host,
    boolean readError, boolean connectError) {
  ...
  synchronized (this) {
    failures = incrementAndGetFailureAttempt(fetchFailure.getInputAttemptIdentifier());
    ...
  }

The health check was refactored from a boolean-returning isShuffleHealthy(...) into a checkShuffleHealthy(...) that throws IOException when unhealthy, and the reporting was moved outside the lock:

try {
  checkShuffleHealthy(fetchFailure);
} catch (IOException e) {
  // reportException should be called outside synchronized(this) due to TEZ-4334
  exceptionReporter.reportException(e);
  return;
}

Note the inline comment naming the JIRA — that is the Tez convention for a non-obvious concurrency invariant, and you should copy it. penalizeHost and checkShuffleHealthy were each made synchronized individually, so the lock is held for the state mutation and released before the callback.

The lesson. When you find a deadlock, the fix is almost never "add a lock." It is "hold the lock across less code." Any callback to another component (reportException, an event dispatch, a listener) must happen outside your monitor. Notice this fix builds directly on TEZ-4336's fetchFailure.getCause() carrier from Stage 3 — the exception it now throws still preserves the original cause.

Case study B — TEZ-3849: a combiner silently drops records

git show d5ac3b75f   # Combiner+PipelinedSorter silently drops records

The symptom. No exception, no error, no crash. A DAG using a combiner with the PipelinedSorter produces fewer output records than input records. Silent data loss — the worst possible bug, because nothing alerts on it.

The root cause. The sort spill loop conflated "is there a next record?" with "advance to the next record." PipelinedSorter did:

boolean hasNext = kvIter.next();       // this ADVANCES and returns whether a value exists
if (hasNext || !sendEmptyPartitionDetails) { ... }
if (combiner == null) {
  while (hasNext) {
    writer.append(kvIter.getKey(), kvIter.getValue());
    hasNext = kvIter.next();           // advance again
  }
} else {
  if (hasNext) { /* hand kvIter to the combiner */ }
}

When a combiner was present, the code had already consumed one record via the initial kvIter.next() before handing the iterator to the combiner — so the combiner started from the second record and the first was lost per partition.

The fix — separate hasNext() from next(). Jacob Tolar introduced a proper hasNext() peek across the iterator stack (TezRawKeyValueIterator, the sorter spans, TezMerger) so callers can test without consuming:

if (combiner == null) {
  while (kvIter.next()) {
    writer.append(kvIter.getKey(), kvIter.getValue());
  }
}

TezMerger.next() was simplified to lean on the new hasNext():

public boolean next() throws IOException {
  if (!hasNext()) {
    return false;
  }
  minSegment = top();
  ...
}

and hasNext()/peekPartition() were added to every layer so the combiner path could look ahead without eating a record.

What the test did. TestMRCombiner and TestPipelinedSorter were extended with a mock iterator whose hasNext() and next() are distinct, then asserted the combiner sees every record. The test is the proof the record count is conserved.

The lesson. An iterator whose next() both tests and advances is a data-loss bug waiting to happen the moment two callers share it. When you touch any TezRawKeyValueIterator, verify hasNext() is side-effect-free. Silent correctness bugs have no stack trace — the only defense is a test that counts.

Case study C — TEZ-3761: a NullPointerException in Fetcher under load

git show a7f93ae1d   # NPE in Fetcher under load

The symptom. Under heavy shuffle load, the unordered Fetcher throws an NPE deep in the copy path. Rare, load-dependent, and it fails the whole input.

The root cause. The Fetcher looked up the source attempt from a shuffle-header mapId and used the result without checking for null. A corrupted or unexpected header — which happens under load, or with a misbehaving shuffle handler — produced a pathComponent that didn't map to any known attempt, so pathToAttemptMap.get(...) returned null and the next dereference NPE'd.

The fix — validate the header, name the mismatch. Jonathan Eagles added two guards that turn an opaque NPE into a self-describing IllegalArgumentException:

if (!pathComponent.startsWith(InputAttemptIdentifier.PATH_PREFIX)) {
  throw new IllegalArgumentException("Invalid map id: " + header.getMapId()
      + ", expected to start with " + InputAttemptIdentifier.PATH_PREFIX
      + ", partition: " + header.getPartition()
      + " while fetching " + inputAttemptIdentifier);
}
srcAttemptId = pathToAttemptMap.get(new PathPartition(pathComponent, header.getPartition()));
if (srcAttemptId == null) {
  throw new IllegalArgumentException("Source attempt not found for map id: "
      + header.getMapId() + ", partition: " + header.getPartition()
      + " while fetching " + inputAttemptIdentifier);
}

Ten lines, no test — because the fix is defensive validation of an externally-supplied header, and the value it adds is a diagnosable error instead of a bare NPE. This is Stage 3's CONTEXT rule (name the map id, the partition, and what you were fetching) applied inside the hot fetch loop.

The lesson. Data that crosses a network boundary (a shuffle header from the NM's shuffle handler) is untrusted. Every field you pull out of it can be malformed. Validate before you dereference, and make the validation message name the exact values so an operator can correlate it with the misbehaving source.

Related: TEZ-3976 (git show 79af4e8d0) added tez.runtime.shuffle.batch.wait and equals/hashCode on InputReadErrorEvent so the ShuffleManager can batch and deduplicate read-error events before sending them to the AM — the structural fix for the "one bad NM floods the AM" storm. Read it as the companion to TEZ-4334: one bounds the lock, the other bounds the event volume.

The IFile boundary-bug pattern

You will meet this class often enough to name it. An IFile.Reader is constructed bounded to a byte range and iterates records until EOF. The recurring defect is a loop that checks the boundary after consuming a record rather than before:

public boolean nextRawKey(...) throws IOException {
  int recordLength = readVInt(dataIn);      // reads PAST the boundary...
  if (recordLength == EOF_MARKER) return false;
  // ...then discovers it should have stopped — one phantom record emitted
}

The correct shape tests bytesRead >= segmentLength first. Locate the loop and the writer that produced the data — they share the invariant, so a reader boundary bug usually has a writer sibling:

grep -n "nextRawKey\|EOF_MARKER\|segmentLength\|bytesRead\|append(" \
  tez-runtime-library/src/main/java/org/apache/tez/runtime/library/common/sort/impl/IFile.java

The test that catches it writes exactly N records, captures the byte length, constructs a Reader bounded to that length, and asserts the (N+1)th nextRawKey returns false without throwing — a boundary test, not a happy-path test. Never "fix" one of these by widening the check; read the writer first.

The contribution playbook for this class

1. Reproduce deterministically first. Concurrency and load bugs need a ControlledClock, a DrainDispatcher, or a deterministic executor — never a Thread.sleep. Stage 9 exists because sleep-based shuffle tests are the #1 flake source.
2. For a concurrency fix, write the lock-ordering down. Which monitor, held across which callbacks? The TEZ-4334 comment is your template.
3. For a correctness fix, write a counting test. Input record count must equal output record count (TEZ-3849). No exception ≠ correct.
4. Register any new config key. A key added to TezRuntimeConfiguration must also be added to the tezRuntimeKeys set or the validator silently ignores it (TEZ-3976 did both).
5. Grep for prior art before patching IFile/TezMerger — the reader and writer usually have sibling bugs. git log --oneline -- <file> on the exact file shows every past fix.
6. Run the targeted suites, three times, to smoke out flakes you introduce.

mvn -pl tez-runtime-library test -Dtest=TestShuffleScheduler,TestPipelinedSorter,TestIFile -q
mvn -pl tez-mapreduce test -Dtest=TestMRCombiner -q

Common mistakes

Exit criteria — when you're ready for the next stage

• You have shipped one shuffle/runtime patch with a deterministic regression test that passes 200 times in a row.
• You can explain the TEZ-4334 lock ordering out loud: what deadlocked, and why moving reportException outside synchronized(this) fixed it.
• You can explain why TEZ-3849 lost records, and you can articulate the hasNext()/next() contract without looking it up.
• You have read MergeManager and TezMerger end to end and can describe the in-memory vs on-disk merge branches.
• A reviewer accepted your fix without asking "is this the same bug as TEZ-XXXX?" — meaning you grepped git log for prior art before patching.

Stage 7 takes you out of core Tez and into Hive-on-Tez compatibility.