HIVE-29780: Fix LLAP plugin rendezvous race under concurrent DAGs (NullPointerException in LlapTaskCommunicator.setResponse) - #6659
Conversation
8b20674 to
df2cc85
Compare
asf-ci-hive
added
tests failed
tests pending
tests unstable
and removed
tests pending
tests failed
labels
abstractdog
changed the title
|
I'll fix related failures in |
There was a problem hiding this comment.
Pull request overview
This PR addresses a Tez LLAP plugin initialization race where concurrent DAGs in the same JVM could cross-wire LlapTaskCommunicator and LlapTaskSchedulerService, leading to scheduler == null and a NullPointerException in LlapTaskCommunicator.setResponse. It replaces the previous single static-slot rendezvous with a per-ApplicationAttemptId handshake so each DAG’s plugin pair is matched only with its counterpart.
Changes:
- Replace the class-static single “instance” rendezvous with
ConcurrentMap<ApplicationAttemptId, ...>parking/removal on both plugins. - Key the handshake on
TaskCommunicatorContext.getApplicationAttemptId()/TaskSchedulerContext.getApplicationAttemptId()while keeping the existing init lock for atomic pairing.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 2 comments.
| File | Description |
|---|---|
| llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java | Parks/removes schedulers by ApplicationAttemptId to avoid cross-DAG plugin pairing. |
| llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java | Parks/removes communicators by ApplicationAttemptId to avoid cross-DAG plugin pairing and documents the race. |
Comments suppressed due to low confidence (1)
llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java:176
- pendingCommunicators entries are never removed if the communicator parks itself (put) and then gets shut down before the matching scheduler is constructed. That can leak the communicator instance and also lets a later scheduler init for the same ApplicationAttemptId pair against a stopped communicator (reintroducing null/incorrect wiring). Consider storing the appAttemptId on the instance and removing the entry in shutdown (e.g., synchronized on pluginInitLock, pendingCommunicators.remove(appAttemptId, this)).
ApplicationAttemptId appAttemptId = getContext().getApplicationAttemptId();
synchronized (pluginInitLock) {
LlapTaskSchedulerService peer = LlapTaskSchedulerService.pendingSchedulers.remove(appAttemptId);
if (peer != null) {
// We are the last to initialize for this DAG.
peer.setTaskCommunicator(this);
this.setScheduler(peer);
} else {
pendingCommunicators.put(appAttemptId, this);
}
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
asf-ci-hive
added
tests passed
tests pending
tests unstable
and removed
tests pending
tests passed
labels
|
illiabarbashov-sketch
left a comment
illiabarbashov-sketch
left a comment
There was a problem hiding this comment.
LGTM. A couple of questions/clarifications asked.
| this.appAttemptId = getContext().getApplicationAttemptId(); | ||
| synchronized (pluginInitLock) { | ||
| LlapTaskSchedulerService peer = LlapTaskSchedulerService.instance; | ||
| LlapTaskSchedulerService peer = LlapTaskSchedulerService.pendingSchedulers.remove(appAttemptId); |
There was a problem hiding this comment.
Would like to know what do you think about extracting this handshake logic into a separate singleton class LlapPluginRendezvous since test already bears that name? The benefit is that this would allow us to avoid static fields cross read between those two classes. While the separate singleton class will work as an intermediary for both.
There was a problem hiding this comment.
absolutely makes sense!
in the long term, tez might have its own solution for the same (I can also find a ticket where this came up in tez), but now, it would be a good to move all this hack to a separate class
| @Override | ||
| public void shutdown() { | ||
| super.shutdown(); | ||
| // Reap our rendezvous entry if we parked ourselves at construction and are being torn down |
There was a problem hiding this comment.
I don't know if we want to provide that much of a context in the comment for 3-liner? On the other hand, that might be another +1 to a separate class to handle rendezvous handshake, we can comment the purpose in the javadoc of that class. What do you think?
There was a problem hiding this comment.
right, it's just not necessary to explain it too much here
…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.
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.