HIVE-29780: Fix LLAP plugin rendezvous race under concurrent DAGs (NullPointerException in LlapTaskCommunicator.setResponse)

[abstractdog]

What changes were proposed in this pull request?

Replace the class-static single-slot rendezvous between LlapTaskCommunicator and LlapTaskSchedulerService with a per-ApplicationAttemptId handshake. Each side now parks itself in a ConcurrentMap<ApplicationAttemptId, ThisSide> keyed by the appAttemptId both plugins for the same DAG share (via TaskCommunicatorContext.getApplicationAttemptId() / TaskSchedulerContext.getApplicationAttemptId()). The existing pluginInitLock stays so the pair-or-park section is atomic per DAG. shutdown() on each side reaps its own leftover entry with ConcurrentMap.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 hits

java.lang.NullPointerException: Cannot invoke
  "LlapTaskSchedulerService.notifyStarted(TezTaskAttemptID)"
  because "this.this$0.scheduler" is null
    at LlapTaskCommunicator$3.setResponse(LlapTaskCommunicator.java:540)

which 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 DAGAppMaster runs 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 INSERT burst on MiniHS2 (clusterType=llap, -DminiHS2.isMetastoreRemote=true) against an insert-only managed partitioned table. Before: 25/30 sessions succeeded, 5 hit the NPE and hive.log had 120 stack traces from Tez task-attempt retries. After: 30/30 sessions succeeded, zero notifyStarted NPEs in hive.log, all 30 writers landed in their own subdirectory as expected for insert-only ACID.

[abstractdog]

I'll fix related failures in TestLlapTaskSchedulerService

[Copilot]

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.

[abstractdog]

makes sense, let me try to add a unit test

[abstractdog]

added TestLlapPluginRendezvous, maybe it's an overkill, considering that this whole patch is a single JVM multiple DAGs test scenario, and manual testing already proved it

[abstractdog]

this makes sense to me
need to be aware that in production, we'll always have a single instance per AM, but taking care of removing the instances properly in the shutdown() method makes sense to me

however, this is not true: "a later communicator init to pair with a stopped scheduler" <- this cannot happen, as both the schedulers and communicators are keyed with ApplicationAttemptId, and every new JVM-local DAGAppMaster gets a new application id...so the risk is not accidental re-pairing, just a simple leak, which has to be avoided

[sonarqubecloud]

Quality Gate passed

Issues
2 New issues
0 Accepted issues

Measures
0 Security Hotspots
0.0% Coverage on New Code
2.2% Duplication on New Code

See analysis details on SonarQube Cloud