branch-4.1: [fix](transaction) Fail fast when a backend restarts during insert #65525

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",

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",

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) {
@@ -255,16 +249,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()
+        }
+    }
+}