Skip to content

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

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
abstractdog wants to merge 1 commit into
base: master
Choose a base branch
from
+24 −16
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
24 changes: 15 additions & 9 deletions llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,6 +79,7 @@
import org.apache.hadoop.security.token.Token;
import org.apache.hadoop.security.token.SecretManager.InvalidToken;
import org.apache.hadoop.yarn.api.ApplicationConstants;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ContainerId;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.NodeId;
Expand Down Expand Up @@ -137,12 +138,17 @@
private volatile QueryIdentifierProto currentQueryIdentifierProto;
private volatile String currentHiveQueryId;

// TODO: this is an ugly hack because Tez plugin isolation does not make sense for LLAP plugins.
// We are going to register a thread-local here for now, so that the scheduler, initializing
// in the same thread after the communicator, will pick up. Or the other way around.
// This only lives for the duration of the service init.
// Cross-plugin rendezvous keyed on ApplicationAttemptId. Tez does not offer a first-class way
// for two LLAP-side plugins (the task communicator and the task scheduler) to find each other,
// and the plugins are constructed independently by the DAG app-master. When multiple DAGs come
// up concurrently the two sides used to race on a single static "instance" slot and could pair
// up across DAGs — the communicator for DAG-A would take the scheduler that was parked by
// DAG-B, leaving DAG-A's real communicator with scheduler == null and producing an NPE in
// notifyStarted on the first task submission. Keying on the ApplicationAttemptId shared by
// both plugins for the same DAG confines each handshake to its own DAG.
static final Object pluginInitLock = new Object();
static LlapTaskCommunicator instance = null;
static final ConcurrentMap<ApplicationAttemptId, LlapTaskCommunicator> pendingCommunicators =

Check warning on line 150 in llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Name 'pendingCommunicators' must match pattern '^[A-Z][A-Z0-9]*(_[A-Z0-9]+)*$'. 


See more on https://sonarcloud.io/project/issues?id=apache_hive&issues=AZ-vqY_g1mSvuoDuGNzd&open=AZ-vqY_g1mSvuoDuGNzd&pullRequest=6659
new ConcurrentHashMap<>();

public LlapTaskCommunicator(
TaskCommunicatorContext taskCommunicatorContext) {
Expand All @@ -158,15 +164,15 @@

credentialMap = new ConcurrentHashMap<>();
sourceStateTracker = new SourceStateTracker(getContext(), this);
ApplicationAttemptId appAttemptId = getContext().getApplicationAttemptId();
synchronized (pluginInitLock) {
LlapTaskSchedulerService peer = LlapTaskSchedulerService.instance;
LlapTaskSchedulerService peer = LlapTaskSchedulerService.pendingSchedulers.remove(appAttemptId);
if (peer != null) {
// We are the last to initialize.
// We are the last to initialize for this DAG.
peer.setTaskCommunicator(this);
this.setScheduler(peer);
LlapTaskSchedulerService.instance = null;
} else {
instance = this;
pendingCommunicators.put(appAttemptId, this);
}
Comment on lines 141 to 176

@abstractdog abstractdog Jul 30, 2026

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

makes sense, let me try to add a unit test

}
}
Expand Down
16 changes: 9 additions & 7 deletions llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.apache.hadoop.io.Text;
import org.apache.hadoop.metrics2.MetricsSource;
import org.apache.hadoop.metrics2.MetricsSystem;
import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
import org.apache.hadoop.yarn.api.records.ApplicationId;

import org.apache.hadoop.hive.registry.impl.TezAmRegistryImpl;
Expand Down Expand Up @@ -198,9 +199,10 @@
"Lock metrics for R/W locks LLAP task scheduler", LOCK_METRICS);
}

// TODO: this is an ugly hack; see the same in LlapTaskCommunicator for discussion.
// This only lives for the duration of the service init.
static LlapTaskSchedulerService instance = null;
// Cross-plugin rendezvous keyed on ApplicationAttemptId — see the discussion on the parallel
// pendingCommunicators map in LlapTaskCommunicator.
static final ConcurrentMap<ApplicationAttemptId, LlapTaskSchedulerService> pendingSchedulers =

Check warning on line 204 in llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java

View check run for this annotation

SonarQubeCloud / SonarCloud Code Analysis

Name 'pendingSchedulers' must match pattern '^[A-Z][A-Z0-9]*(_[A-Z0-9]+)*$'. 


See more on https://sonarcloud.io/project/issues?id=apache_hive&issues=AZ-vqY-n1mSvuoDuGNzc&open=AZ-vqY-n1mSvuoDuGNzc&pullRequest=6659
new ConcurrentHashMap<>();

private final Configuration conf;

Expand Down Expand Up @@ -470,15 +472,15 @@
this.workloadManagementEnabled =
!StringUtils.isEmpty(conf.get(ConfVars.HIVE_SERVER2_TEZ_INTERACTIVE_QUEUE.varname, "").trim());

ApplicationAttemptId appAttemptId = getContext().getApplicationAttemptId();
synchronized (LlapTaskCommunicator.pluginInitLock) {
LlapTaskCommunicator peer = LlapTaskCommunicator.instance;
LlapTaskCommunicator peer = LlapTaskCommunicator.pendingCommunicators.remove(appAttemptId);
if (peer != null) {
// We are the last to initialize.
// We are the last to initialize for this DAG.
this.setTaskCommunicator(peer);
peer.setScheduler(this);
LlapTaskCommunicator.instance = null;
} else {
instance = this;
pendingSchedulers.put(appAttemptId, this);
}
Comment on lines +475 to 484

@abstractdog abstractdog Jul 30, 2026 •
edited
Loading

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

}
}
Expand Down
Loading