Copilot commented on code in PR #6659:
URL: https://github.com/apache/hive/pull/6659#discussion_r3680499836
##########
llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskCommunicator.java:
##########
@@ -158,15 +164,15 @@ public LlapTaskCommunicator(
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);
}
Review Comment:
The new per-ApplicationAttemptId rendezvous logic is central to preventing
cross-DAG pairing, but there does not appear to be a unit test covering (a) two
concurrent appAttemptIds not cross-wiring and (b) parked-side shutdown reaping
the pending map entry. Adding a focused test (e.g., constructing
communicator/scheduler pairs with mocked contexts and distinct
ApplicationAttemptIds) would help prevent regressions.
This issue also appears on line 167 of the same file.
##########
llap-tez/src/java/org/apache/hadoop/hive/llap/tezplugins/LlapTaskSchedulerService.java:
##########
@@ -470,15 +472,15 @@ public LlapTaskSchedulerService(TaskSchedulerContext
taskSchedulerContext, Clock
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);
}
Review Comment:
pendingSchedulers entries are never removed if the scheduler parks itself
(put) and then shuts down before the matching communicator is constructed. This
can leak the scheduler instance and allow a later communicator init to pair
with a stopped scheduler. Consider capturing the ApplicationAttemptId on the
instance and removing the entry in shutdown (e.g., synchronized on
LlapTaskCommunicator.pluginInitLock, pendingSchedulers.remove(appAttemptId,
this)).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
