diff --git a/be/src/exec/sink/writer/vtablet_writer.cpp b/be/src/exec/sink/writer/vtablet_writer.cpp index 4159ee08b51447..46d9432d77c430 100644 --- a/be/src/exec/sink/writer/vtablet_writer.cpp +++ b/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( - "VTabletWriter.close.sleep", "sleep_sec", 1); + auto sleep_sec = dp->param("sleep_sec", 1); + auto token = dp->param("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 48324fc90a92c9..2dc07157773cc5 100644 --- a/be/src/exec/sink/writer/vtablet_writer_v2.cpp +++ b/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( - "VTabletWriterV2.close.sleep", "sleep_sec", 1); + auto sleep_sec = dp->param("sleep_sec", 1); + auto token = dp->param("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 38ba2ebbc79501..4e5a6d804cd372 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.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> latch; private volatile Optional> topFragmentLatch; - private volatile List 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 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; } 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 1aa4c7f0478e50..c6110d6a35be01 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 instanceIds; // mutate states @@ -48,6 +51,7 @@ public SingleFragmentPipelineTask(Backend backend, int fragmentId, Set 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 00000000000000..31becae01dc9db --- /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.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 00000000000000..e194ebe31887f2 --- /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() + } + } +}