[fix](transaction) Fail fast when a backend restarts during insert

be/src/exec/sink/writer/vtablet_writer.cpp

@@ -2066,8 +2066,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",

be/src/exec/sink/writer/vtablet_writer_v2.cpp

@@ -670,8 +670,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",

fe/fe-core/src/main/java/org/apache/doris/qe/runtime/LoadProcessor.java

@@ -30,7 +30,6 @@
 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 LoadProcessor(CoordinatorContext coordinatorContext, long jobId) {
  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 @@ protected void afterSetPipelineExecutionTask(PipelineExecutionTask pipelineExecu
  }
  }
  }
- this.topFragmentTasks = topFragmentTasks;
-
  // only wait top fragments
  MarkedCountDownLatch<Integer, Long> topFragmentLatch = new MarkedCountDownLatch<>(topFragmentTasks.size());
  for (SingleFragmentPipelineTask topFragmentTask : topFragmentTasks) {
@@ -261,16 +255,14 @@ protected void doProcessReportExecStatus(TReportExecStatusParams params, SingleF
  }
  }
 
- /*
- * 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;
  }

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 SingleFragmentPipelineTask(Backend backend, int fragmentId, Set<TUniqueId
  this.fragmentId = fragmentId;
  this.instanceIds = instanceIds;
  this.lastMissingHeartbeatTime = backend.getLastMissingHeartbeatTime();
+ this.backendProcessEpoch = backend.getProcessEpoch();
  }
 
  // update profile.
@@ -62,12 +66,21 @@ public boolean processReportExecStatus(TReportExecStatusParams reportExecStatus)
  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() {

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());
+ }
+}

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()
+ }
+ }
+}