This is an automated email from the ASF dual-hosted git repository.
Mryange pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 7839b20878b [fix](fe) Guard fragment cleanup until dispatch completes
(#66767)
7839b20878b is described below
commit 7839b20878b2e0e497b71d71b3b2cf1e7fd36494
Author: Mryange <[email protected]>
AuthorDate: Mon Aug 17 10:11:59 2026 +0800
[fix](fe) Guard fragment cleanup until dispatch completes (#66767)
Production stress-test logs captured the lifecycle race for a load
query. At
13:35:52.506, BE received `FINISHED` and destroyed Query Context
`6aa1098800c457e-bad0398f81aa610f`. About 61 ms later, its phase-two
`exec_plan_fragment_start` RPC arrived and failed with `Failed to get
query fragments context`,
causing the INSERT SELECT to fail. Root cause: FE could broadcast
successful cleanup as soon as
execution completion was reported, without waiting for all fragment
dispatch RPCs to finish. This
change delays `FINISHED` cleanup until both execution and fragment
dispatch complete. It also
ensures that if the dispatch deadline has already expired after RPC
futures were submitted, FE
records the error and actively cancels the coordinator so prepared or
running backend contexts are
not left behind until backend timeout.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [ ] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [ ] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [ ] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
.../org/apache/doris/qe/AbstractJobProcessor.java | 17 +++++
.../java/org/apache/doris/qe/JobProcessor.java | 2 +
.../doris/qe/runtime/PipelineExecutionTask.java | 4 +
.../apache/doris/qe/AbstractJobProcessorTest.java | 88 ++++++++++++++++++++++
.../qe/runtime/PipelineExecutionTaskTest.java | 76 +++++++++++++++++++
5 files changed, 187 insertions(+)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/AbstractJobProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/AbstractJobProcessor.java
index 991f839a511..647391dffbc 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/AbstractJobProcessor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/AbstractJobProcessor.java
@@ -45,6 +45,9 @@ public abstract class AbstractJobProcessor implements
JobProcessor {
private final Logger logger = LogManager.getLogger(getClass());
protected final AtomicBoolean finished = new AtomicBoolean(false);
+ // FINISHED cleanup must wait until all fragment dispatch RPCs, including
phase-two starts, complete.
+ private final AtomicBoolean executionFinished = new AtomicBoolean(false);
+ private final AtomicBoolean fragmentDispatchCompleted = new
AtomicBoolean(false);
protected final CoordinatorContext coordinatorContext;
protected volatile Optional<PipelineExecutionTask> executionTask;
protected volatile Optional<Map<BackendFragmentId,
SingleFragmentPipelineTask>> backendFragmentTasks;
@@ -74,6 +77,20 @@ public abstract class AbstractJobProcessor implements
JobProcessor {
@Override
public void tryFinishSchedule() {
+ executionFinished.set(true);
+ tryBroadcastExecutionFinished();
+ }
+
+ @Override
+ public void markFragmentDispatchCompleted() {
+ fragmentDispatchCompleted.set(true);
+ tryBroadcastExecutionFinished();
+ }
+
+ private void tryBroadcastExecutionFinished() {
+ if (!executionFinished.get() || !fragmentDispatchCompleted.get()) {
+ return;
+ }
if (finished.compareAndSet(false, true)) {
this.executionTask.ifPresent(sqlPipelineTask -> {
for (MultiFragmentsPipelineTask fragmentsTask :
sqlPipelineTask.getChildrenTasks().values()) {
diff --git a/fe/fe-core/src/main/java/org/apache/doris/qe/JobProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/JobProcessor.java
index bc2b991032a..833f5c7773a 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/JobProcessor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/JobProcessor.java
@@ -29,4 +29,6 @@ public interface JobProcessor {
boolean updateFragmentExecStatus(TReportExecStatusParams params);
void tryFinishSchedule();
+
+ void markFragmentDispatchCompleted();
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/PipelineExecutionTask.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/PipelineExecutionTask.java
index 6a52e3a6d9f..59670ccbbf1 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/PipelineExecutionTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/PipelineExecutionTask.java
@@ -97,6 +97,7 @@ public class PipelineExecutionTask extends
AbstractRuntimeTask<BackendWorker, Mu
if (coordinatorContext.twoPhaseExecution()) {
sendAndWaitPhaseTwoRpc();
}
+
coordinatorContext.getJobProcessor().markFragmentDispatchCompleted();
return null;
});
}
@@ -168,6 +169,9 @@ public class PipelineExecutionTask extends
AbstractRuntimeTask<BackendWorker, Mu
queryOptions.isSetQueryTimeout(),
queryOptions.getQueryTimeout(),
timeoutDeadline, currentTimeMillis);
}
+ Status cancelStatus = new Status(TStatusCode.INTERNAL_ERROR, msg);
+ coordinatorContext.updateStatusIfOk(cancelStatus);
+ coordinatorContext.cancelSchedule(cancelStatus);
throw new UserException(msg);
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/AbstractJobProcessorTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/AbstractJobProcessorTest.java
new file mode 100644
index 00000000000..64f57cfec1e
--- /dev/null
+++ b/fe/fe-core/src/test/java/org/apache/doris/qe/AbstractJobProcessorTest.java
@@ -0,0 +1,88 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.qe;
+
+import org.apache.doris.common.Status;
+import org.apache.doris.nereids.trees.plans.distribute.worker.BackendWorker;
+import org.apache.doris.qe.runtime.MultiFragmentsPipelineTask;
+import org.apache.doris.qe.runtime.PipelineExecutionTask;
+import org.apache.doris.qe.runtime.SingleFragmentPipelineTask;
+import org.apache.doris.thrift.TReportExecStatusParams;
+
+import org.junit.jupiter.api.Test;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.Optional;
+
+class AbstractJobProcessorTest {
+ @Test
+ void finishBeforeFragmentDispatchDoesNotCancelPreparedFragments() {
+ MultiFragmentsPipelineTask fragmentsTask =
Mockito.mock(MultiFragmentsPipelineTask.class);
+ TestJobProcessor processor = createProcessor(fragmentsTask);
+
+ processor.tryFinishSchedule();
+ Mockito.verifyNoInteractions(fragmentsTask);
+
+ processor.markFragmentDispatchCompleted();
+ Mockito.verify(fragmentsTask).cancelExecute(Status.FINISHED);
+
+ processor.tryFinishSchedule();
+ processor.markFragmentDispatchCompleted();
+ Mockito.verifyNoMoreInteractions(fragmentsTask);
+ }
+
+ @Test
+ void fragmentDispatchBeforeFinishBroadcastsWhenExecutionFinishes() {
+ MultiFragmentsPipelineTask fragmentsTask =
Mockito.mock(MultiFragmentsPipelineTask.class);
+ TestJobProcessor processor = createProcessor(fragmentsTask);
+
+ processor.markFragmentDispatchCompleted();
+ Mockito.verifyNoInteractions(fragmentsTask);
+
+ processor.tryFinishSchedule();
+ Mockito.verify(fragmentsTask).cancelExecute(Status.FINISHED);
+ }
+
+ private static TestJobProcessor createProcessor(MultiFragmentsPipelineTask
fragmentsTask) {
+ BackendWorker worker = Mockito.mock(BackendWorker.class);
+ PipelineExecutionTask executionTask =
Mockito.mock(PipelineExecutionTask.class);
+
Mockito.when(executionTask.getChildrenTasks()).thenReturn(Collections.singletonMap(worker,
fragmentsTask));
+
+ TestJobProcessor processor = new
TestJobProcessor(Mockito.mock(CoordinatorContext.class));
+ processor.setExecutionTask(executionTask);
+ return processor;
+ }
+
+ private static class TestJobProcessor extends AbstractJobProcessor {
+ TestJobProcessor(CoordinatorContext coordinatorContext) {
+ super(coordinatorContext);
+ }
+
+ void setExecutionTask(PipelineExecutionTask executionTask) {
+ this.executionTask = Optional.of(executionTask);
+ }
+
+ @Override
+ protected void doProcessReportExecStatus(
+ TReportExecStatusParams params, SingleFragmentPipelineTask
fragmentTask) {}
+
+ @Override
+ public void cancel(Status cancelReason) {}
+ }
+}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/PipelineExecutionTaskTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/PipelineExecutionTaskTest.java
new file mode 100644
index 00000000000..762ffdafd18
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/PipelineExecutionTaskTest.java
@@ -0,0 +1,76 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.doris.qe.runtime;
+
+import org.apache.doris.common.Status;
+import org.apache.doris.common.UserException;
+import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.nereids.trees.plans.distribute.worker.BackendWorker;
+import org.apache.doris.proto.InternalService.PExecPlanFragmentResult;
+import org.apache.doris.qe.CoordinatorContext;
+import org.apache.doris.rpc.BackendServiceProxy;
+import org.apache.doris.thrift.TQueryOptions;
+import org.apache.doris.thrift.TUniqueId;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentMatchers;
+import org.mockito.Mockito;
+
+import java.util.Collections;
+import java.util.concurrent.Callable;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Supplier;
+
+class PipelineExecutionTaskTest {
+ @Test
+ void expiredDeadlineCancelsAlreadySubmittedFragments() throws Exception {
+ CoordinatorContext coordinatorContext =
Mockito.mock(CoordinatorContext.class);
+ TQueryOptions queryOptions = new TQueryOptions();
+ queryOptions.setExecutionTimeout(1);
+ queryOptions.setQueryTimeout(1);
+ Deencapsulation.setField(coordinatorContext, "queryOptions",
queryOptions);
+ Deencapsulation.setField(coordinatorContext, "queryId", new
TUniqueId(1, 2));
+ Deencapsulation.setField(coordinatorContext, "timeoutDeadline",
(Supplier<Long>) () -> 0L);
+
Mockito.when(coordinatorContext.withLock(ArgumentMatchers.<Callable<Object>>any()))
+ .thenAnswer(invocation ->
invocation.<Callable<Object>>getArgument(0).call());
+ Mockito.when(coordinatorContext.twoPhaseExecution()).thenReturn(false);
+
+ MultiFragmentsPipelineTask fragmentsTask =
Mockito.mock(MultiFragmentsPipelineTask.class);
+
Mockito.when(fragmentsTask.getChildrenTasks()).thenReturn(Collections.emptyMap());
+ Mockito.when(fragmentsTask.sendPhaseOneRpc(false))
+
.thenReturn(CompletableFuture.completedFuture(PExecPlanFragmentResult.getDefaultInstance()));
+ PipelineExecutionTask executionTask = new PipelineExecutionTask(
+ coordinatorContext,
+ Mockito.mock(BackendServiceProxy.class),
+ Collections.singletonMap(Mockito.mock(BackendWorker.class),
fragmentsTask));
+
+ UserException exception = Assertions.assertThrows(UserException.class,
executionTask::execute);
+
+ Assertions.assertTrue(exception.getMessage().contains("timeout before
waiting send fragments rpc"));
+ Mockito.verify(fragmentsTask).sendPhaseOneRpc(false);
+
Mockito.verify(coordinatorContext).updateStatusIfOk(ArgumentMatchers.argThat(
+ status -> hasDeadlineTimeoutMessage(status)));
+
Mockito.verify(coordinatorContext).cancelSchedule(ArgumentMatchers.argThat(
+ status -> hasDeadlineTimeoutMessage(status)));
+ }
+
+ private static boolean hasDeadlineTimeoutMessage(Status status) {
+ return status.getErrorMsg().contains("timeout before waiting send
fragments rpc");
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
