Skip to content

Commit a565aca

Browse files
yujun777yujun777
authored
[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).
1 parent 7129a3e commit a565aca

2 files changed

Lines changed: 168 additions & 120 deletions

File tree

‎regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/Suite.groovy‎

Lines changed: 92 additions & 120 deletions
Original file line numberDiff line numberDiff line change
@@ -113,6 +113,10 @@ class Suite implements GroovyInterceptable {
113113
private AmazonS3 s3Client = null
114114
private FileSystem fs = null
115115

116+
// MTMV task ids that a wait in this suite already returned as terminal. A task that just
117+
// finished is briefly absent from tasks(), so a later wait can fall back to one of these.
118+
private final Set<String> finishedMTMVTaskIds = Collections.synchronizedSet(new HashSet<>())
119+
116120
Suite(String name, String group, SuiteContext context, SuiteCluster cluster) {
117121
this.name = name
118122
this.group = group
@@ -2019,39 +2023,99 @@ class Suite implements GroovyInterceptable {
20192023
return debugPoint
20202024
}
20212025

2022-
def waitingMTMVTaskFinishedByMvName = { mvName, dbName = context.dbName ->
2023-
// Wait for the newly submitted MTMV task to become visible in tasks().
2024-
Thread.sleep(2000);
2025-
String showTasks = """
2026-
select TaskId, Status, MvName, MvDatabaseName from tasks('type'='mv')
2027-
where MvDatabaseName = '${dbName}' and MvName = '${mvName}'
2028-
order by CreateTime DESC limit 1
2029-
"""
2026+
/**
2027+
* Poll a tasks('type'='mv') query until the newest row is terminal and return that row.
2028+
* A task that just finished is briefly missing from tasks(): the job removes it from its
2029+
* running list before the MV history gets it, so while that window is open the newest row
2030+
* is the previous task of the same MV. A terminal row is therefore trusted only when the
2031+
* same task was already seen running by this wait, or is seen terminal twice in a row and
2032+
* no earlier wait returned it; the window lasts tens of milliseconds, far less than the
2033+
* poll interval, so one extra poll is enough to tell the new task from the previous one
2034+
* even when the previous task was not waited for before.
2035+
* poll() runs the query once, log() receives the progress messages, and the two intervals
2036+
* exist so tests can shorten them. The last row seen is returned when the timeout expires,
2037+
* so callers can report a status.
2038+
*/
2039+
static List<Object> pollMTMVTaskTerminal(String showTasks, String caller, Set<String> finishedTaskIds,
2040+
Closure<List<List<Object>>> poll, Closure<String> log, long pollIntervalMs, long skipFinishedMs) {
20302041
String status = "NULL"
2031-
List<List<Object>> result
2032-
long timeoutTimestamp = System.currentTimeMillis() + 5 * 60 * 1000 // 5 min
2042+
String runningTaskId = null
2043+
String confirmedTaskId = null
20332044
String lastLoggedStatus = null
2034-
List<Object> toCheckTaskRow = null
2035-
while (timeoutTimestamp > System.currentTimeMillis() && (status == 'PENDING' || status == 'RUNNING' || status == 'NULL')) {
2036-
result = sql(showTasks)
2045+
List<Object> taskRow = null
2046+
long skipFinishedDeadline = 0
2047+
long timeoutTimestamp = System.currentTimeMillis() + 5 * 60 * 1000 // 5 min
2048+
while (timeoutTimestamp > System.currentTimeMillis()) {
2049+
List<List<Object>> result = poll()
20372050
if (result.isEmpty()) {
20382051
if (lastLoggedStatus != "NULL") {
2039-
logger.info("waitingMTMVTaskFinishedByMvName toCheckTaskRow is empty")
2052+
log("${caller} task row is empty")
20402053
lastLoggedStatus = "NULL"
20412054
}
2042-
Thread.sleep(500);
2055+
confirmedTaskId = null
2056+
Thread.sleep(pollIntervalMs);
20432057
continue;
20442058
}
2045-
toCheckTaskRow = result[0]
2046-
status = toCheckTaskRow.get(1).toString()
2059+
taskRow = result[0]
2060+
String taskId = taskRow.get(0).toString()
2061+
status = taskRow.get(1).toString()
20472062
if (lastLoggedStatus != status) {
2048-
logger.info("The state of ${showTasks} is ${status}, taskId is ${toCheckTaskRow.get(0)}")
2063+
log("The state of ${showTasks} is ${status}, taskId is ${taskId}")
20492064
lastLoggedStatus = status
20502065
}
2051-
if (status == 'PENDING' || status == 'RUNNING' || status == 'NULL') {
2052-
Thread.sleep(500);
2066+
if (status == 'PENDING' || status == 'RUNNING') {
2067+
runningTaskId = taskId
2068+
confirmedTaskId = null
2069+
Thread.sleep(pollIntervalMs);
2070+
continue;
20532071
}
2072+
if (finishedTaskIds.contains(taskId)) {
2073+
// A wait already returned this task, so it is the previous task and the one
2074+
// being waited for is still in the gap. Skip it for as long as any real gap
2075+
// could last; a caller that waits twice for the same task (nothing new was
2076+
// submitted) then falls back to it instead of hitting the 5 minute timeout.
2077+
if (skipFinishedDeadline == 0) {
2078+
skipFinishedDeadline = System.currentTimeMillis() + skipFinishedMs
2079+
}
2080+
if (System.currentTimeMillis() < skipFinishedDeadline) {
2081+
confirmedTaskId = null
2082+
Thread.sleep(pollIntervalMs);
2083+
continue;
2084+
}
2085+
}
2086+
if (taskId == runningTaskId || taskId == confirmedTaskId) {
2087+
finishedTaskIds.add(taskId)
2088+
return taskRow
2089+
}
2090+
// The first sighting of a terminal task may be the previous task seen through the
2091+
// gap described above; confirm the same task once more before trusting it.
2092+
confirmedTaskId = taskId
2093+
Thread.sleep(pollIntervalMs);
20542094
}
2095+
return taskRow
2096+
}
2097+
2098+
/**
2099+
* Wait for the newest MTMV task matching showTasks to reach a terminal state and return
2100+
* its row. See pollMTMVTaskTerminal for why a terminal row needs to be confirmed.
2101+
*/
2102+
List<Object> waitMTMVTaskTerminal(String showTasks, String caller,
2103+
long pollIntervalMs = 500, long skipFinishedMs = 30 * 1000) {
2104+
return pollMTMVTaskTerminal(showTasks, caller, finishedMTMVTaskIds,
2105+
{ -> sql(showTasks) }, { String message -> logger.info(message) },
2106+
pollIntervalMs, skipFinishedMs)
2107+
}
2108+
2109+
def waitingMTMVTaskFinishedByMvName = { mvName, dbName = context.dbName ->
2110+
// Wait for the newly submitted MTMV task to become visible in tasks().
2111+
Thread.sleep(2000);
2112+
String showTasks = """
2113+
select TaskId, Status, MvName, MvDatabaseName from tasks('type'='mv')
2114+
where MvDatabaseName = '${dbName}' and MvName = '${mvName}'
2115+
order by CreateTime DESC limit 1
2116+
"""
2117+
List<Object> toCheckTaskRow = waitMTMVTaskTerminal(showTasks, "waitingMTMVTaskFinishedByMvName")
2118+
String status = toCheckTaskRow == null ? "NULL" : toCheckTaskRow.get(1).toString()
20552119
if (status != "SUCCESS") {
20562120
logger.info("status is ${status}")
20572121
}
@@ -2071,31 +2135,8 @@ class Suite implements GroovyInterceptable {
20712135
order by CreateTime DESC limit 1
20722136
"""
20732137

2074-
String status = "NULL"
2075-
List<List<Object>> result
2076-
long timeoutTimestamp = System.currentTimeMillis() + 5 * 60 * 1000 // 5 min
2077-
String lastLoggedStatus = null
2078-
List<Object> toCheckTaskRow = null
2079-
while (timeoutTimestamp > System.currentTimeMillis() && (status == 'PENDING' || status == 'RUNNING' || status == 'NULL')) {
2080-
result = sql(showTasks)
2081-
if (result.isEmpty()) {
2082-
if (lastLoggedStatus != "NULL") {
2083-
logger.info("waitingMTMVTaskFinishedByMvName toCheckTaskRow is empty")
2084-
lastLoggedStatus = "NULL"
2085-
}
2086-
Thread.sleep(500);
2087-
continue;
2088-
}
2089-
toCheckTaskRow = result[0]
2090-
status = toCheckTaskRow.get(1).toString()
2091-
if (lastLoggedStatus != status) {
2092-
logger.info("The state of ${showTasks} is ${status}, taskId is ${toCheckTaskRow.get(0)}")
2093-
lastLoggedStatus = status
2094-
}
2095-
if (status == 'PENDING' || status == 'RUNNING' || status == 'NULL' || status == 'CANCELED') {
2096-
Thread.sleep(500);
2097-
}
2098-
}
2138+
List<Object> toCheckTaskRow = waitMTMVTaskTerminal(showTasks, "waitingMTMVTaskFinishedByMvNameAllowCancel")
2139+
String status = toCheckTaskRow == null ? "NULL" : toCheckTaskRow.get(1).toString()
20992140
if (status != "SUCCESS") {
21002141
logger.info("status is not success")
21012142
Assert.assertNotNull(toCheckTaskRow)
@@ -2183,31 +2224,8 @@ class Suite implements GroovyInterceptable {
21832224
select TaskId, Status, MvName, MvDatabaseName from tasks('type'='mv')
21842225
where JobName = '${jobName}' order by CreateTime DESC limit 1
21852226
"""
2186-
String status = "NULL"
2187-
List<List<Object>> result
2188-
long timeoutTimestamp = System.currentTimeMillis() + 5 * 60 * 1000 // 5 min
2189-
String lastLoggedStatus = null
2190-
List<Object> taskRow = null
2191-
do {
2192-
result = sql(showTasks)
2193-
if (result.isEmpty()) {
2194-
if (lastLoggedStatus != "NULL") {
2195-
logger.info("waitingMTMVTaskFinished task row is empty")
2196-
lastLoggedStatus = "NULL"
2197-
}
2198-
status = "NULL"
2199-
} else {
2200-
taskRow = result[0]
2201-
status = taskRow.get(1).toString()
2202-
if (lastLoggedStatus != status) {
2203-
logger.info("The state of ${showTasks} is ${status}, taskId is ${taskRow.get(0)}")
2204-
lastLoggedStatus = status
2205-
}
2206-
}
2207-
if (status == 'PENDING' || status == 'RUNNING' || status == 'NULL') {
2208-
Thread.sleep(500);
2209-
}
2210-
} while (timeoutTimestamp > System.currentTimeMillis() && (status == 'PENDING' || status == 'RUNNING' || status == 'NULL'))
2227+
List<Object> taskRow = waitMTMVTaskTerminal(showTasks, "waitingMTMVTaskFinished")
2228+
String status = taskRow == null ? "NULL" : taskRow.get(1).toString()
22112229
if (status != "SUCCESS") {
22122230
logger.info("status is ${status}")
22132231
}
@@ -2226,31 +2244,8 @@ class Suite implements GroovyInterceptable {
22262244
select TaskId, Status from tasks('type'='mv')
22272245
where JobName = '${jobName}' order by CreateTime DESC limit 1
22282246
"""
2229-
String status = "NULL"
2230-
List<List<Object>> result
2231-
long timeoutTimestamp = System.currentTimeMillis() + 5 * 60 * 1000 // 5 min
2232-
String lastLoggedStatus = null
2233-
List<Object> taskRow = null
2234-
do {
2235-
result = sql(showTasks)
2236-
if (result.isEmpty()) {
2237-
if (lastLoggedStatus != "NULL") {
2238-
logger.info("waitingMTMVTaskFinishedWithoutAnalyze task row is empty")
2239-
lastLoggedStatus = "NULL"
2240-
}
2241-
status = "NULL"
2242-
} else {
2243-
taskRow = result[0]
2244-
status = taskRow.get(1).toString()
2245-
if (lastLoggedStatus != status) {
2246-
logger.info("The state of ${showTasks} is ${status}, taskId is ${taskRow.get(0)}")
2247-
lastLoggedStatus = status
2248-
}
2249-
}
2250-
if (status == 'PENDING' || status == 'RUNNING' || status == 'NULL') {
2251-
Thread.sleep(500);
2252-
}
2253-
} while (timeoutTimestamp > System.currentTimeMillis() && (status == 'PENDING' || status == 'RUNNING' || status == 'NULL'))
2247+
List<Object> taskRow = waitMTMVTaskTerminal(showTasks, "waitingMTMVTaskFinishedWithoutAnalyze")
2248+
String status = taskRow == null ? "NULL" : taskRow.get(1).toString()
22542249
if (status != "SUCCESS") {
22552250
logger.info("status is not success")
22562251
}
@@ -2264,31 +2259,8 @@ class Suite implements GroovyInterceptable {
22642259
select TaskId, Status from tasks('type'='mv')
22652260
where JobName = '${jobName}' order by CreateTime DESC limit 1
22662261
"""
2267-
String status = "NULL"
2268-
List<List<Object>> result
2269-
long timeoutTimestamp = System.currentTimeMillis() + 5 * 60 * 1000 // 5 min
2270-
String lastLoggedStatus = null
2271-
List<Object> taskRow = null
2272-
do {
2273-
result = sql(showTasks)
2274-
if (result.isEmpty()) {
2275-
if (lastLoggedStatus != "NULL") {
2276-
logger.info("waitingMTMVTaskFinishedNotNeedSuccess task row is empty")
2277-
lastLoggedStatus = "NULL"
2278-
}
2279-
status = "NULL"
2280-
} else {
2281-
taskRow = result[0]
2282-
status = taskRow.get(1).toString()
2283-
if (lastLoggedStatus != status) {
2284-
logger.info("The state of ${showTasks} is ${status}, taskId is ${taskRow.get(0)}")
2285-
lastLoggedStatus = status
2286-
}
2287-
}
2288-
if (status == 'PENDING' || status == 'RUNNING' || status == 'NULL') {
2289-
Thread.sleep(500);
2290-
}
2291-
} while (timeoutTimestamp > System.currentTimeMillis() && (status == 'PENDING' || status == 'RUNNING' || status == 'NULL'))
2262+
List<Object> taskRow = waitMTMVTaskTerminal(showTasks, "waitingMTMVTaskFinishedNotNeedSuccess")
2263+
String status = taskRow == null ? "NULL" : taskRow.get(1).toString()
22922264
if (status != "SUCCESS") {
22932265
logger.info("status is not success")
22942266
}

‎regression-test/framework/src/test/groovy/org/apache/doris/regression/suite/SuiteMTMVTaskWaitTest.groovy‎

Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
// Licensed to the Apache Software Foundation (ASF) under one
2+
// or more contributor license agreements. See the NOTICE file
3+
// distributed with this work for additional information
4+
// regarding copyright ownership. The ASF licenses this file
5+
// to you under the Apache License, Version 2.0 (the
6+
// "License"); you may not use this file except in compliance
7+
// with the License. You may obtain a copy of the License at
8+
//
9+
// http://www.apache.org/licenses/LICENSE-2.0
10+
//
11+
// Unless required by applicable law or agreed to in writing,
12+
// software distributed under the License is distributed on an
13+
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
14+
// KIND, either express or implied. See the License for the
15+
// specific language governing permissions and limitations
16+
// under the License.
17+
18+
package org.apache.doris.regression.suite
19+
20+
import org.junit.jupiter.api.Test
21+
22+
import static org.junit.jupiter.api.Assertions.assertEquals
23+
24+
class SuiteMTMVTaskWaitTest {
25+
private static final String SHOW_TASKS = "select TaskId, Status from tasks('type'='mv')"
26+
private static final Closure<String> IGNORE_LOG = { String message -> }
27+
28+
private static List<Object> task(String taskId, String status) {
29+
return [taskId, status] as List<Object>
30+
}
31+
32+
private static Closure<List<List<Object>>> polls(List<List<List<Object>>> script) {
33+
Deque<List<List<Object>>> queue = new ArrayDeque<>(script)
34+
return { ->
35+
if (queue.isEmpty()) {
36+
throw new IllegalStateException("poll script exhausted, the wait did not converge")
37+
}
38+
return queue.poll()
39+
}
40+
}
41+
42+
private static String waitFor(Set<String> finished, List<List<List<Object>>> script) {
43+
return Suite.pollMTMVTaskTerminal(SHOW_TASKS, "test", finished, polls(script), IGNORE_LOG, 1, 5)
44+
.get(0).toString()
45+
}
46+
47+
@Test
48+
void followsATaskThatWasSeenRunning() {
49+
assertEquals("X", waitFor(new HashSet<>(), [[task("X", "RUNNING")], [task("X", "SUCCESS")]]))
50+
}
51+
52+
@Test
53+
void confirmsATerminalRowThatWasNeverSeenRunning() {
54+
assertEquals("X", waitFor(new HashSet<>(), [[task("X", "SUCCESS")], [task("X", "SUCCESS")]]))
55+
}
56+
57+
@Test
58+
void ignoresThePreviousTaskWhileTheNewTaskIsMissing() {
59+
Set<String> finished = new HashSet<>()
60+
assertEquals("P", waitFor(finished, [[task("P", "RUNNING")], [task("P", "SUCCESS")]]))
61+
62+
// X runs and finishes; while it moves from the running list to the MV history the
63+
// newest row of the MV is P again, and P must not be returned a second time.
64+
assertEquals("X", waitFor(finished, [[task("X", "RUNNING")], [task("P", "SUCCESS")],
65+
[task("X", "SUCCESS")]]))
66+
}
67+
68+
@Test
69+
void fallsBackToTheSameTaskWhenNothingNewIsSubmitted() {
70+
Set<String> finished = new HashSet<>()
71+
assertEquals("P", waitFor(finished, [[task("P", "RUNNING")], [task("P", "SUCCESS")]]))
72+
73+
// Nothing new was submitted: the wait must return P instead of polling to the timeout.
74+
assertEquals("P", waitFor(finished, (1..100).collect { [task("P", "SUCCESS")] }))
75+
}
76+
}

0 commit comments

Comments
 (0)