This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 7674d494cc5 branch-4.1: [fix](transaction) Fail fast when a backend
restarts during insert #65525 (#65566)
7674d494cc5 is described below
commit 7674d494cc5b9e0491821a66e3587ef4db7e7c82
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Tue Jul 14 17:45:41 2026 +0800
branch-4.1: [fix](transaction) Fail fast when a backend restarts during
insert #65525 (#65566)
Cherry-picked from #65525
Co-authored-by: hui lai <[email protected]>
---
be/src/exec/sink/writer/vtablet_writer.cpp | 5 +-
be/src/exec/sink/writer/vtablet_writer_v2.cpp | 5 +-
.../org/apache/doris/qe/runtime/LoadProcessor.java | 22 ++--
.../qe/runtime/SingleFragmentPipelineTask.java | 19 ++-
.../qe/runtime/SingleFragmentPipelineTaskTest.java | 96 ++++++++++++++++
.../test_insert_fail_fast_after_be_restart.groovy | 128 +++++++++++++++++++++
6 files changed, 253 insertions(+), 22 deletions(-)
diff --git a/be/src/exec/sink/writer/vtablet_writer.cpp
b/be/src/exec/sink/writer/vtablet_writer.cpp
index 2ac24fbf402..2ba4d139016 100644
--- a/be/src/exec/sink/writer/vtablet_writer.cpp
+++ b/be/src/exec/sink/writer/vtablet_writer.cpp
@@ -2071,8 +2071,9 @@ Status VTabletWriter::close(Status exec_status) {
TEST_INJECTION_POINT("VOlapTableSink::close");
DBUG_EXECUTE_IF("VTabletWriter.close.sleep", {
- auto sleep_sec =
DebugPoints::instance()->get_debug_param_or_default<int32_t>(
- "VTabletWriter.close.sleep", "sleep_sec", 1);
+ auto sleep_sec = dp->param<int32_t>("sleep_sec", 1);
+ auto token = dp->param<std::string>("token", "");
+ LOG(INFO) << "hit debug point VTabletWriter.close.sleep, token=" <<
token;
std::this_thread::sleep_for(std::chrono::seconds(sleep_sec));
});
DBUG_EXECUTE_IF("VTabletWriter.close.close_status_not_ok",
diff --git a/be/src/exec/sink/writer/vtablet_writer_v2.cpp
b/be/src/exec/sink/writer/vtablet_writer_v2.cpp
index c4708a01164..ce78cb3b270 100644
--- a/be/src/exec/sink/writer/vtablet_writer_v2.cpp
+++ b/be/src/exec/sink/writer/vtablet_writer_v2.cpp
@@ -671,8 +671,9 @@ Status VTabletWriterV2::close(Status exec_status) {
}
DBUG_EXECUTE_IF("VTabletWriterV2.close.sleep", {
- auto sleep_sec =
DebugPoints::instance()->get_debug_param_or_default<int32_t>(
- "VTabletWriterV2.close.sleep", "sleep_sec", 1);
+ auto sleep_sec = dp->param<int32_t>("sleep_sec", 1);
+ auto token = dp->param<std::string>("token", "");
+ LOG(INFO) << "hit debug point VTabletWriterV2.close.sleep, token=" <<
token;
std::this_thread::sleep_for(std::chrono::seconds(sleep_sec));
});
DBUG_EXECUTE_IF("VTabletWriterV2.close.cancel",
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/LoadProcessor.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/LoadProcessor.java
index 45995a7aad7..a605f3039aa 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/LoadProcessor.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/LoadProcessor.java
@@ -30,7 +30,6 @@ import org.apache.doris.qe.CoordinatorContext;
import org.apache.doris.qe.LoadContext;
import org.apache.doris.thrift.TFragmentInstanceReport;
import org.apache.doris.thrift.TReportExecStatusParams;
-import org.apache.doris.thrift.TStatusCode;
import org.apache.doris.thrift.TUniqueId;
import com.google.common.collect.Lists;
@@ -53,7 +52,6 @@ public class LoadProcessor extends AbstractJobProcessor {
// key: fragmentId, value: backendId
private volatile Optional<MarkedCountDownLatch<Integer, Long>> latch;
private volatile Optional<MarkedCountDownLatch<Integer, Long>>
topFragmentLatch;
- private volatile List<SingleFragmentPipelineTask> topFragmentTasks;
public LoadProcessor(CoordinatorContext coordinatorContext, long jobId) {
super(coordinatorContext);
@@ -76,8 +74,6 @@ public class LoadProcessor extends AbstractJobProcessor {
String.valueOf(jobId), coordinatorContext.scanRangeNum.get()
);
- topFragmentTasks = Lists.newArrayList();
-
LOG.info("dispatch load job: {} to {}",
DebugUtil.printId(queryId),
coordinatorContext.backends.get().keySet()
);
@@ -105,8 +101,6 @@ public class LoadProcessor extends AbstractJobProcessor {
}
}
}
- this.topFragmentTasks = topFragmentTasks;
-
// only wait top fragments
MarkedCountDownLatch<Integer, Long> topFragmentLatch = new
MarkedCountDownLatch<>(topFragmentTasks.size());
for (SingleFragmentPipelineTask topFragmentTask : topFragmentTasks) {
@@ -255,16 +249,14 @@ public class LoadProcessor extends AbstractJobProcessor {
}
}
- /*
- * Check the state of backends in needCheckBackendExecStates.
- * return true if all of them are OK. Otherwise, return false.
- */
+ // Check backend health for every unfinished load fragment task.
private boolean checkHealthy() {
- for (SingleFragmentPipelineTask topFragmentTask : topFragmentTasks) {
- if (!topFragmentTask.isBackendHealthy(jobId)) {
- long backendId = topFragmentTask.getBackend().getId();
- Status unhealthyStatus = new Status(
- TStatusCode.INTERNAL_ERROR, "backend " + backendId + "
is down");
+ for (SingleFragmentPipelineTask fragmentTask :
backendFragmentTasks.get().values()) {
+ if (fragmentTask.isDone()) {
+ continue;
+ }
+ Status unhealthyStatus =
fragmentTask.getBackendHealthStatus(jobId);
+ if (!unhealthyStatus.ok()) {
coordinatorContext.updateStatusIfOk(unhealthyStatus);
return false;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/SingleFragmentPipelineTask.java
b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/SingleFragmentPipelineTask.java
index 1aa4c7f0478..c6110d6a35b 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/SingleFragmentPipelineTask.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/qe/runtime/SingleFragmentPipelineTask.java
@@ -17,10 +17,12 @@
package org.apache.doris.qe.runtime;
+import org.apache.doris.common.Status;
import org.apache.doris.qe.QueryStatisticsItem.FragmentInstanceInfo;
import org.apache.doris.system.Backend;
import org.apache.doris.thrift.TNetworkAddress;
import org.apache.doris.thrift.TReportExecStatusParams;
+import org.apache.doris.thrift.TStatusCode;
import org.apache.doris.thrift.TUniqueId;
import com.google.common.collect.Lists;
@@ -38,6 +40,7 @@ public class SingleFragmentPipelineTask extends
LeafRuntimeTask {
private final Backend backend;
private final int fragmentId;
private final long lastMissingHeartbeatTime;
+ private final long backendProcessEpoch;
private final Set<TUniqueId> instanceIds;
// mutate states
@@ -48,6 +51,7 @@ public class SingleFragmentPipelineTask extends
LeafRuntimeTask {
this.fragmentId = fragmentId;
this.instanceIds = instanceIds;
this.lastMissingHeartbeatTime = backend.getLastMissingHeartbeatTime();
+ this.backendProcessEpoch = backend.getProcessEpoch();
}
// update profile.
@@ -62,12 +66,21 @@ public class SingleFragmentPipelineTask extends
LeafRuntimeTask {
return this.done.compareAndSet(false, true);
}
- public boolean isBackendHealthy(long jobId) {
+ public Status getBackendHealthStatus(long jobId) {
if (backend.getLastMissingHeartbeatTime() > lastMissingHeartbeatTime
&& !backend.isAlive()) {
LOG.warn("backend {} is down while joining the coordinator. job
id: {}", backend.getId(), jobId);
- return false;
+ return new Status(TStatusCode.INTERNAL_ERROR, "backend {} is
down", backend.getId());
+ }
+
+ long currentProcessEpoch = backend.getProcessEpoch();
+ if (backendProcessEpoch > 0 && currentProcessEpoch > 0 &&
backendProcessEpoch != currentProcessEpoch) {
+ Status unhealthyStatus = new Status(TStatusCode.INTERNAL_ERROR,
+ "backend {} process epoch changed from {} to {},
indicating that the backend restarted",
+ backend.getId(), backendProcessEpoch, currentProcessEpoch);
+ LOG.warn("{} while joining the coordinator. job id: {}",
unhealthyStatus.getErrorMsg(), jobId);
+ return unhealthyStatus;
}
- return true;
+ return Status.OK;
}
public boolean isDone() {
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/SingleFragmentPipelineTaskTest.java
b/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/SingleFragmentPipelineTaskTest.java
new file mode 100644
index 00000000000..31becae01dc
--- /dev/null
+++
b/fe/fe-core/src/test/java/org/apache/doris/qe/runtime/SingleFragmentPipelineTaskTest.java
@@ -0,0 +1,96 @@
+// 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.system.Backend;
+import org.apache.doris.thrift.TStatusCode;
+import org.apache.doris.thrift.TUniqueId;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+import java.util.Collections;
+
+class SingleFragmentPipelineTaskTest {
+ @Test
+ void backendWithUnchangedProcessEpochIsHealthy() {
+ Backend backend = createBackend(100L);
+ SingleFragmentPipelineTask task = createTask(backend);
+
+ Assertions.assertTrue(task.getBackendHealthStatus(-1L).ok());
+ }
+
+ @Test
+ void backendRestartIsUnhealthyEvenWhenBackendIsAlive() {
+ Backend backend = createBackend(100L);
+ SingleFragmentPipelineTask task = createTask(backend);
+
+ backend.setLastStartTime(200L);
+
+ Status status = task.getBackendHealthStatus(-1L);
+ Assertions.assertEquals(TStatusCode.INTERNAL_ERROR,
status.getErrorCode());
+ Assertions.assertEquals(
+ "backend 1 process epoch changed from 100 to 200, indicating
that the backend restarted",
+ status.getErrorMsg());
+ }
+
+ @Test
+ void zeroCurrentProcessEpochIsIgnoredForCompatibility() {
+ Backend backend = createBackend(100L);
+ SingleFragmentPipelineTask task = createTask(backend);
+
+ backend.setLastStartTime(0L);
+
+ Assertions.assertTrue(task.getBackendHealthStatus(-1L).ok());
+ }
+
+ @Test
+ void zeroCapturedProcessEpochIsIgnoredForCompatibility() {
+ Backend backend = createBackend(0L);
+ SingleFragmentPipelineTask task = createTask(backend);
+
+ backend.setLastStartTime(100L);
+
+ Assertions.assertTrue(task.getBackendHealthStatus(-1L).ok());
+ }
+
+ @Test
+ void backendDownAfterMissingHeartbeatIsUnhealthy() {
+ Backend backend = createBackend(100L);
+ SingleFragmentPipelineTask task = createTask(backend);
+
+ backend.setLastMissingHeartbeatTime(1L);
+ backend.setAlive(false);
+
+ Status status = task.getBackendHealthStatus(-1L);
+ Assertions.assertEquals(TStatusCode.INTERNAL_ERROR,
status.getErrorCode());
+ Assertions.assertEquals("backend 1 is down", status.getErrorMsg());
+ }
+
+ private static Backend createBackend(long processEpoch) {
+ Backend backend = new Backend(1L, "127.0.0.1", 9050);
+ backend.setAlive(true);
+ backend.setLastStartTime(processEpoch);
+ return backend;
+ }
+
+ private static SingleFragmentPipelineTask createTask(Backend backend) {
+ return new SingleFragmentPipelineTask(backend, 0,
Collections.<TUniqueId>emptySet());
+ }
+}
diff --git
a/regression-test/suites/load_p0/insert/test_insert_fail_fast_after_be_restart.groovy
b/regression-test/suites/load_p0/insert/test_insert_fail_fast_after_be_restart.groovy
new file mode 100644
index 00000000000..e194ebe3188
--- /dev/null
+++
b/regression-test/suites/load_p0/insert/test_insert_fail_fast_after_be_restart.groovy
@@ -0,0 +1,128 @@
+// 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.
+
+import org.apache.doris.regression.suite.ClusterOptions
+
+import java.util.concurrent.TimeUnit
+
+suite("test_insert_fail_fast_after_be_restart", "docker") {
+ def options = new ClusterOptions()
+ options.enableDebugPoints()
+ options.setFeNum(1)
+ options.setBeNum(1)
+ options.cloudMode = false
+ // Keep the BE logically alive during restart so the test exercises the
process-epoch branch,
+ // rather than the existing backend-down branch.
+ options.feConfigs += ["max_backend_heartbeat_failure_tolerance_count=100"]
+
+ docker(options) {
+ GetDebugPoint().clearDebugPointsForAllBEs()
+
+ try {
+ sql "DROP TABLE IF EXISTS test_insert_fail_fast_after_be_restart"
+ sql """
+ CREATE TABLE test_insert_fail_fast_after_be_restart (
+ k BIGINT NOT NULL,
+ v BIGINT NOT NULL
+ )
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ )
+ """
+
+ def backendBeforeRestart = sql_return_maparray("SHOW BACKENDS")[0]
+ def backendId = backendBeforeRestart.BackendId.toString()
+ def previousLastStartTime =
backendBeforeRestart.LastStartTime.toString()
+
assertTrue(backendBeforeRestart.Alive.toString().equalsIgnoreCase("true"))
+
+ // Hold the load fragment before it can report completion to FE.
Restarting the BE
+ // at this point drops that report while the replacement process
quickly becomes alive.
+ def debugPointToken = "insert_restart_${System.nanoTime()}"
+ def debugPointParams = [sleep_sec: 300, token: debugPointToken]
+
GetDebugPoint().enableDebugPointForAllBEs("VTabletWriter.close.sleep",
debugPointParams)
+
GetDebugPoint().enableDebugPointForAllBEs("VTabletWriterV2.close.sleep",
debugPointParams)
+
+ def insertFuture = thread {
+ sql "SET enable_nereids_planner = true"
+ sql "SET enable_fallback_to_original_planner = false"
+ sql "SET insert_timeout = 300"
+ try {
+ sql """
+ INSERT INTO test_insert_fail_fast_after_be_restart
+ SELECT number, number FROM numbers("number" = "1024")
+ """
+ return null
+ } catch (Throwable t) {
+ logger.info("INSERT failed after BE restart as expected:
${t.message}")
+ return t.message
+ }
+ }
+
+ def beLogFile = new File(cluster.getBeByIndex(1).getLogFilePath())
+ def debugPointMarkers = [
+ "hit debug point VTabletWriter.close.sleep,
token=${debugPointToken}",
+ "hit debug point VTabletWriterV2.close.sleep,
token=${debugPointToken}"
+ ]
+ def waitDeadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(60)
+ def debugPointHit = false
+ while (System.nanoTime() < waitDeadline) {
+ if (beLogFile.exists()) {
+ def beLog = beLogFile.text
+ if (debugPointMarkers.any { beLog.contains(it) }) {
+ debugPointHit = true
+ break
+ }
+ }
+ sleep(200)
+ }
+ assertTrue(debugPointHit, "INSERT did not reach the writer close
debug point")
+
+ cluster.restartBackends()
+
+ def restartedBackendObserved = false
+ def heartbeatDeadline = System.nanoTime() +
TimeUnit.SECONDS.toNanos(60)
+ while (System.nanoTime() < heartbeatDeadline) {
+ def currentBackend = sql_return_maparray("SHOW BACKENDS")
+ .find { it.BackendId.toString() == backendId }
+ if (currentBackend != null
+ &&
currentBackend.Alive.toString().equalsIgnoreCase("true")
+ && currentBackend.LastStartTime.toString() !=
previousLastStartTime) {
+ restartedBackendObserved = true
+ break
+ }
+ sleep(200)
+ }
+ assertTrue(restartedBackendObserved,
+ "FE did not observe the restarted BE as alive with a
changed process epoch")
+
+ // LoadProcessor checks backend health every 30 seconds. The
restarted BE is alive,
+ // so this only finishes before insert_timeout when FE also
compares process epochs.
+ def errorMessage = insertFuture.get(60, TimeUnit.SECONDS)
+ assertNotNull(errorMessage, "INSERT should fail after its BE
process restarts")
+ assertTrue(errorMessage.contains("process epoch changed"),
+ "unexpected INSERT error after BE restart:
${errorMessage}")
+ assertTrue(errorMessage.contains("backend restarted"),
+ "INSERT error should explain that the backend restarted:
${errorMessage}")
+
+ assertEquals(0, sql("SELECT COUNT(*) FROM
test_insert_fail_fast_after_be_restart")[0][0] as int)
+ } finally {
+ GetDebugPoint().clearDebugPointsForAllBEs()
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
