github-actions[bot] commented on code in PR #65525:
URL: https://github.com/apache/doris/pull/65525#discussion_r3569202349
##########
be/src/exec/sink/writer/vtablet_writer.cpp:
##########
@@ -2068,6 +2068,9 @@ Status VTabletWriter::close(Status exec_status) {
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 token =
DebugPoints::instance()->get_debug_param_or_default<std::string>(
Review Comment:
Inside `DBUG_EXECUTE_IF`, `dp` has already been fetched by the macro.
Calling `DebugPoints::instance()->get_debug_param_or_default()` fetches the
same point again, and `get_debug_point()` advances/removes `execute`-limited
points. With this new second lookup, a point such as
`VTabletWriter.close.sleep?sleep_sec=300&execute=2&token=t` enters the block,
reads `sleep_sec`, then the token read consumes the limit and returns the
default empty token/removes the point on the first hit. Please read both params
from the existing `dp` (`dp->param<int32_t>("sleep_sec", 1)` and
`dp->param<std::string>("token", "")`), and make the same change in
`VTabletWriterV2.close.sleep`.
##########
regression-test/suites/load_p0/insert/test_insert_fail_fast_after_be_restart.groovy:
##########
@@ -0,0 +1,106 @@
+// 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
+
+ docker(options) {
+ def tableName = "test_insert_fail_fast_after_be_restart"
+ GetDebugPoint().clearDebugPointsForAllBEs()
+
+ try {
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ sql """
+ CREATE TABLE ${tableName} (
+ k BIGINT NOT NULL,
+ v BIGINT NOT NULL
+ )
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ )
+ """
+
+ // 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 ${tableName}
+ 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()
+
+ // 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)
Review Comment:
The tokenized log wait fixes the earlier race where the restart could happen
before close reached the debug point, but the test can still fail through the
old backend-down path instead of proving the new epoch check.
`restartBackends()` may wait up to 120s, while `LoadProcessor.join()` polls
health every 30s and `getBackendHealthStatus()` checks `!backend.isAlive()`
before comparing process epochs. On a slow docker restart, the insert can
return `backend 1 is down`, so these assertions fail without exercising the
restarted-but-alive epoch branch. Please also synchronize until FE has observed
the restarted BE as alive with a changed positive epoch before expecting the
epoch error, or avoid asserting that exact message as the proof.
##########
regression-test/suites/load_p0/insert/test_insert_fail_fast_after_be_restart.groovy:
##########
@@ -0,0 +1,106 @@
+// 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
+
+ docker(options) {
+ def tableName = "test_insert_fail_fast_after_be_restart"
+ GetDebugPoint().clearDebugPointsForAllBEs()
+
+ try {
+ sql "DROP TABLE IF EXISTS ${tableName}"
+ sql """
+ CREATE TABLE ${tableName} (
+ k BIGINT NOT NULL,
+ v BIGINT NOT NULL
+ )
+ DUPLICATE KEY(k)
+ DISTRIBUTED BY HASH(k) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ )
+ """
+
+ // 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 ${tableName}
+ 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()
+
+ // 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 ${tableName}")[0][0] as
int)
+ } finally {
+ GetDebugPoint().clearDebugPointsForAllBEs()
+ try_sql "DROP TABLE IF EXISTS ${tableName}"
Review Comment:
Please keep only the pre-test `DROP TABLE IF EXISTS` and leave the table
after the test for debugging, per the regression-test standard. Since this is
also a single ordinary test table, the same rule asks us to hardcode
`test_insert_fail_fast_after_be_restart` in the SQL instead of routing it
through `def tableName`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
