HIVE-29780: Fix LLAP plugin rendezvous race under concurrent DAGs (NullPointerException in LlapTaskCommunicator$3.setResponse) - #6659
Open
abstractdog wants to merge 1 commit into
Open
Conversation
…llPointerException in LlapTaskCommunicator.setResponse)
LlapTaskCommunicator and LlapTaskSchedulerService are constructed
independently by the Tez app-master and have to find each other to
wire the callback that reports task starts back to the scheduler
(scheduler.notifyStarted(taskAttemptID) in the SubmitWork response
callback). They used a class-static handshake:
static final Object pluginInitLock = new Object();
static LlapTaskCommunicator instance = null; // one slot
static LlapTaskSchedulerService instance = null;
// in each constructor, under pluginInitLock:
if (peer.instance != null) { pair(); peer.instance = null; }
else { this.instance = this; }
This single-slot handshake is race-prone when multiple DAGs come up
concurrently in the same JVM. The communicator from DAG-A could end up
paired with the scheduler from DAG-B (whichever happened to park itself
in "instance" first), and DAG-A's real communicator was then left with
scheduler == null. The first task submission for that DAG then hit
LlapTaskCommunicator$3.setResponse:540 → scheduler.notifyStarted(...)
→ NPE, and the vertex died with OWN_TASK_FAILURE / return code 2.
A production Tez DAGAppMaster only runs one DAG at a time, so this race
is not reachable there. It shows up in JVMs that host many concurrent
DAGs — MiniHS2, MiniLlapCluster, tests that fan out concurrent inserts.
In a 30-way concurrent insert-only INSERT reproduction on MiniHS2 this
took out ~5 of 30 sessions.
Fix: key the rendezvous on ApplicationAttemptId (both TaskCommunicator
Context and TaskSchedulerContext expose it, and the two plugins for the
same DAG share it). Replace the class-static "instance" slot with a
ConcurrentMap<ApplicationAttemptId, ThisSide>. The pluginInitLock stays
so the pair-or-park section remains atomic per DAG. shutdown() on each
side reaps its own leftover entry via ConcurrentMap.remove(key, this),
so a failed DAG init that only constructed one plugin doesn't leak.
Production impact: none. The map holds at most one entry per DAG init;
in a production AM (one DAG per AM lifetime) that's at most one entry
total, put and removed at plugin construction / shutdown. Nothing on
the task submission or scheduling hot path changes.
Verification (30-way concurrent INSERT into an insert-only managed
partitioned Parquet table on s3a://):
Before: 25 of 30 sessions succeeded, 5 hit the NPE (hive.log had
120 stack traces from Tez task-attempt retries).
After: 30 of 30 sessions succeeded, hive.log has 0 notifyStarted
NPEs. All 30 writers landed in their own
delta_<writeId>_<writeId>_0000 subdirectory as expected for
insert-only ACID.
asf-ci-hive
added
tests failed
tests pending
and removed
tests pending
tests failed
labels
|
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
What changes were proposed in this pull request?
Replace the class-static single-slot rendezvous between
LlapTaskCommunicatorandLlapTaskSchedulerServicewith a per-ApplicationAttemptIdhandshake. Each side now parks itself in aConcurrentMap<ApplicationAttemptId, ThisSide>keyed by the appAttemptId both plugins for the same DAG share (viaTaskCommunicatorContext.getApplicationAttemptId()/TaskSchedulerContext.getApplicationAttemptId()). The existingpluginInitLockstays so the pair-or-park section is atomic per DAG.shutdown()on each side reaps its own leftover entry withConcurrentMap.remove(key, this)so a failed DAG init doesn't leak.Why are the changes needed?
The single-slot handshake pairs plugins across DAGs when multiple DAGs come up concurrently in the same JVM: the communicator from DAG-A ends up bound to the scheduler from DAG-B, DAG-A's real communicator is left with
scheduler == null, and the first task submission hitswhich surfaces as
Vertex Map 1 killed/failed due to:OWN_TASK_FAILURE→MoveTask return code 2. In a 30-way concurrent-INSERT reproduction on MiniHS2 (LLAP) 5 of 30 sessions failed with this NPE.Does this PR introduce any user-facing change?
No. A production Tez
DAGAppMasterruns one DAG at a time so the race is not reachable there, and the map holds at most one entry at any moment; the hot paths (task submission, scheduling) are unchanged.How was this patch tested?
30-way concurrent
INSERTburst on MiniHS2 (clusterType=llap,-DminiHS2.isMetastoreRemote=true) against an insert-only managed partitioned table. Before: 25/30 sessions succeeded, 5 hit the NPE andhive.loghad 120 stack traces from Tez task-attempt retries. After: 30/30 sessions succeeded, zeronotifyStartedNPEs inhive.log, all 30 writers landed in their own subdirectory as expected for insert-only ACID.