This is an automated email from the ASF dual-hosted git repository.
yujun777 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new a565aca4478 [fix](regression) Stop MTMV task waits from latching onto
the previous task (#67710)
a565aca4478 is described below
commit a565aca4478ee22bb5f9332fd0ba69668dcb001e
Author: yujun <[email protected]>
AuthorDate: Thu Sep 10 11:32:59 2026 +0800
[fix](regression) Stop MTMV task waits from latching onto the previous task
(#67710)
The shared MTMV task waits (`waitingMTMVTaskFinishedByMvName`,
`waitingMTMVTaskFinishedByMvNameAllowCancel`, `waitingMTMVTaskFinished`,
`waitingMTMVTaskFinishedWithoutAnalyze`,
`waitingMTMVTaskFinishedNotNeedSuccess`) read the newest row of
`tasks('type'='mv')` with `order by CreateTime DESC limit 1` and trusted
it as the task they had just submitted. A task that just finished is
briefly missing from that list: the job removes it from its running list
before the MV history gets it, so a poll that lands in that window reads
the previous task of the same MV, whose status is already terminal. The
committed-not-visible regression hit this on CI (expected SUCCESS but
read the previous FAILED task).
Key changes:
- Add `pollMTMVTaskTerminal()`: a terminal row is trusted only when the
same task was seen running by this wait, or is seen terminal twice in a
row and no earlier wait in the suite returned it (so a wait cannot latch
onto an already-reported task).
- The five `waitingMTMVTaskFinished*` helpers now share that wait
instead of five copies of the polling loop; their SQL, logging,
assertions and analyze step are unchanged.
- Add `SuiteMTMVTaskWaitTest` covering a task seen running, a terminal
row that needs confirmation, the previous task showing through the gap,
and a repeated wait for a task that already finished (bounded fallback).
---
.../org/apache/doris/regression/suite/Suite.groovy | 212 +++++++++------------
.../regression/suite/SuiteMTMVTaskWaitTest.groovy | 76 ++++++++
2 files changed, 168 insertions(+), 120 deletions(-)
diff --git
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
index 875029ec84f..d906c610ed8 100644
---
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
+++
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy
@@ -113,6 +113,10 @@ class Suite implements GroovyInterceptable {
private AmazonS3 s3Client = null
private FileSystem fs = null
+ // MTMV task ids that a wait in this suite already returned as terminal. A
task that just
+ // finished is briefly absent from tasks(), so a later wait can fall back
to one of these.
+ private final Set<String> finishedMTMVTaskIds =
Collections.synchronizedSet(new HashSet<>())
+
Suite(String name, String group, SuiteContext context, SuiteCluster
cluster) {
this.name = name
this.group = group
@@ -2019,39 +2023,99 @@ class Suite implements GroovyInterceptable {
return debugPoint
}
- def waitingMTMVTaskFinishedByMvName = { mvName, dbName = context.dbName ->
- // Wait for the newly submitted MTMV task to become visible in tasks().
- Thread.sleep(2000);
- String showTasks = """
- select TaskId, Status, MvName, MvDatabaseName from
tasks('type'='mv')
- where MvDatabaseName = '${dbName}' and MvName = '${mvName}'
- order by CreateTime DESC limit 1
- """
+ /**
+ * Poll a tasks('type'='mv') query until the newest row is terminal and
return that row.
+ * A task that just finished is briefly missing from tasks(): the job
removes it from its
+ * running list before the MV history gets it, so while that window is
open the newest row
+ * is the previous task of the same MV. A terminal row is therefore
trusted only when the
+ * same task was already seen running by this wait, or is seen terminal
twice in a row and
+ * no earlier wait returned it; the window lasts tens of milliseconds, far
less than the
+ * poll interval, so one extra poll is enough to tell the new task from
the previous one
+ * even when the previous task was not waited for before.
+ * poll() runs the query once, log() receives the progress messages, and
the two intervals
+ * exist so tests can shorten them. The last row seen is returned when the
timeout expires,
+ * so callers can report a status.
+ */
+ static List<Object> pollMTMVTaskTerminal(String showTasks, String caller,
Set<String> finishedTaskIds,
+ Closure<List<List<Object>>> poll, Closure<String> log, long
pollIntervalMs, long skipFinishedMs) {
String status = "NULL"
- List<List<Object>> result
- long timeoutTimestamp = System.currentTimeMillis() + 5 * 60 * 1000 //
5 min
+ String runningTaskId = null
+ String confirmedTaskId = null
String lastLoggedStatus = null
- List<Object> toCheckTaskRow = null
- while (timeoutTimestamp > System.currentTimeMillis() && (status ==
'PENDING' || status == 'RUNNING' || status == 'NULL')) {
- result = sql(showTasks)
+ List<Object> taskRow = null
+ long skipFinishedDeadline = 0
+ long timeoutTimestamp = System.currentTimeMillis() + 5 * 60 * 1000 //
5 min
+ while (timeoutTimestamp > System.currentTimeMillis()) {
+ List<List<Object>> result = poll()
if (result.isEmpty()) {
if (lastLoggedStatus != "NULL") {
- logger.info("waitingMTMVTaskFinishedByMvName
toCheckTaskRow is empty")
+ log("${caller} task row is empty")
lastLoggedStatus = "NULL"
}
- Thread.sleep(500);
+ confirmedTaskId = null
+ Thread.sleep(pollIntervalMs);
continue;
}
- toCheckTaskRow = result[0]
- status = toCheckTaskRow.get(1).toString()
+ taskRow = result[0]
+ String taskId = taskRow.get(0).toString()
+ status = taskRow.get(1).toString()
if (lastLoggedStatus != status) {
- logger.info("The state of ${showTasks} is ${status}, taskId is
${toCheckTaskRow.get(0)}")
+ log("The state of ${showTasks} is ${status}, taskId is
${taskId}")
lastLoggedStatus = status
}
- if (status == 'PENDING' || status == 'RUNNING' || status ==
'NULL') {
- Thread.sleep(500);
+ if (status == 'PENDING' || status == 'RUNNING') {
+ runningTaskId = taskId
+ confirmedTaskId = null
+ Thread.sleep(pollIntervalMs);
+ continue;
}
+ if (finishedTaskIds.contains(taskId)) {
+ // A wait already returned this task, so it is the previous
task and the one
+ // being waited for is still in the gap. Skip it for as long
as any real gap
+ // could last; a caller that waits twice for the same task
(nothing new was
+ // submitted) then falls back to it instead of hitting the 5
minute timeout.
+ if (skipFinishedDeadline == 0) {
+ skipFinishedDeadline = System.currentTimeMillis() +
skipFinishedMs
+ }
+ if (System.currentTimeMillis() < skipFinishedDeadline) {
+ confirmedTaskId = null
+ Thread.sleep(pollIntervalMs);
+ continue;
+ }
+ }
+ if (taskId == runningTaskId || taskId == confirmedTaskId) {
+ finishedTaskIds.add(taskId)
+ return taskRow
+ }
+ // The first sighting of a terminal task may be the previous task
seen through the
+ // gap described above; confirm the same task once more before
trusting it.
+ confirmedTaskId = taskId
+ Thread.sleep(pollIntervalMs);
}
+ return taskRow
+ }
+
+ /**
+ * Wait for the newest MTMV task matching showTasks to reach a terminal
state and return
+ * its row. See pollMTMVTaskTerminal for why a terminal row needs to be
confirmed.
+ */
+ List<Object> waitMTMVTaskTerminal(String showTasks, String caller,
+ long pollIntervalMs = 500, long skipFinishedMs = 30 * 1000) {
+ return pollMTMVTaskTerminal(showTasks, caller, finishedMTMVTaskIds,
+ { -> sql(showTasks) }, { String message ->
logger.info(message) },
+ pollIntervalMs, skipFinishedMs)
+ }
+
+ def waitingMTMVTaskFinishedByMvName = { mvName, dbName = context.dbName ->
+ // Wait for the newly submitted MTMV task to become visible in tasks().
+ Thread.sleep(2000);
+ String showTasks = """
+ select TaskId, Status, MvName, MvDatabaseName from
tasks('type'='mv')
+ where MvDatabaseName = '${dbName}' and MvName = '${mvName}'
+ order by CreateTime DESC limit 1
+ """
+ List<Object> toCheckTaskRow = waitMTMVTaskTerminal(showTasks,
"waitingMTMVTaskFinishedByMvName")
+ String status = toCheckTaskRow == null ? "NULL" :
toCheckTaskRow.get(1).toString()
if (status != "SUCCESS") {
logger.info("status is ${status}")
}
@@ -2071,31 +2135,8 @@ class Suite implements GroovyInterceptable {
order by CreateTime DESC limit 1
"""
- String status = "NULL"
- List<List<Object>> result
- long timeoutTimestamp = System.currentTimeMillis() + 5 * 60 * 1000 //
5 min
- String lastLoggedStatus = null
- List<Object> toCheckTaskRow = null
- while (timeoutTimestamp > System.currentTimeMillis() && (status ==
'PENDING' || status == 'RUNNING' || status == 'NULL')) {
- result = sql(showTasks)
- if (result.isEmpty()) {
- if (lastLoggedStatus != "NULL") {
- logger.info("waitingMTMVTaskFinishedByMvName
toCheckTaskRow is empty")
- lastLoggedStatus = "NULL"
- }
- Thread.sleep(500);
- continue;
- }
- toCheckTaskRow = result[0]
- status = toCheckTaskRow.get(1).toString()
- if (lastLoggedStatus != status) {
- logger.info("The state of ${showTasks} is ${status}, taskId is
${toCheckTaskRow.get(0)}")
- lastLoggedStatus = status
- }
- if (status == 'PENDING' || status == 'RUNNING' || status == 'NULL'
|| status == 'CANCELED') {
- Thread.sleep(500);
- }
- }
+ List<Object> toCheckTaskRow = waitMTMVTaskTerminal(showTasks,
"waitingMTMVTaskFinishedByMvNameAllowCancel")
+ String status = toCheckTaskRow == null ? "NULL" :
toCheckTaskRow.get(1).toString()
if (status != "SUCCESS") {
logger.info("status is not success")
Assert.assertNotNull(toCheckTaskRow)
@@ -2183,31 +2224,8 @@ class Suite implements GroovyInterceptable {
select TaskId, Status, MvName, MvDatabaseName from
tasks('type'='mv')
where JobName = '${jobName}' order by CreateTime DESC limit 1
"""
- String status = "NULL"
- List<List<Object>> result
- long timeoutTimestamp = System.currentTimeMillis() + 5 * 60 * 1000 //
5 min
- String lastLoggedStatus = null
- List<Object> taskRow = null
- do {
- result = sql(showTasks)
- if (result.isEmpty()) {
- if (lastLoggedStatus != "NULL") {
- logger.info("waitingMTMVTaskFinished task row is empty")
- lastLoggedStatus = "NULL"
- }
- status = "NULL"
- } else {
- taskRow = result[0]
- status = taskRow.get(1).toString()
- if (lastLoggedStatus != status) {
- logger.info("The state of ${showTasks} is ${status},
taskId is ${taskRow.get(0)}")
- lastLoggedStatus = status
- }
- }
- if (status == 'PENDING' || status == 'RUNNING' || status ==
'NULL') {
- Thread.sleep(500);
- }
- } while (timeoutTimestamp > System.currentTimeMillis() && (status ==
'PENDING' || status == 'RUNNING' || status == 'NULL'))
+ List<Object> taskRow = waitMTMVTaskTerminal(showTasks,
"waitingMTMVTaskFinished")
+ String status = taskRow == null ? "NULL" : taskRow.get(1).toString()
if (status != "SUCCESS") {
logger.info("status is ${status}")
}
@@ -2226,31 +2244,8 @@ class Suite implements GroovyInterceptable {
select TaskId, Status from tasks('type'='mv')
where JobName = '${jobName}' order by CreateTime DESC limit 1
"""
- String status = "NULL"
- List<List<Object>> result
- long timeoutTimestamp = System.currentTimeMillis() + 5 * 60 * 1000 //
5 min
- String lastLoggedStatus = null
- List<Object> taskRow = null
- do {
- result = sql(showTasks)
- if (result.isEmpty()) {
- if (lastLoggedStatus != "NULL") {
- logger.info("waitingMTMVTaskFinishedWithoutAnalyze task
row is empty")
- lastLoggedStatus = "NULL"
- }
- status = "NULL"
- } else {
- taskRow = result[0]
- status = taskRow.get(1).toString()
- if (lastLoggedStatus != status) {
- logger.info("The state of ${showTasks} is ${status},
taskId is ${taskRow.get(0)}")
- lastLoggedStatus = status
- }
- }
- if (status == 'PENDING' || status == 'RUNNING' || status ==
'NULL') {
- Thread.sleep(500);
- }
- } while (timeoutTimestamp > System.currentTimeMillis() && (status ==
'PENDING' || status == 'RUNNING' || status == 'NULL'))
+ List<Object> taskRow = waitMTMVTaskTerminal(showTasks,
"waitingMTMVTaskFinishedWithoutAnalyze")
+ String status = taskRow == null ? "NULL" : taskRow.get(1).toString()
if (status != "SUCCESS") {
logger.info("status is not success")
}
@@ -2264,31 +2259,8 @@ class Suite implements GroovyInterceptable {
select TaskId, Status from tasks('type'='mv')
where JobName = '${jobName}' order by CreateTime DESC limit 1
"""
- String status = "NULL"
- List<List<Object>> result
- long timeoutTimestamp = System.currentTimeMillis() + 5 * 60 * 1000 //
5 min
- String lastLoggedStatus = null
- List<Object> taskRow = null
- do {
- result = sql(showTasks)
- if (result.isEmpty()) {
- if (lastLoggedStatus != "NULL") {
- logger.info("waitingMTMVTaskFinishedNotNeedSuccess task
row is empty")
- lastLoggedStatus = "NULL"
- }
- status = "NULL"
- } else {
- taskRow = result[0]
- status = taskRow.get(1).toString()
- if (lastLoggedStatus != status) {
- logger.info("The state of ${showTasks} is ${status},
taskId is ${taskRow.get(0)}")
- lastLoggedStatus = status
- }
- }
- if (status == 'PENDING' || status == 'RUNNING' || status ==
'NULL') {
- Thread.sleep(500);
- }
- } while (timeoutTimestamp > System.currentTimeMillis() && (status ==
'PENDING' || status == 'RUNNING' || status == 'NULL'))
+ List<Object> taskRow = waitMTMVTaskTerminal(showTasks,
"waitingMTMVTaskFinishedNotNeedSuccess")
+ String status = taskRow == null ? "NULL" : taskRow.get(1).toString()
if (status != "SUCCESS") {
logger.info("status is not success")
}
diff --git
a/regression-test/framework/src/test/groovy/org/apache/doris/regression/suite/SuiteMTMVTaskWaitTest.groovy
b/regression-test/framework/src/test/groovy/org/apache/doris/regression/suite/SuiteMTMVTaskWaitTest.groovy
new file mode 100644
index 00000000000..39e5a7d8ea6
--- /dev/null
+++
b/regression-test/framework/src/test/groovy/org/apache/doris/regression/suite/SuiteMTMVTaskWaitTest.groovy
@@ -0,0 +1,76 @@
+// 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.regression.suite
+
+import org.junit.jupiter.api.Test
+
+import static org.junit.jupiter.api.Assertions.assertEquals
+
+class SuiteMTMVTaskWaitTest {
+ private static final String SHOW_TASKS = "select TaskId, Status from
tasks('type'='mv')"
+ private static final Closure<String> IGNORE_LOG = { String message -> }
+
+ private static List<Object> task(String taskId, String status) {
+ return [taskId, status] as List<Object>
+ }
+
+ private static Closure<List<List<Object>>> polls(List<List<List<Object>>>
script) {
+ Deque<List<List<Object>>> queue = new ArrayDeque<>(script)
+ return { ->
+ if (queue.isEmpty()) {
+ throw new IllegalStateException("poll script exhausted, the
wait did not converge")
+ }
+ return queue.poll()
+ }
+ }
+
+ private static String waitFor(Set<String> finished,
List<List<List<Object>>> script) {
+ return Suite.pollMTMVTaskTerminal(SHOW_TASKS, "test", finished,
polls(script), IGNORE_LOG, 1, 5)
+ .get(0).toString()
+ }
+
+ @Test
+ void followsATaskThatWasSeenRunning() {
+ assertEquals("X", waitFor(new HashSet<>(), [[task("X", "RUNNING")],
[task("X", "SUCCESS")]]))
+ }
+
+ @Test
+ void confirmsATerminalRowThatWasNeverSeenRunning() {
+ assertEquals("X", waitFor(new HashSet<>(), [[task("X", "SUCCESS")],
[task("X", "SUCCESS")]]))
+ }
+
+ @Test
+ void ignoresThePreviousTaskWhileTheNewTaskIsMissing() {
+ Set<String> finished = new HashSet<>()
+ assertEquals("P", waitFor(finished, [[task("P", "RUNNING")],
[task("P", "SUCCESS")]]))
+
+ // X runs and finishes; while it moves from the running list to the MV
history the
+ // newest row of the MV is P again, and P must not be returned a
second time.
+ assertEquals("X", waitFor(finished, [[task("X", "RUNNING")],
[task("P", "SUCCESS")],
+ [task("X", "SUCCESS")]]))
+ }
+
+ @Test
+ void fallsBackToTheSameTaskWhenNothingNewIsSubmitted() {
+ Set<String> finished = new HashSet<>()
+ assertEquals("P", waitFor(finished, [[task("P", "RUNNING")],
[task("P", "SUCCESS")]]))
+
+ // Nothing new was submitted: the wait must return P instead of
polling to the timeout.
+ assertEquals("P", waitFor(finished, (1..100).collect { [task("P",
"SUCCESS")] }))
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
