This is an automated email from the ASF dual-hosted git repository.
amaechler pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/druid.git
The following commit(s) were added to refs/heads/master by this push:
new d61f8cf9352 chore: restore query ID verification in MSQ workers
(#20530)
d61f8cf9352 is described below
commit d61f8cf93527d914d0750b3c06a7a24162bf5830
Author: Nanjesh Ramesh <[email protected]>
AuthorDate: Tue Oct 20 08:32:35 2026 -0700
chore: restore query ID verification in MSQ workers (#20530)
---
.../main/java/org/apache/druid/msq/exec/WorkerImpl.java | 17 +++++++++++++----
.../main/java/org/apache/druid/msq/kernel/StageId.java | 3 ---
2 files changed, 13 insertions(+), 7 deletions(-)
diff --git
a/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java
b/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java
index 9a99f8d4812..fae87486d36 100644
--- a/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java
+++ b/multi-stage-query/src/main/java/org/apache/druid/msq/exec/WorkerImpl.java
@@ -1133,7 +1133,7 @@ public class WorkerImpl implements Worker
*/
public void addKernel(final WorkerStageKernel kernel)
{
- final StageId stageId =
kernel.getWorkOrder().getStageDefinition().getId();
+ final StageId stageId =
verifyQueryId(kernel.getWorkOrder().getStageDefinition().getId());
if (holderMap.putIfAbsent(stageId.getStageNumber(), new
KernelHolder(kernel)) != null) {
// Already added. Do nothing.
@@ -1149,7 +1149,7 @@ public class WorkerImpl implements Worker
*/
public void finishProcessing(final StageId stageId)
{
- final KernelHolder kernel = holderMap.get(stageId.getStageNumber());
+ final KernelHolder kernel =
holderMap.get(verifyQueryId(stageId).getStageNumber());
if (kernel != null) {
try {
@@ -1170,7 +1170,7 @@ public class WorkerImpl implements Worker
*/
public void removeKernel(final StageId stageId)
{
- final KernelHolder removed = holderMap.remove(stageId.getStageNumber());
+ final KernelHolder removed =
holderMap.remove(verifyQueryId(stageId).getStageNumber());
if (removed == null) {
throw new ISE("No kernel for stage[%s]", stageId);
@@ -1224,7 +1224,7 @@ public class WorkerImpl implements Worker
@Nullable
public WorkerStageKernel getKernelFor(final StageId stageId)
{
- final KernelHolder holder = holderMap.get(stageId.getStageNumber());
+ final KernelHolder holder =
holderMap.get(verifyQueryId(stageId).getStageNumber());
if (holder != null) {
return holder.kernel;
} else {
@@ -1273,6 +1273,15 @@ public class WorkerImpl implements Worker
{
this.done = true;
}
+
+ private StageId verifyQueryId(final StageId stageId)
+ {
+ if (!stageId.getQueryId().equals(workerContext.queryId())) {
+ throw new ISE("Unexpected queryId[%s], expected queryId[%s]",
stageId.getQueryId(), workerContext.queryId());
+ }
+
+ return stageId;
+ }
}
/**
diff --git
a/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/StageId.java
b/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/StageId.java
index a928e5834fc..5b98eed0da9 100644
--- a/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/StageId.java
+++ b/multi-stage-query/src/main/java/org/apache/druid/msq/kernel/StageId.java
@@ -31,9 +31,6 @@ import java.util.Objects;
/**
* Globally unique stage identifier: query ID plus stage number.
- *
- * Note: Versions till Druid 30 had a bug in the QueryKits which populated the
{@link #queryId} field with random
- * UUIDs. Therefore, all usage of the field must be vetted instead of assuming
that it will be the expected query id
*/
public class StageId implements Comparable<StageId>
{
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
