github-actions[bot] commented on code in PR #65153:
URL: https://github.com/apache/doris/pull/65153#discussion_r3568137704
##########
fe/fe-core/src/main/java/org/apache/doris/cloud/CloudWarmUpJob.java:
##########
@@ -1123,6 +1128,13 @@ private void runRunningJob() throws Exception {
if (this.isPeriodic()) {
// wait for next schedule
this.jobState = JobState.PENDING;
+ long nextScheduleTimeMs = finishedTimeMs +
syncInterval * 1000;
Review Comment:
This log does not match the scheduler condition it is trying to explain.
`shouldWait()` delays periodic jobs until `startTimeMs + syncInterval`, but
this diagnostic reports `finishedTimeMs + syncInterval` and computes
`triggerImmediately` from that later time. For any run that lasts longer than
the interval, the scheduler can run again immediately while the log says the
next schedule is still in the future. Please compute the logged next
schedule/trigger flag from the same timestamp basis as `shouldWait()` (or
change the scheduler if the intended semantics are delay-after-finish).
##########
regression-test/framework/src/main/groovy/org/apache/doris/regression/util/WarmupMetricsUtils.groovy:
##########
@@ -265,4 +265,259 @@ class WarmupMetricsUtils {
logger.warn("waitForMetricsStable timed out after ${timeoutMs}ms")
return getWarmupMetrics(sqlRunner, srcCluster, dstCluster)
}
+
+ static List<Map> showWarmupJobs(Closure sqlRunner) {
+ def rows = sqlRunner("SHOW WARM UP JOB")
+ return rows.collect { row -> normalizeWarmupJobRow(row) }
+ }
+
+ static Map showWarmupJob(Closure sqlRunner, Object jobId) {
+ def rows = sqlRunner("SHOW WARM UP JOB WHERE ID = ${jobId}")
+ if (rows == null || rows.isEmpty()) {
+ throw new RuntimeException("warmup job ${jobId} not found")
+ }
+ return normalizeWarmupJobRow(rows[0])
+ }
+
+ static Map normalizeWarmupJobRow(Object row) {
+ List values = row as List
+ return [
+ jobId : values[0]?.toString(),
+ srcCluster : values[1]?.toString(),
+ dstCluster : values[2]?.toString(),
+ status : values[3]?.toString(),
+ type : values[4]?.toString(),
+ syncMode : values[5]?.toString(),
+ createTime : values[6]?.toString(),
+ startTime : values[7]?.toString(),
+ finishedTime : values[8]?.toString(),
+ allBatch : values[9]?.toString(),
+ finishedBatch : values[10]?.toString(),
+ errMsg : values[11]?.toString(),
+ tableOrPattern : values.size() > 12 ? values[12]?.toString() :
"",
+ tableFilter : values.size() > 13 ? values[13]?.toString() :
"",
+ matchedTables : values.size() > 14 ? values[14]?.toString() :
"",
+ syncStats : values.size() > 15 ? values[15]?.toString() :
"",
+ raw : values,
+ ]
+ }
+
+ static boolean isNormalWarmupJob(Map row) {
+ String syncMode = row.syncMode ?: ""
+ return syncMode.startsWith("ONCE") || syncMode.startsWith("PERIODIC")
+ }
+
+ static boolean isEventDrivenWarmupJob(Map row) {
+ String syncMode = row.syncMode ?: ""
+ return syncMode.startsWith("EVENT_DRIVEN")
+ }
+
+ static List<Map> showWarmupJobsByDst(Closure sqlRunner, String dstCluster)
{
+ return showWarmupJobs(sqlRunner).findAll { it.dstCluster == dstCluster
}
+ }
+
+ static long countRunningNormalWarmupByDst(Closure sqlRunner, String
dstCluster) {
+ return showWarmupJobsByDst(sqlRunner, dstCluster).count {
+ isNormalWarmupJob(it) && it.status == "RUNNING"
+ } as long
+ }
+
+ static List<Map> snapshotWarmupJobs(Closure sqlRunner) {
+ return showWarmupJobs(sqlRunner).collect { new LinkedHashMap<>(it) }
+ }
+
+ static Map<String, Map> snapshotWarmupJobsById(Closure sqlRunner) {
+ Map<String, Map> result = [:]
+ snapshotWarmupJobs(sqlRunner).each { result[it.jobId] = it }
+ return result
+ }
+
+ static void waitForOnlyOneRunningNormalWarmup(Closure sqlRunner, String
dstCluster,
+ long timeoutMs = 30000) {
+ long deadline = System.currentTimeMillis() + timeoutMs
+ long lastCount = -1
+ while (System.currentTimeMillis() < deadline) {
+ lastCount = countRunningNormalWarmupByDst(sqlRunner, dstCluster)
+ if (lastCount <= 1) {
Review Comment:
This can return before the scheduler has exercised the lock path. Right
after the tests create the normal warm-up jobs they can all still be `PENDING`,
because `tryRegisterRunningJob()` is only called from `runPendingJob()` when
the scheduler runs; that zero-running snapshot satisfies `lastCount <= 1`. The
new queue tests can therefore pass without proving that one normal job is
running while another same-destination normal job is kept out, especially after
FE restart. Please wait until at least one expected normal job has actually
reached `RUNNING` or attempted to run, then assert over a scheduler cycle that
no second normal job becomes `RUNNING`.
##########
regression-test/framework/src/main/groovy/org/apache/doris/regression/util/WarmupMetricsUtils.groovy:
##########
@@ -265,4 +265,259 @@ class WarmupMetricsUtils {
logger.warn("waitForMetricsStable timed out after ${timeoutMs}ms")
return getWarmupMetrics(sqlRunner, srcCluster, dstCluster)
}
+
+ static List<Map> showWarmupJobs(Closure sqlRunner) {
+ def rows = sqlRunner("SHOW WARM UP JOB")
+ return rows.collect { row -> normalizeWarmupJobRow(row) }
+ }
+
+ static Map showWarmupJob(Closure sqlRunner, Object jobId) {
+ def rows = sqlRunner("SHOW WARM UP JOB WHERE ID = ${jobId}")
+ if (rows == null || rows.isEmpty()) {
+ throw new RuntimeException("warmup job ${jobId} not found")
+ }
+ return normalizeWarmupJobRow(rows[0])
+ }
+
+ static Map normalizeWarmupJobRow(Object row) {
+ List values = row as List
+ return [
+ jobId : values[0]?.toString(),
+ srcCluster : values[1]?.toString(),
+ dstCluster : values[2]?.toString(),
+ status : values[3]?.toString(),
+ type : values[4]?.toString(),
+ syncMode : values[5]?.toString(),
+ createTime : values[6]?.toString(),
+ startTime : values[7]?.toString(),
+ finishedTime : values[8]?.toString(),
+ allBatch : values[9]?.toString(),
+ finishedBatch : values[10]?.toString(),
+ errMsg : values[11]?.toString(),
+ tableOrPattern : values.size() > 12 ? values[12]?.toString() :
"",
+ tableFilter : values.size() > 13 ? values[13]?.toString() :
"",
+ matchedTables : values.size() > 14 ? values[14]?.toString() :
"",
+ syncStats : values.size() > 15 ? values[15]?.toString() :
"",
+ raw : values,
+ ]
+ }
+
+ static boolean isNormalWarmupJob(Map row) {
+ String syncMode = row.syncMode ?: ""
+ return syncMode.startsWith("ONCE") || syncMode.startsWith("PERIODIC")
+ }
+
+ static boolean isEventDrivenWarmupJob(Map row) {
+ String syncMode = row.syncMode ?: ""
+ return syncMode.startsWith("EVENT_DRIVEN")
+ }
+
+ static List<Map> showWarmupJobsByDst(Closure sqlRunner, String dstCluster)
{
+ return showWarmupJobs(sqlRunner).findAll { it.dstCluster == dstCluster
}
+ }
+
+ static long countRunningNormalWarmupByDst(Closure sqlRunner, String
dstCluster) {
+ return showWarmupJobsByDst(sqlRunner, dstCluster).count {
+ isNormalWarmupJob(it) && it.status == "RUNNING"
+ } as long
+ }
+
+ static List<Map> snapshotWarmupJobs(Closure sqlRunner) {
+ return showWarmupJobs(sqlRunner).collect { new LinkedHashMap<>(it) }
+ }
+
+ static Map<String, Map> snapshotWarmupJobsById(Closure sqlRunner) {
+ Map<String, Map> result = [:]
+ snapshotWarmupJobs(sqlRunner).each { result[it.jobId] = it }
+ return result
+ }
+
+ static void waitForOnlyOneRunningNormalWarmup(Closure sqlRunner, String
dstCluster,
+ long timeoutMs = 30000) {
+ long deadline = System.currentTimeMillis() + timeoutMs
+ long lastCount = -1
+ while (System.currentTimeMillis() < deadline) {
+ lastCount = countRunningNormalWarmupByDst(sqlRunner, dstCluster)
+ if (lastCount <= 1) {
+ return
+ }
+ Thread.sleep(1000)
+ }
+ throw new RuntimeException("expected at most one running normal warmup
on ${dstCluster}, last=${lastCount}")
+ }
+
+ static Map waitForWarmupJobsRecovered(Closure sqlRunner, Map<String, Map>
beforeSnapshot,
+ Closure<Boolean> predicate, long
timeoutMs = 60000) {
+ long deadline = System.currentTimeMillis() + timeoutMs
+ Map<String, Map> current = [:]
+ while (System.currentTimeMillis() < deadline) {
+ current = snapshotWarmupJobsById(sqlRunner)
+ if (predicate(beforeSnapshot, current)) {
+ return current
+ }
+ Thread.sleep(1000)
+ }
+ return current
+ }
+
+ static Map waitForPeriodicCycles(Closure sqlRunner, Object jobId, int
minTransitions,
+ long timeoutMs = 120000) {
+ long deadline = System.currentTimeMillis() + timeoutMs
+ List<String> states = []
+ String lastState = null
+ while (System.currentTimeMillis() < deadline) {
+ def row = showWarmupJob(sqlRunner, jobId)
+ if (row.status != lastState) {
+ states << row.status
+ lastState = row.status
+ }
+ int transitions = 0
+ for (int i = 1; i < states.size(); i++) {
+ if (states[i - 1] == "RUNNING" && (states[i] == "PENDING" ||
states[i] == "WAITING")) {
+ transitions++
+ }
+ }
+ if (transitions >= minTransitions) {
+ return [states: states, row: row]
+ }
+ Thread.sleep(1000)
+ }
+ return [states: states, row: showWarmupJob(sqlRunner, jobId)]
Review Comment:
This helper can return a timeout result that the new scheduler test still
treats as success. The loop only observes the requested cycle when `transitions
>= minTransitions`, but after the deadline it returns the states anyway; a job
that goes `[PENDING, RUNNING]` and never reschedules before timeout satisfies
the caller's `contains("RUNNING")` and `contains(PENDING/WAITING)` assertions
without ever proving a RUNNING-to-PENDING/WAITING periodic cycle. Please make
the helper fail on timeout or return/assert the transition count so the test
only passes after the requested cycle was actually observed.
##########
regression-test/framework/src/main/groovy/org/apache/doris/regression/util/WarmupMetricsUtils.groovy:
##########
@@ -265,4 +265,259 @@ class WarmupMetricsUtils {
logger.warn("waitForMetricsStable timed out after ${timeoutMs}ms")
return getWarmupMetrics(sqlRunner, srcCluster, dstCluster)
}
+
+ static List<Map> showWarmupJobs(Closure sqlRunner) {
+ def rows = sqlRunner("SHOW WARM UP JOB")
+ return rows.collect { row -> normalizeWarmupJobRow(row) }
+ }
+
+ static Map showWarmupJob(Closure sqlRunner, Object jobId) {
+ def rows = sqlRunner("SHOW WARM UP JOB WHERE ID = ${jobId}")
+ if (rows == null || rows.isEmpty()) {
+ throw new RuntimeException("warmup job ${jobId} not found")
+ }
+ return normalizeWarmupJobRow(rows[0])
+ }
+
+ static Map normalizeWarmupJobRow(Object row) {
+ List values = row as List
+ return [
+ jobId : values[0]?.toString(),
+ srcCluster : values[1]?.toString(),
+ dstCluster : values[2]?.toString(),
+ status : values[3]?.toString(),
+ type : values[4]?.toString(),
+ syncMode : values[5]?.toString(),
+ createTime : values[6]?.toString(),
+ startTime : values[7]?.toString(),
+ finishedTime : values[8]?.toString(),
+ allBatch : values[9]?.toString(),
+ finishedBatch : values[10]?.toString(),
+ errMsg : values[11]?.toString(),
+ tableOrPattern : values.size() > 12 ? values[12]?.toString() :
"",
+ tableFilter : values.size() > 13 ? values[13]?.toString() :
"",
+ matchedTables : values.size() > 14 ? values[14]?.toString() :
"",
+ syncStats : values.size() > 15 ? values[15]?.toString() :
"",
+ raw : values,
+ ]
+ }
+
+ static boolean isNormalWarmupJob(Map row) {
+ String syncMode = row.syncMode ?: ""
+ return syncMode.startsWith("ONCE") || syncMode.startsWith("PERIODIC")
+ }
+
+ static boolean isEventDrivenWarmupJob(Map row) {
+ String syncMode = row.syncMode ?: ""
+ return syncMode.startsWith("EVENT_DRIVEN")
+ }
+
+ static List<Map> showWarmupJobsByDst(Closure sqlRunner, String dstCluster)
{
+ return showWarmupJobs(sqlRunner).findAll { it.dstCluster == dstCluster
}
+ }
+
+ static long countRunningNormalWarmupByDst(Closure sqlRunner, String
dstCluster) {
+ return showWarmupJobsByDst(sqlRunner, dstCluster).count {
+ isNormalWarmupJob(it) && it.status == "RUNNING"
+ } as long
+ }
+
+ static List<Map> snapshotWarmupJobs(Closure sqlRunner) {
+ return showWarmupJobs(sqlRunner).collect { new LinkedHashMap<>(it) }
+ }
+
+ static Map<String, Map> snapshotWarmupJobsById(Closure sqlRunner) {
+ Map<String, Map> result = [:]
+ snapshotWarmupJobs(sqlRunner).each { result[it.jobId] = it }
+ return result
+ }
+
+ static void waitForOnlyOneRunningNormalWarmup(Closure sqlRunner, String
dstCluster,
+ long timeoutMs = 30000) {
+ long deadline = System.currentTimeMillis() + timeoutMs
+ long lastCount = -1
+ while (System.currentTimeMillis() < deadline) {
+ lastCount = countRunningNormalWarmupByDst(sqlRunner, dstCluster)
+ if (lastCount <= 1) {
+ return
+ }
+ Thread.sleep(1000)
+ }
+ throw new RuntimeException("expected at most one running normal warmup
on ${dstCluster}, last=${lastCount}")
+ }
+
+ static Map waitForWarmupJobsRecovered(Closure sqlRunner, Map<String, Map>
beforeSnapshot,
+ Closure<Boolean> predicate, long
timeoutMs = 60000) {
+ long deadline = System.currentTimeMillis() + timeoutMs
+ Map<String, Map> current = [:]
+ while (System.currentTimeMillis() < deadline) {
+ current = snapshotWarmupJobsById(sqlRunner)
+ if (predicate(beforeSnapshot, current)) {
+ return current
+ }
+ Thread.sleep(1000)
+ }
+ return current
+ }
+
+ static Map waitForPeriodicCycles(Closure sqlRunner, Object jobId, int
minTransitions,
+ long timeoutMs = 120000) {
+ long deadline = System.currentTimeMillis() + timeoutMs
+ List<String> states = []
+ String lastState = null
+ while (System.currentTimeMillis() < deadline) {
+ def row = showWarmupJob(sqlRunner, jobId)
+ if (row.status != lastState) {
+ states << row.status
+ lastState = row.status
+ }
+ int transitions = 0
+ for (int i = 1; i < states.size(); i++) {
+ if (states[i - 1] == "RUNNING" && (states[i] == "PENDING" ||
states[i] == "WAITING")) {
+ transitions++
+ }
+ }
+ if (transitions >= minTransitions) {
+ return [states: states, row: row]
+ }
+ Thread.sleep(1000)
+ }
+ return [states: states, row: showWarmupJob(sqlRunner, jobId)]
+ }
+
+ static List<Map> sampleJobTimeline(Closure sqlRunner, Object jobId, long
durationMs,
+ long intervalMs = 1000) {
+ long deadline = System.currentTimeMillis() + durationMs
+ List<Map> samples = []
+ while (System.currentTimeMillis() < deadline) {
+ def row = showWarmupJob(sqlRunner, jobId)
+ samples << [
+ ts : System.currentTimeMillis(),
+ status : row.status,
+ start : row.startTime,
+ finished: row.finishedTime,
+ syncMode: row.syncMode,
+ ]
+ Thread.sleep(intervalMs)
+ }
+ return samples
+ }
+
+ static List<Map> collectWarmupJobsByCluster(Closure sqlRunner, String
clusterName) {
+ return showWarmupJobs(sqlRunner).findAll {
+ it.srcCluster == clusterName || it.dstCluster == clusterName
+ }
+ }
+
+ static void assertAffectedJobsCancelled(Closure sqlRunner,
Collection<String> jobIds,
+ long timeoutMs = 60000) {
+ long deadline = System.currentTimeMillis() + timeoutMs
+ Map<String, String> lastStates = [:]
+ while (System.currentTimeMillis() < deadline) {
+ boolean allCancelled = true
+ for (String jobId in jobIds) {
+ def row = showWarmupJob(sqlRunner, jobId)
+ lastStates[jobId] = row.status
+ if (row.status != "CANCELLED") {
+ allCancelled = false
+ }
+ }
+ if (allCancelled) {
+ return
+ }
+ Thread.sleep(1000)
+ }
+ throw new RuntimeException("expected jobs cancelled,
lastStates=${lastStates}")
+ }
+
+ static void assertUnrelatedJobsUnaffected(Closure sqlRunner,
Collection<String> jobIds,
+ Collection<String>
allowedStatuses = ["RUNNING", "PENDING", "FINISHED"]) {
+ for (String jobId in jobIds) {
+ def row = showWarmupJob(sqlRunner, jobId)
+ if (!allowedStatuses.contains(row.status)) {
+ throw new RuntimeException("unrelated warmup job ${jobId}
entered unexpected status ${row.status}")
+ }
+ }
+ }
+
+ static String expectCreateConflict(Closure sqlRunner, Closure createJob) {
+ try {
+ createJob.call()
+ } catch (Throwable t) {
+ return t.message ?: t.toString()
+ }
+ throw new RuntimeException("expected warmup create conflict, but
statement succeeded")
+ }
+
+ static void assertConflictMessage(String message, Collection<String>
expectedKeywords) {
+ expectedKeywords.each { keyword ->
+ if (!(message?.toLowerCase()?.contains(keyword.toLowerCase()))) {
+ throw new RuntimeException("expected conflict message to
contain '${keyword}', actual=${message}")
+ }
+ }
+ }
+
+ static List<String> getVcgWarmupJobIds(Closure sqlRunner, String
srcCluster, String dstCluster) {
+ return showWarmupJobs(sqlRunner).findAll {
+ it.srcCluster == srcCluster && it.dstCluster == dstCluster
Review Comment:
These helpers should not treat every source/destination match as a recreated
active VCG warm-up job. `SHOW WARM UP JOB` is backed by `cloudWarmUpJobs`;
cancelled/finished jobs are removed from `runnableCloudWarmUpJobs` but stay
visible until the history TTL. Because this filter ignores status, type, and
sync mode, `waitForVcgWarmupRecreated()` can return historical non-old jobs and
the VCG tests can pass without active `CLUSTER` periodic plus event-driven jobs
being rebuilt. Please filter or validate active statuses and the expected
`CLUSTER` periodic/event-driven modes before treating recreation as successful.
##########
regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_normal_queue_semantics.groovy:
##########
@@ -0,0 +1,140 @@
+// 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 org.apache.doris.regression.util.WarmupMetricsUtils
+
+suite('test_warm_up_normal_queue_semantics', 'docker') {
+ def options = new ClusterOptions()
+ options.feNum = 3
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'fetch_cluster_cache_hotspot_interval_ms=1000',
+ ]
+ options.beConfigs += [
+ 'file_cache_enter_disk_resource_limit_mode_percent=99',
+ 'enable_evict_file_cache_in_advance=false',
+ 'file_cache_background_monitor_interval_ms=1000',
+ ]
+ options.cloudMode = true
+
+ docker(options) {
+ Closure sqlRunner = { String q -> sql(q) }
+
+ def srcCluster = "warmup_queue_src"
+ def dstCluster = "warmup_queue_dst"
+ def auxCluster = "warmup_queue_aux"
+ def dbName = "test_warmup_queue_semantics_db"
+ def tableName = "queue_tbl"
+
+ cluster.addBackend(1, srcCluster)
+ cluster.addBackend(1, dstCluster)
+ cluster.addBackend(1, auxCluster)
+
+ def restartMasterFe = {
+ def oldMasterFe = cluster.getMasterFe()
+ cluster.restartFrontends(oldMasterFe.index)
+ boolean restarted = false
+ for (int i = 0; i < 30; i++) {
+ if (cluster.getFeByIndex(oldMasterFe.index).alive) {
+ restarted = true
+ break
+ }
+ sleep(1000)
+ }
+ assertTrue(restarted)
+ context.reconnectFe()
+ }
+
+ sql """use @${srcCluster}"""
+ sql """CREATE DATABASE IF NOT EXISTS ${dbName}"""
+ sql """use ${dbName}"""
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ sql """CREATE TABLE ${tableName} (
+ id INT,
+ name STRING
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("file_cache_ttl_seconds" = "3600")"""
+ for (int i = 0; i < 20; i++) {
+ sql """INSERT INTO ${tableName} VALUES (${i}, 'name_${i}')"""
+ }
+
+ def periodicJobId = sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ PROPERTIES (
+ "sync_mode" = "periodic",
+ "sync_interval_sec" = "1"
+ )
+ """)[0][0]
+ def onceTableJobId = sql("""WARM UP CLUSTER ${dstCluster} WITH TABLE
${tableName}""")[0][0]
+ def onceClusterJobId = sql("""WARM UP CLUSTER ${dstCluster} WITH
CLUSTER ${auxCluster}""")[0][0]
+ def eventJobId = sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ ON TABLES (INCLUDE '${dbName}.${tableName}')
+ PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load")
+ """)[0][0]
+
+ def queueRows = WarmupMetricsUtils.showWarmupJobsByDst(sqlRunner,
dstCluster)
+ assert queueRows.find { it.jobId == periodicJobId.toString() } != null
+ assert queueRows.find { it.jobId == onceTableJobId.toString() } != null
+ assert queueRows.find { it.jobId == onceClusterJobId.toString() } !=
null
+ assert queueRows.find { it.jobId == eventJobId.toString() } != null
+
+ WarmupMetricsUtils.waitForOnlyOneRunningNormalWarmup(sqlRunner,
dstCluster, 30000)
+ long runningNormal =
WarmupMetricsUtils.countRunningNormalWarmupByDst(sqlRunner, dstCluster)
+ assertTrue(runningNormal <= 1,
+ "same dst cluster should have at most one running normal
warmup, got ${runningNormal}")
+
+ def eventRow = WarmupMetricsUtils.showWarmupJob(sqlRunner, eventJobId)
+ assertTrue(WarmupMetricsUtils.isEventDrivenWarmupJob(eventRow))
+ assertTrue(eventRow.status in ["RUNNING", "PENDING", "WAITING"])
+
+ Map<String, Map> beforeRestartSnapshot =
WarmupMetricsUtils.snapshotWarmupJobsById(sqlRunner)
+ restartMasterFe()
+ sql """use @${srcCluster}"""
+ sql """use ${dbName}"""
+ Map<String, Map> afterRestartSnapshot =
WarmupMetricsUtils.waitForWarmupJobsRecovered(
+ sqlRunner, beforeRestartSnapshot, { before, current ->
+ [periodicJobId, onceTableJobId, onceClusterJobId,
eventJobId].every { current.containsKey(it.toString()) }
+ }, 60000)
+ assertTrue(afterRestartSnapshot.keySet().containsAll(
+ [periodicJobId, onceTableJobId, onceClusterJobId,
eventJobId].collect { it.toString() }))
+
+ WarmupMetricsUtils.waitForOnlyOneRunningNormalWarmup(sqlRunner,
dstCluster, 30000)
+ runningNormal =
WarmupMetricsUtils.countRunningNormalWarmupByDst(sqlRunner, dstCluster)
+ assertTrue(runningNormal <= 1,
+ "after FE restart same dst cluster should still have at most
one running normal warmup, got ${runningNormal}")
+
+ def beforeMetrics = WarmupMetricsUtils.getWarmupMetrics(sqlRunner,
srcCluster, dstCluster)
+ for (int i = 20; i < 24; i++) {
+ sql """INSERT INTO ${tableName} VALUES (${i},
'after_restart_${i}')"""
+ }
+ def afterMetrics = WarmupMetricsUtils.waitForWarmupFinish(sqlRunner,
srcCluster, dstCluster,
+ beforeMetrics.finished + 1, 120000)
+ assertTrue(afterMetrics.requested > beforeMetrics.requested,
Review Comment:
This can still pass when the destination never finishes the warm-up.
`waitForWarmupFinish()` only logs and returns a snapshot on timeout, and
`requested` comes from the source cluster, so a load event can increment
`requested` while the target side never reaches `finished >= before.finished +
1`. The same pattern appears in the new overlap and VCG target-restart tests.
Please either make the helper fail on timeout or assert the returned metrics
actually satisfy the destination completion condition you waited for, for
example `afterMetrics.finished >= beforeMetrics.finished + 1` and
`afterMetrics.finished + afterMetrics.failed >= afterMetrics.submitted`.
##########
docker/runtime/doris-compose/utils.py:
##########
@@ -251,17 +269,17 @@ def _command_available(cmd):
)
return cmd
- # Prefer Compose v2.
- if _command_available(["docker", "compose", "version"]):
+ # Prefer Compose v2 when it can talk to the daemon.
+ if _compose_v2_usable(["docker", "compose"]):
return ["docker", "compose"]
- # Fallback for old machines.
+ # Fallback for old machines or environments with a newer standalone client.
if _command_available(["docker-compose", "version"]):
Review Comment:
Since the v2 path only accepts `docker compose` after `compose ls` talks to
the daemon, this fallback should apply an equivalent daemon-touching check
before returning `docker-compose`. In a mixed install, `docker compose ls` can
fail for the API/daemon mismatch this PR is trying to catch, but
`docker-compose version` can still succeed, so detection returns the standalone
client and the real `up`/`stop` command fails later. Please either verify the
standalone client with a daemon command too, or narrow the function
contract/error text so it does not claim the returned client is usable with the
current daemon.
--
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]
