This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new 8e2a55e3f31 branch-4.1: [test](regression) Add warmup mixed-mode
coverage and debug observability #65153 (#65568)
8e2a55e3f31 is described below
commit 8e2a55e3f3139c23635835134986ad9ce68219f1
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Jul 15 12:08:56 2026 +0800
branch-4.1: [test](regression) Add warmup mixed-mode coverage and debug
observability #65153 (#65568)
Cherry-picked from #65153
Co-authored-by: chunping <[email protected]>
---
docker/runtime/doris-compose/utils.py | 36 ++-
.../apache/doris/cloud/CacheHotspotManager.java | 40 ++-
.../org/apache/doris/cloud/CloudWarmUpJob.java | 15 +-
.../cloud/catalog/CloudInstanceStatusChecker.java | 19 +-
.../cloud/CacheHotspotManagerTableFilterTest.java | 61 ++++
.../doris/cloud/cache/CacheHotspotManagerTest.java | 46 +++
.../catalog/CloudInstanceStatusCheckerTest.java | 5 +-
.../regression/util/WarmupMetricsUtils.groovy | 310 ++++++++++++++++++++-
...p_cluster_event_conflict_and_diagnostics.groovy | 111 ++++++++
...test_warm_up_cluster_periodic_add_new_be.groovy | 184 +++++++++---
..._up_cluster_periodic_scheduler_semantics.groovy | 137 +++++++++
...arm_up_mixed_cluster_change_and_failover.groovy | 117 ++++++++
.../test_warm_up_normal_queue_semantics.groovy | 146 ++++++++++
...arm_up_event_on_tables_overlap_semantics.groovy | 125 +++++++++
...t_vcg_warmup_allows_manual_periodic_jobs.groovy | 103 +++++++
...est_vcg_warmup_failover_cancels_old_jobs.groovy | 125 +++++++++
...test_vcg_warmup_precedence_and_reconcile.groovy | 153 ++++++++++
.../test_vcg_warmup_restart_master_fe.groovy | 140 ++++++++++
.../test_vcg_warmup_restart_target_be.groovy | 130 +++++++++
...vcg_warmup_shared_compute_group_conflict.groovy | 99 +++++++
...g_warmup_switch_and_drop_cancels_history.groovy | 130 +++++++++
...g_warmup_waits_for_existing_cluster_jobs.groovy | 107 +++++++
..._warmup_with_manual_once_queue_semantics.groovy | 111 ++++++++
...mup_with_manual_periodic_queue_semantics.groovy | 110 ++++++++
24 files changed, 2503 insertions(+), 57 deletions(-)
diff --git a/docker/runtime/doris-compose/utils.py
b/docker/runtime/doris-compose/utils.py
index 99912f0c003..0f41728d3f2 100644
--- a/docker/runtime/doris-compose/utils.py
+++ b/docker/runtime/doris-compose/utils.py
@@ -223,8 +223,8 @@ def detect_docker_compose_cmd():
Priority:
1. DORIS_DOCKER_COMPOSE_CMD env override
e.g. "docker compose" or "docker-compose" or "/path/to/docker-compose"
- 2. docker compose v2
- 3. docker-compose v1 / wrapper
+ 2. the first compose client that can talk to the current daemon
+ (prefer docker compose v2, then docker-compose v1 / wrapper)
"""
def _command_available(cmd):
try:
@@ -238,26 +238,44 @@ def detect_docker_compose_cmd():
except (subprocess.CalledProcessError, FileNotFoundError):
return False
+ def _compose_usable(cmd):
+ if not _command_available(cmd + ["version"]):
+ return False
+ try:
+ # `ls` talks to the daemon and catches API-version mismatches that
+ # `version` alone misses on mixed docker/compose installations.
+ subprocess.run(
+ cmd + ["ls"],
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ check=True,
+ )
+ return True
+ except subprocess.CalledProcessError:
+ return False
+ except FileNotFoundError:
+ return False
+
override = os.environ.get("DORIS_DOCKER_COMPOSE_CMD")
if override:
cmd = shlex.split(override)
- if not _command_available(cmd + ["version"]):
+ if not _compose_usable(cmd):
raise RuntimeError(
"DORIS_DOCKER_COMPOSE_CMD is set but not usable:
{}".format(override)
)
return cmd
- # Prefer Compose v2.
- if _command_available(["docker", "compose", "version"]):
+ # Prefer Compose v2 when it can talk to the daemon.
+ if _compose_usable(["docker", "compose"]):
return ["docker", "compose"]
- # Fallback for old machines.
- if _command_available(["docker-compose", "version"]):
+ # Fallback for old machines or environments with a newer standalone client.
+ if _compose_usable(["docker-compose"]):
return ["docker-compose"]
raise RuntimeError(
- "Neither 'docker compose' nor 'docker-compose' is available. "
- "Please install Docker Compose v2 plugin or legacy docker-compose."
+ "Neither 'docker compose' nor 'docker-compose' is usable with the
current "
+ "Docker daemon. Please install a compatible Docker Compose client."
)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java
index 9155745a2e1..e35fa2c6b22 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/CacheHotspotManager.java
@@ -404,6 +404,12 @@ public class CacheHotspotManager extends MasterDaemon {
continue;
}
if (isTableLevelLoadEventWarmUpJob(newJob) !=
isTableLevelLoadEventWarmUpJob(existingJob)) {
+ LOG.info("warmup-conflict detect newJobId={} existingJobId={}
srcCluster={} dstCluster={} "
+ + "newJobType={} existingJobType={}
newTableFilter={} existingTableFilter={}",
+ newJob.getJobId(), existingJob.getJobId(),
+ newJob.getSrcClusterName(), newJob.getDstClusterName(),
+ newJob.getJobType(), existingJob.getJobType(),
+ newJob.getTableFilterExpr(),
existingJob.getTableFilterExpr());
throw buildLoadEventWarmUpConflictException(newJob,
existingJob);
}
}
@@ -507,6 +513,10 @@ public class CacheHotspotManager extends MasterDaemon {
public boolean tryRegisterRunningJob(CloudWarmUpJob job) {
if (job.isEventDriven()) {
// Event-driven jobs do not require registration, always allow
+ LOG.debug("warmup-lock register-skip jobId={} srcCluster={}
dstCluster={} syncMode={} jobType={} "
+ + "reason=event-driven",
+ job.getJobId(), job.getSrcClusterName(),
job.getDstClusterName(),
+ job.getSyncMode(), job.getJobType());
return true;
}
@@ -516,9 +526,16 @@ public class CacheHotspotManager extends MasterDaemon {
// Try to register the job atomically if absent
Long existingJobId = clusterToRunningJobId.putIfAbsent(clusterName,
jobId);
boolean success = (existingJobId == null) || (existingJobId == jobId);
- if (!success) {
- LOG.info("Job {} skipped: waiting for job {} to finish on
destination cluster {}",
- jobId, existingJobId, clusterName);
+ if (success) {
+ LOG.debug("warmup-lock register jobId={} srcCluster={}
dstCluster={} syncMode={} jobType={} "
+ + "existingJobId={} registerResult={}",
+ jobId, job.getSrcClusterName(), clusterName,
job.getSyncMode(), job.getJobType(),
+ existingJobId, "success");
+ } else {
+ LOG.info("warmup-lock register jobId={} srcCluster={}
dstCluster={} syncMode={} jobType={} "
+ + "existingJobId={} registerResult={}",
+ jobId, job.getSrcClusterName(), clusterName,
job.getSyncMode(), job.getJobType(),
+ existingJobId, "blocked");
}
return success;
}
@@ -539,19 +556,32 @@ public class CacheHotspotManager extends MasterDaemon {
private boolean deregisterRunningJob(CloudWarmUpJob job) {
if (job.isEventDriven()) {
// Event-driven jobs are not registered, so nothing to deregister
+ LOG.debug("warmup-lock deregister-skip jobId={} srcCluster={}
dstCluster={} syncMode={} jobType={} "
+ + "reason=event-driven",
+ job.getJobId(), job.getSrcClusterName(),
job.getDstClusterName(),
+ job.getSyncMode(), job.getJobType());
return true;
}
String clusterName = job.getDstClusterName();
long jobId = job.getJobId();
-
- return clusterToRunningJobId.remove(clusterName, jobId);
+ Long currentRegisteredJobId = clusterToRunningJobId.get(clusterName);
+ boolean removed = clusterToRunningJobId.remove(clusterName, jobId);
+ LOG.info("warmup-lock deregister jobId={} srcCluster={} dstCluster={}
syncMode={} jobType={} "
+ + "currentRegisteredJobId={} releaseResult={}
jobFinalState={}",
+ jobId, job.getSrcClusterName(), clusterName,
job.getSyncMode(), job.getJobType(),
+ currentRegisteredJobId, removed, job.getJobState());
+ return removed;
}
public void notifyJobStop(CloudWarmUpJob job) {
if (job.isOnce() || job.isPeriodic()) {
this.deregisterRunningJob(job);
}
+ LOG.info("warmup-lock notify-stop jobId={} srcCluster={} dstCluster={}
syncMode={} jobType={} "
+ + "jobState={} isDone={}",
+ job.getJobId(), job.getSrcClusterName(),
job.getDstClusterName(),
+ job.getSyncMode(), job.getJobType(), job.getJobState(),
job.isDone());
if (!job.isDone()) {
return;
}
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/CloudWarmUpJob.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/CloudWarmUpJob.java
index 1d9ee1a821c..e2f58cc49b0 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/cloud/CloudWarmUpJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/cloud/CloudWarmUpJob.java
@@ -926,6 +926,9 @@ public class CloudWarmUpJob implements Writable {
// make sure only one job runs concurrently for one destination cluster
if (!((CloudEnv)
Env.getCurrentEnv()).getCacheHotspotMgr().tryRegisterRunningJob(this)) {
+ LOG.debug("warmup-lock pending-blocked jobId={} srcCluster={}
dstCluster={} syncMode={} jobType={} "
+ + "state={}",
+ jobId, srcClusterName, dstClusterName, syncMode, jobType,
jobState);
return;
}
@@ -948,7 +951,9 @@ public class CloudWarmUpJob implements Writable {
MetricRepo.increaseClusterWarmUpJobExecCount(dstClusterName);
this.jobState = JobState.RUNNING;
Env.getCurrentEnv().getEditLog().logModifyCloudWarmUpJob(this);
- LOG.info("transfer cloud warm up job {} state to {}", jobId,
this.jobState);
+ LOG.info("warmup-lock state-transition jobId={} srcCluster={}
dstCluster={} syncMode={} jobType={} "
+ + "fromState=PENDING toState={} totalTablets={}",
+ jobId, srcClusterName, dstClusterName, syncMode, jobType,
this.jobState, totalTablets);
}
private List<TJobMeta> buildJobMetas(long beId, long batchId) {
@@ -1123,6 +1128,14 @@ public class CloudWarmUpJob implements Writable {
if (this.isPeriodic()) {
// wait for next schedule
this.jobState = JobState.PENDING;
+ long nowMs = System.currentTimeMillis();
+ long nextScheduleTimeMs = startTimeMs +
syncInterval * 1000;
+ LOG.debug("warmup-periodic reschedule jobId={}
srcCluster={} dstCluster={} "
+ + "syncIntervalSec={}
lastStartTimeMs={} lastFinishTimeMs={} "
+ + "nextScheduleTimeMs={} nowMs={}
triggerImmediately={}",
+ jobId, srcClusterName, dstClusterName,
syncInterval,
+ startTimeMs, finishedTimeMs,
nextScheduleTimeMs, nowMs,
+ nowMs >= nextScheduleTimeMs);
} else {
// release job
this.jobState = JobState.FINISHED;
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java
index 4b3b7cf79b7..73805a950ce 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/cloud/catalog/CloudInstanceStatusChecker.java
@@ -205,6 +205,11 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
private void cancelCacheJobs(ComputeGroup vcgInFe, List<String> jobIds) {
CacheHotspotManager cacheHotspotManager = ((CloudEnv)
Env.getCurrentEnv()).getCacheHotspotMgr();
+ if (!jobIds.isEmpty()) {
+ LOG.info("warmup-vcg cancel-cache-jobs vcgName={}
activeComputeGroup={} standbyComputeGroup={} "
+ + "jobIds={}",
+ vcgInFe.getName(), vcgInFe.getActiveComputeGroup(),
vcgInFe.getStandbyComputeGroup(), jobIds);
+ }
for (String jobId : jobIds) {
try {
if (Env.getCurrentEnv().isMaster()) {
@@ -285,6 +290,10 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
String srcCg = virtualGroupInFe.getActiveComputeGroup();
String dstCg = virtualGroupInFe.getStandbyComputeGroup();
try {
+ LOG.info("warmup-vcg rebuild-start vcgName={} srcCluster={}
dstCluster={} subComputeGroups={} "
+ + "oldJobIds={}",
+ virtualGroupInFe.getName(), srcCg, dstCg,
virtualGroupInFe.getSubComputeGroups(),
+ jobIdsInMs);
cacheHotspotManager.cancelTableLevelLoadEventWarmUpJobsForVirtualComputeGroup(
virtualGroupInFe.getName(), srcCg, dstCg,
virtualGroupInFe.getSubComputeGroups(),
"vcg cancel table-level load-event warm up job before
rebuilding file cache jobs");
@@ -317,11 +326,13 @@ public class CloudInstanceStatusChecker extends
MasterDaemon {
// send jobIds to ms
List<String> newJobIds =
Arrays.asList(Long.toString(jobIdPeriodic), Long.toString(jobIdEvent));
CloudSystemInfoService.updateFileCacheJobIds(virtualGroupInFe,
newJobIds);
- LOG.info("virtual compute group {}, generate new jobIds
periodic={}, event={}, and old jobIds {}",
- virtualGroupInFe, jobIdPeriodic, jobIdEvent,
jobIdsInMs);
+ LOG.info("warmup-vcg rebuild-finish vcgName={} srcCluster={}
dstCluster={} "
+ + "createdPeriodicJobId={}
createdEventJobId={} oldJobIds={}",
+ virtualGroupInFe.getName(), srcCg, dstCg,
jobIdPeriodic, jobIdEvent, jobIdsInMs);
} catch (AnalysisException e) {
- LOG.warn("virtual compute err, name: {}, failed to generate
file cache warm up jobs: {}",
- virtualGroupInFe.getName(), e.getMessage(), e);
+ LOG.warn("warmup-vcg rebuild-failed vcgName={} srcCluster={}
dstCluster={} oldJobIds={} "
+ + "failureReason={}",
+ virtualGroupInFe.getName(), srcCg, dstCg, jobIdsInMs,
e.getMessage(), e);
return;
}
virtualGroupInFe.setNeedRebuildFileCache(false);
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/CacheHotspotManagerTableFilterTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/CacheHotspotManagerTableFilterTest.java
index 255268e66c5..1b1742ece8a 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/CacheHotspotManagerTableFilterTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/CacheHotspotManagerTableFilterTest.java
@@ -567,6 +567,35 @@ public class CacheHotspotManagerTableFilterTest {
Assertions.assertEquals(1, manager.getAllJobInfos(10).size());
}
+ @Test
+ public void
testCreateJobRejectsClusterLevelWhenTableLevelLoadEventExistsAndLogsConflict()
throws Exception {
+ databases.add(mockDb("ods", mockTable(1001, "orders")));
+
+ WarmUpClusterCommand tableLevel = buildEventDrivenStmt("write_cg",
"read_cg",
+ new TableFilterRule(RuleType.INCLUDE, "ods.*"));
+ WarmUpClusterCommand clusterLevel = buildEventDrivenStmt("write_cg",
"read_cg");
+
+ long tableJobId = manager.createJob(tableLevel);
+ RecordingAppender appender = new
RecordingAppender("warmup-conflict-log-test");
+ Logger logger = (Logger)
LogManager.getLogger(CacheHotspotManager.class);
+ appender.start();
+ logger.addAppender(appender);
+ try {
+ AnalysisException exception =
Assertions.assertThrows(AnalysisException.class,
+ () -> manager.createJob(clusterLevel));
+ Assertions.assertTrue(exception.getMessage().contains("Cannot
create cluster-level load-event warm up job"));
+ } finally {
+ logger.removeAppender(appender);
+ appender.stop();
+ }
+
+ String logs = appender.messagesAsString();
+ Assertions.assertTrue(logs.contains("warmup-conflict detect"), logs);
+ Assertions.assertTrue(logs.contains("existingJobId=" + tableJobId),
logs);
+ Assertions.assertTrue(logs.contains("srcCluster=write_cg"), logs);
+ Assertions.assertTrue(logs.contains("dstCluster=read_cg"), logs);
+ }
+
@Test
public void testVirtualComputeGroupCancelsExistingTableLevelLoadEvent()
throws Exception {
databases.add(mockDb("ods", mockTable(1001, "orders")));
@@ -664,6 +693,38 @@ public class CacheHotspotManagerTableFilterTest {
Assertions.assertFalse(cancelReasons.containsKey(finishedJob.getJobId()));
}
+ @Test
+ public void testCancelTableFilterJobsForClusterChangeLogsAffectedJobs()
throws Exception {
+ databases.add(mockDb("ods", mockTable(1001, "orders")));
+
+ CloudWarmUpJob srcMatchedJob = createEventDrivenJob("write_cg",
"read_cg",
+ new TableFilterRule(RuleType.INCLUDE, "ods.*"));
+ CloudWarmUpJob dstMatchedJob = createEventDrivenJob("other_write_cg",
"write_cg",
+ new TableFilterRule(RuleType.INCLUDE, "ods.*"));
+ createEventDrivenJob("other_write_cg", "other_read_cg",
+ new TableFilterRule(RuleType.INCLUDE, "ods.*"));
+
+ CacheHotspotManager spyManager = Mockito.spy(manager);
+ Mockito.doAnswer(invocation ->
null).when(spyManager).cancel(Mockito.anyLong(), Mockito.anyString());
+
+ RecordingAppender appender = new
RecordingAppender("warmup-system-cancel-log-test");
+ Logger logger = (Logger)
LogManager.getLogger(CacheHotspotManager.class);
+ appender.start();
+ logger.addAppender(appender);
+ try {
+ spyManager.cancelTableFilterJobsForClusterChange("write_cg",
+ "system cancel: compute group write_cg renamed to
write_cg_new");
+ } finally {
+ logger.removeAppender(appender);
+ appender.stop();
+ }
+
+ String logs = appender.messagesAsString();
+ Assertions.assertTrue(logs.contains("write_cg"), logs);
+
Assertions.assertTrue(logs.contains(String.valueOf(srcMatchedJob.getJobId())),
logs);
+
Assertions.assertTrue(logs.contains(String.valueOf(dstMatchedJob.getJobId())),
logs);
+ }
+
// ===== Async materialized view (MTMV) matching =====
@Test
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/cache/CacheHotspotManagerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/cache/CacheHotspotManagerTest.java
index e25eb98d140..e921f5d079f 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/cache/CacheHotspotManagerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/cache/CacheHotspotManagerTest.java
@@ -37,6 +37,11 @@ import
org.apache.doris.nereids.trees.plans.commands.WarmUpClusterCommand;
import org.apache.doris.persist.EditLog;
import org.apache.doris.system.Backend;
+import org.apache.logging.log4j.LogManager;
+import org.apache.logging.log4j.core.LogEvent;
+import org.apache.logging.log4j.core.Logger;
+import org.apache.logging.log4j.core.appender.AbstractAppender;
+import org.apache.logging.log4j.core.config.Property;
import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
@@ -340,6 +345,30 @@ public class CacheHotspotManagerTest {
Assert.assertEquals(1,
cacheHotspotManager.getCloudWarmUpJobs().size());
}
+ @Test
+ public void testTryRegisterRunningJobLogsBlockedResult() {
+ CloudWarmUpJob runningJob = newClusterJob(10L, "src_a", "dst_a",
SyncMode.ONCE, JobState.RUNNING, 100L);
+ CloudWarmUpJob blockedJob = newClusterJob(11L, "src_b", "dst_a",
SyncMode.ONCE, JobState.PENDING, 200L);
+
+ RecordingAppender appender = new
RecordingAppender("warmup-lock-register-log-test");
+ Logger logger = (Logger)
LogManager.getLogger(CacheHotspotManager.class);
+ appender.start();
+ logger.addAppender(appender);
+ try {
+
Assert.assertTrue(cacheHotspotManager.tryRegisterRunningJob(runningJob));
+
Assert.assertFalse(cacheHotspotManager.tryRegisterRunningJob(blockedJob));
+ } finally {
+ logger.removeAppender(appender);
+ appender.stop();
+ }
+
+ String logs = appender.messagesAsString();
+ Assert.assertTrue(logs, logs.contains("warmup-lock register"));
+ Assert.assertTrue(logs, logs.contains("jobId=11"));
+ Assert.assertTrue(logs, logs.contains("existingJobId=10"));
+ Assert.assertTrue(logs, logs.contains("registerResult=blocked"));
+ }
+
private WarmUpClusterCommand newTableStmt(String dstClusterName, boolean
force,
Triple<String, String, String>... tables) {
WarmUpClusterCommand stmt = new WarmUpClusterCommand(new ArrayList<>(),
@@ -429,4 +458,21 @@ public class CacheHotspotManagerTest {
}
Assert.fail("Timed out waiting for once pending create lock ref count
" + expectedRefCount);
}
+
+ private static class RecordingAppender extends AbstractAppender {
+ private final List<String> messages = new ArrayList<>();
+
+ RecordingAppender(String name) {
+ super(name, null, null, true, Property.EMPTY_ARRAY);
+ }
+
+ @Override
+ public void append(LogEvent event) {
+ messages.add(event.getMessage().getFormattedMessage());
+ }
+
+ String messagesAsString() {
+ return String.join("\n", messages);
+ }
+ }
}
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudInstanceStatusCheckerTest.java
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudInstanceStatusCheckerTest.java
index ff19f67dcb6..95eb739de64 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudInstanceStatusCheckerTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/cloud/catalog/CloudInstanceStatusCheckerTest.java
@@ -166,7 +166,10 @@ public class CloudInstanceStatusCheckerTest {
String logs = appender.messagesAsString();
Assertions.assertFalse(logs.contains("failed to create virtual compute
group vcg"), logs);
- Assertions.assertTrue(logs.contains("generate new jobIds"), logs);
+ Assertions.assertTrue(logs.contains("warmup-vcg rebuild-start"), logs);
+ Assertions.assertTrue(logs.contains("warmup-vcg rebuild-finish"),
logs);
+ Assertions.assertTrue(logs.contains("srcCluster=active_cg"), logs);
+ Assertions.assertTrue(logs.contains("dstCluster=standby_cg"), logs);
}
private void addComputeGroup(String computeGroupId, String
computeGroupName) {
diff --git
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/WarmupMetricsUtils.groovy
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/WarmupMetricsUtils.groovy
index aa877adb413..846ce4ac34b 100644
---
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/WarmupMetricsUtils.groovy
+++
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/WarmupMetricsUtils.groovy
@@ -167,9 +167,9 @@ class WarmupMetricsUtils {
}
Thread.sleep(2000)
}
- logger.warn("waitForWarmupFinish timed out after ${timeoutMs}ms, " +
- "expected finished >= ${expectedFinished}")
- return getWarmupMetrics(sqlRunner, srcCluster, dstCluster)
+ def latest = getWarmupMetrics(sqlRunner, srcCluster, dstCluster)
+ throw new RuntimeException("waitForWarmupFinish timed out after
${timeoutMs}ms, "
+ + "expected finished >= ${expectedFinished}, latest=${latest}")
}
/**
@@ -265,4 +265,308 @@ 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 boolean isActiveWarmupJob(Map row) {
+ return row.status in ["PENDING", "RUNNING", "WAITING"]
+ }
+
+ static boolean isClusterWarmupJob(Map row) {
+ return row.type == "CLUSTER"
+ }
+
+ 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,
+ Collection<String>
expectedJobIds = null,
+ long observeMs = 5000) {
+ long deadline = System.currentTimeMillis() + timeoutMs
+ List<Map> lastRunningRows = []
+ while (System.currentTimeMillis() < deadline) {
+ List<Map> runningRows = showWarmupJobsByDst(sqlRunner,
dstCluster).findAll {
+ isNormalWarmupJob(it) && it.status == "RUNNING"
+ }
+ lastRunningRows = runningRows
+ if (runningRows.size() > 1) {
+ throw new RuntimeException("expected at most one running
normal warmup on ${dstCluster}, "
+ + "runningRows=${runningRows}")
+ }
+ boolean hasExpectedRunning = expectedJobIds == null ||
runningRows.any {
+ expectedJobIds.contains(it.jobId)
+ }
+ if (runningRows.size() == 1 && hasExpectedRunning) {
+ assertAtMostOneRunningNormalWarmupDuring(sqlRunner,
dstCluster, observeMs)
+ return
+ }
+ Thread.sleep(1000)
+ }
+ throw new RuntimeException("expected one normal warmup to reach
RUNNING on ${dstCluster}, "
+ + "expectedJobIds=${expectedJobIds},
lastRunningRows=${lastRunningRows}")
+ }
+
+ static void assertAtMostOneRunningNormalWarmupDuring(Closure sqlRunner,
String dstCluster, long observeMs) {
+ long deadline = System.currentTimeMillis() + observeMs
+ while (System.currentTimeMillis() < deadline) {
+ List<Map> runningRows = showWarmupJobsByDst(sqlRunner,
dstCluster).findAll {
+ isNormalWarmupJob(it) && it.status == "RUNNING"
+ }
+ if (runningRows.size() > 1) {
+ throw new RuntimeException("expected at most one running
normal warmup on ${dstCluster}, "
+ + "runningRows=${runningRows}")
+ }
+ Thread.sleep(1000)
+ }
+ }
+
+ 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)
+ }
+ throw new RuntimeException("expected periodic job ${jobId} to observe
${minTransitions} "
+ + "RUNNING-to-waiting transitions within ${timeoutMs}ms,
states=${states}")
+ }
+
+ 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<Map> getVcgWarmupJobs(Closure sqlRunner, String srcCluster,
String dstCluster) {
+ return showWarmupJobs(sqlRunner).findAll {
+ it.srcCluster == srcCluster && it.dstCluster == dstCluster
+ && isActiveWarmupJob(it) && isClusterWarmupJob(it)
+ && (it.syncMode.startsWith("PERIODIC") ||
it.syncMode.startsWith("EVENT_DRIVEN"))
+ }
+ }
+
+ static List<String> getVcgWarmupJobIds(Closure sqlRunner, String
srcCluster, String dstCluster) {
+ return getVcgWarmupJobs(sqlRunner, srcCluster, dstCluster).collect {
it.jobId }
+ }
+
+ static boolean hasPeriodicAndEventDrivenWarmup(Collection<Map> rows) {
+ return rows.any { it.syncMode.startsWith("PERIODIC") }
+ && rows.any { it.syncMode.startsWith("EVENT_DRIVEN") }
+ }
+
+ static List<Map> waitForWarmupJobsByPair(Closure sqlRunner, String
srcCluster, String dstCluster,
+ int minCount = 1, long timeoutMs
= 120000) {
+ long deadline = System.currentTimeMillis() + timeoutMs
+ List<Map> rows = []
+ while (System.currentTimeMillis() < deadline) {
+ rows = showWarmupJobs(sqlRunner).findAll {
+ it.srcCluster == srcCluster && it.dstCluster == dstCluster
+ }
+ if (rows.size() >= minCount) {
+ return rows
+ }
+ Thread.sleep(1000)
+ }
+ return rows
+ }
+
+ static List<String> waitForVcgWarmupRecreated(Closure sqlRunner, String
srcCluster, String dstCluster,
+ Collection<String>
oldJobIds, int minNewJobs = 1,
+ long timeoutMs = 120000) {
+ long deadline = System.currentTimeMillis() + timeoutMs
+ List<Map> newRows = []
+ while (System.currentTimeMillis() < deadline) {
+ newRows = getVcgWarmupJobs(sqlRunner, srcCluster,
dstCluster).findAll {
+ !oldJobIds.contains(it.jobId)
+ }
+ if (newRows.size() >= minNewJobs &&
hasPeriodicAndEventDrivenWarmup(newRows)) {
+ return newRows.collect { it.jobId }
+ }
+ Thread.sleep(1000)
+ }
+ throw new RuntimeException("expected active VCG cluster warmup jobs
recreated for "
+ + "${srcCluster}->${dstCluster}, oldJobIds=${oldJobIds},
lastRows=${newRows}")
+ }
+
+ static void assertHistoricalJobsCancelled(Closure sqlRunner,
Collection<String> jobIds,
+ long timeoutMs = 120000) {
+ assertAffectedJobsCancelled(sqlRunner, jobIds.collect { it.toString()
}, timeoutMs)
+ }
+
+ static Map waitForWarmupStatsResume(Closure sqlRunner, Object jobId,
Closure<Boolean> predicate,
+ long timeoutMs = 120000) {
+ return waitForJobSyncStats(sqlRunner, jobId, predicate, timeoutMs)
+ }
+
+ static void waitForJobStatus(Closure sqlRunner, Object jobId,
Collection<String> expectedStatuses,
+ long timeoutMs = 60000) {
+ long deadline = System.currentTimeMillis() + timeoutMs
+ String lastStatus = null
+ while (System.currentTimeMillis() < deadline) {
+ def row = showWarmupJob(sqlRunner, jobId)
+ lastStatus = row.status
+ if (expectedStatuses.contains(lastStatus)) {
+ return
+ }
+ Thread.sleep(1000)
+ }
+ throw new RuntimeException("warmup job ${jobId} status ${lastStatus}
not in ${expectedStatuses}")
+ }
}
diff --git
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_conflict_and_diagnostics.groovy
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_conflict_and_diagnostics.groovy
new file mode 100644
index 00000000000..92725326d4b
--- /dev/null
+++
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_conflict_and_diagnostics.groovy
@@ -0,0 +1,111 @@
+// 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_cluster_event_conflict_and_diagnostics', 'docker') {
+ def options = new ClusterOptions()
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'cloud_warm_up_table_filter_refresh_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
+ options.beNum = 1
+
+ docker(options) {
+ Closure sqlRunner = { String q -> sql(q) }
+
+ def srcCluster = "warmup_conflict_src"
+ def dstCluster = "warmup_conflict_dst"
+ def dbName = "test_warmup_conflict_db"
+ def tableName = "conflict_tbl"
+ def jobIds = []
+
+ cluster.addBackend(1, srcCluster)
+ cluster.addBackend(1, dstCluster)
+
+ 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,
+ value STRING
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("file_cache_ttl_seconds" = "3600")"""
+
+ def clusterJobId = sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load")
+ """)[0][0]
+ jobIds << clusterJobId
+
+ String clusterFirstConflict =
WarmupMetricsUtils.expectCreateConflict(sqlRunner, {
+ sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ ON TABLES (INCLUDE '${dbName}.${tableName}')
+ PROPERTIES ("sync_mode" = "event_driven", "sync_event" =
"load")
+ """)
+ })
+ WarmupMetricsUtils.assertConflictMessage(clusterFirstConflict, [
+ "conflicting", "cluster-level", "table-level", srcCluster,
dstCluster
+ ])
+
+ sql """CANCEL WARM UP JOB WHERE ID = ${clusterJobId}"""
+ def cancelledCluster = WarmupMetricsUtils.showWarmupJob(sqlRunner,
clusterJobId)
+ assertEquals("CANCELLED", cancelledCluster.status)
+
+ def tableJobId = sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ ON TABLES (INCLUDE '${dbName}.${tableName}')
+ PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load")
+ """)[0][0]
+ jobIds << tableJobId
+ assert WarmupMetricsUtils.waitForMatchedTables(sqlRunner, tableJobId,
+ ["${dbName}.${tableName}".toString()] as Set) ==
+ ["${dbName}.${tableName}".toString()] as Set
+
+ String tableFirstConflict =
WarmupMetricsUtils.expectCreateConflict(sqlRunner, {
+ sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ PROPERTIES ("sync_mode" = "event_driven", "sync_event" =
"load")
+ """)
+ })
+ WarmupMetricsUtils.assertConflictMessage(tableFirstConflict, [
+ "conflicting", "table-level", "cluster-level", srcCluster,
dstCluster
+ ])
+
+ sql """CANCEL WARM UP JOB WHERE ID = ${tableJobId}"""
+ def cancelledTable = WarmupMetricsUtils.showWarmupJob(sqlRunner,
tableJobId)
+ assertEquals("CANCELLED", cancelledTable.status)
+
+ jobIds.each { jid ->
+ try {
+ sql """CANCEL WARM UP JOB WHERE ID = ${jid}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_periodic_add_new_be.groovy
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_periodic_add_new_be.groovy
index fa476e80ba6..d0f38a024d4 100644
---
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_periodic_add_new_be.groovy
+++
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_periodic_add_new_be.groovy
@@ -84,11 +84,73 @@ suite('test_warm_up_cluster_periodic_add_new_be', 'docker')
{
}
}
- def getTTLCacheSize = { ip, port ->
- return getBrpcMetrics(ip, port, "ttl_cache_size")
+ def logFileCacheQueueMetrics = { cluster, phase ->
+ def queueMetrics = [
+ normal : [
+ size : "file_cache_normal_queue_cache_size",
+ count: "file_cache_normal_queue_element_count",
+ evict: "file_cache_normal_queue_evict_size",
+ ],
+ index : [
+ size : "file_cache_index_queue_cache_size",
+ count: "file_cache_index_queue_element_count",
+ evict: "file_cache_index_queue_evict_size",
+ ],
+ disposable: [
+ size : "file_cache_disposable_queue_cache_size",
+ count: "file_cache_disposable_queue_element_count",
+ evict: "file_cache_disposable_queue_evict_size",
+ ],
+ ttl : [
+ size : "file_cache_ttl_cache_lru_queue_size",
+ count: "file_cache_ttl_cache_lru_queue_element_count",
+ evict: "file_cache_ttl_cache_evict_size",
+ ],
+ ]
+
+ def backends = sql """SHOW BACKENDS"""
+ def cluster_bes = backends.findAll {
it[19].contains("""\"compute_group_name\" : \"${cluster}\"""") }
+ for (be in cluster_bes) {
+ def ip = be[1]
+ def port = be[5]
+ def fileCacheSize = getBrpcMetrics(ip, port,
"file_cache_cache_size")
+ def queueMetricDetails = []
+ queueMetrics.each { queueName, metrics ->
+ def size = getBrpcMetrics(ip, port, metrics.size)
+ def count = getBrpcMetrics(ip, port, metrics.count)
+ def evict = getBrpcMetrics(ip, port, metrics.evict)
+ queueMetricDetails.add("${queueName}_queue{size=${size},
element_count=${count}, evict_size=${evict}}")
+ }
+ logger.info("${phase} ${cluster} be ${ip}:${port}
file_cache_size=${fileCacheSize}, "
+ + "file_cache_queue_metrics: ${queueMetricDetails.join(',
')}")
+ }
}
- def getClusterTTLCacheSizeSum = { cluster ->
+ def logFileCacheQueueCacheSizeMetrics = { cluster, phase ->
+ def queueCacheSizeMetrics = [
+ normal : "file_cache_normal_queue_cache_size",
+ index : "file_cache_index_queue_cache_size",
+ disposable: "file_cache_disposable_queue_cache_size",
+ ttl : "file_cache_ttl_cache_lru_queue_size",
+ ]
+
+ def backends = sql """SHOW BACKENDS"""
+ def cluster_bes = backends.findAll {
it[19].contains("""\"compute_group_name\" : \"${cluster}\"""") }
+ for (be in cluster_bes) {
+ def ip = be[1]
+ def port = be[5]
+ queueCacheSizeMetrics.each { queueName, metricName ->
+ def size = getBrpcMetrics(ip, port, metricName)
+ logger.info("${phase} ${cluster} be ${ip}:${port}
${queueName}_queue cache_size=${size}")
+ }
+ }
+ }
+
+ def getFileCacheSize = { ip, port ->
+ return getBrpcMetrics(ip, port, "file_cache_cache_size")
+ }
+
+ def getClusterFileCacheSizeSum = { cluster ->
def backends = sql """SHOW BACKENDS"""
def cluster_bes = backends.findAll {
it[19].contains("""\"compute_group_name\" : \"${cluster}\"""") }
@@ -97,44 +159,79 @@ suite('test_warm_up_cluster_periodic_add_new_be',
'docker') {
for (be in cluster_bes) {
def ip = be[1]
def port = be[5]
- def size = getTTLCacheSize(ip, port)
+ def size = getFileCacheSize(ip, port)
sum += size
- logger.info("be be ${ip}:${port} ttl cache size ${size}")
+ logger.info("${cluster} be ${ip}:${port} file cache size ${size}")
}
return sum
}
- def checkTTLCacheSizeSumEqual = { cluster1, cluster2 ->
+ def checkFileCacheSizeSumEqual = { cluster1, cluster2 ->
+ def srcSum = getClusterFileCacheSizeSum(cluster1)
+ def dstSum = getClusterFileCacheSizeSum(cluster2)
+
+ logger.info("file_cache_cache_size: src=${srcSum} dst=${dstSum}")
+ assertTrue(srcSum > 0, "file_cache_cache_size should > 0")
+ assertEquals(srcSum, dstSum)
+ }
+
+ def getClusterMetricSum = { cluster, metricName ->
def backends = sql """SHOW BACKENDS"""
- def srcBes = backends.findAll {
it[19].contains("""\"compute_group_name\" : \"${cluster1}\"""") }
- def tgtBes = backends.findAll {
it[19].contains("""\"compute_group_name\" : \"${cluster2}\"""") }
+ def cluster_bes = backends.findAll {
it[19].contains("""\"compute_group_name\" : \"${cluster}\"""") }
- long srcSum = 0
- for (src in srcBes) {
- def ip = src[1]
- def port = src[5]
- def size = getTTLCacheSize(ip, port)
- srcSum += size
- logger.info("src be ${ip}:${port} ttl cache size ${size}")
+ long sum = 0
+ for (be in cluster_bes) {
+ def ip = be[1]
+ def port = be[5]
+ def value = getBrpcMetrics(ip, port, metricName)
+ sum += value
+ logger.info("${cluster} be ${ip}:${port} ${metricName} ${value}")
}
- long tgtSum = 0
- for (tgt in tgtBes) {
- def ip = tgt[1]
- def port = tgt[5]
- def size = getTTLCacheSize(ip, port)
- tgtSum += size
- logger.info("dst be ${ip}:${port} ttl cache size ${size}")
+ return sum
+ }
+
+ def getPeriodicWarmupMetrics = { cluster ->
+ [
+ submitted: getClusterMetricSum(cluster,
"file_cache_once_or_periodic_warm_up_submitted_segment_num"),
+ finished : getClusterMetricSum(cluster,
"file_cache_once_or_periodic_warm_up_finished_segment_num"),
+ ]
+ }
+
+ def loadCustomerData = { clusterName, tableName, rowCount ->
+ sql """use @${clusterName}"""
+
+ def dataFile =
File.createTempFile("test_warm_up_cluster_periodic_add_new_be_", ".csv")
+ dataFile.deleteOnExit()
+ dataFile.withWriter("UTF-8") { writer ->
+ for (int i = 1; i <= rowCount; i++) {
+ writer.write("${i},name_${i}\n")
+ }
}
- logger.info("ttl_cache_size: src=${srcSum} dst=${tgtSum}")
- assertTrue(srcSum > 0, "ttl_cache_size should > 0")
- assertEquals(srcSum, tgtSum)
+ streamLoad {
+ table tableName
+ set 'column_separator', ','
+ set 'compute_group', clusterName
+ file dataFile.getAbsolutePath()
+ time 60000
+
+ check { result, exception, startTime, endTime ->
+ if (exception != null) {
+ throw exception
+ }
+ logger.info("Stream load result: ${result}".toString())
+ def json = parseJson(result)
+ assertEquals("success", json.Status.toLowerCase())
+ assertEquals(rowCount, json.NumberTotalRows)
+ assertEquals(rowCount, json.NumberLoadedRows)
+ }
+ }
}
- def waitUntil = { condition, timeoutMs ->
+ def waitUntil = { condition, timeoutMs, description ->
long start = System.currentTimeMillis()
while (System.currentTimeMillis() - start < timeoutMs) {
if (condition()) {
@@ -142,6 +239,7 @@ suite('test_warm_up_cluster_periodic_add_new_be', 'docker')
{
}
sleep(1000)
}
+ throw new RuntimeException("Timed out after ${timeoutMs}ms waiting for
${description}")
}
docker(options) {
@@ -172,7 +270,9 @@ suite('test_warm_up_cluster_periodic_add_new_be', 'docker')
{
// Simple setup to simulate data load and access
sql """CREATE TABLE IF NOT EXISTS customer (id INT, name STRING)
DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 3 PROPERTIES
("file_cache_ttl_seconds" = "3600")"""
- sql """INSERT INTO customer VALUES (1, 'A'), (2, 'B'), (3, 'C')"""
+ loadCustomerData(clusterName1, "customer", 3000)
+ logFileCacheQueueCacheSizeMetrics(clusterName1, "after load before
clear cache")
+ logFileCacheQueueCacheSizeMetrics(clusterName2, "after load before
clear cache")
// Start warm up job
def jobId_ = sql """
@@ -193,19 +293,35 @@ suite('test_warm_up_cluster_periodic_add_new_be',
'docker') {
sql """truncate table __internal_schema.cloud_cache_hotspot;"""
clearFileCacheOnAllBackends()
- for (int i = 0; i < 1000; i++) {
+ def beforeWarmupMetrics = getPeriodicWarmupMetrics(clusterName2)
+ logFileCacheQueueMetrics(clusterName1, "after clear cache before
warmup")
+ logFileCacheQueueMetrics(clusterName2, "after clear cache before
warmup")
+ for (int i = 0; i < 500; i++) {
sql """SELECT * FROM customer"""
}
- waitUntil({ getClusterTTLCacheSizeSum(clusterName1) > 0 }, 60000)
- waitUntil({ getClusterTTLCacheSizeSum(clusterName1) ==
getClusterTTLCacheSizeSum(clusterName2) }, 60000)
+ waitUntil({
+ def metrics = getPeriodicWarmupMetrics(clusterName2)
+ metrics.submitted > beforeWarmupMetrics.submitted &&
metrics.finished > beforeWarmupMetrics.finished
+ }, 120000, "periodic warmup metrics to advance from
${beforeWarmupMetrics}")
- def hotspot = sql """select * from
__internal_schema.cloud_cache_hotspot;"""
- logger.info("hotspot: {}", hotspot)
+ waitUntil({ getClusterFileCacheSizeSum(clusterName1) > 0 },
+ 60000, "${clusterName1} file cache size > 0")
+ waitUntil({
+ logFileCacheQueueMetrics(clusterName1, "wait for warmup")
+ logFileCacheQueueMetrics(clusterName2, "wait for warmup")
+ getClusterFileCacheSizeSum(clusterName1) ==
getClusterFileCacheSizeSum(clusterName2)
+ }, 60000, "${clusterName1} and ${clusterName2} file cache size to
match")
logFileCacheDownloadMetrics(clusterName2)
- assertTrue(getClusterTTLCacheSizeSum(clusterName1) > 0)
- checkTTLCacheSizeSumEqual(clusterName1, clusterName2)
+ logFileCacheQueueMetrics(clusterName1, "after warmup")
+ logFileCacheQueueMetrics(clusterName2, "after warmup")
+ def afterWarmupMetrics = getPeriodicWarmupMetrics(clusterName2)
+ assertTrue(afterWarmupMetrics.submitted >
beforeWarmupMetrics.submitted,
+ "periodic warmup should submit segments,
before=${beforeWarmupMetrics}, after=${afterWarmupMetrics}")
+ assertTrue(afterWarmupMetrics.finished > beforeWarmupMetrics.finished,
+ "periodic warmup should finish segments,
before=${beforeWarmupMetrics}, after=${afterWarmupMetrics}")
+ checkFileCacheSizeSumEqual(clusterName1, clusterName2)
def jobInfo = sql """SHOW WARM UP JOB WHERE ID = ${jobId}"""
assertEquals(jobInfo[0][0], jobId)
diff --git
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_periodic_scheduler_semantics.groovy
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_periodic_scheduler_semantics.groovy
new file mode 100644
index 00000000000..022a550ff86
--- /dev/null
+++
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_periodic_scheduler_semantics.groovy
@@ -0,0 +1,137 @@
+// 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_cluster_periodic_scheduler_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_periodic_src"
+ def dstCluster = "warmup_periodic_dst"
+ def dbName = "test_warmup_periodic_scheduler_db"
+ def tableName = "periodic_tbl"
+
+ cluster.addBackend(1, srcCluster)
+ cluster.addBackend(1, dstCluster)
+
+ 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()
+ }
+
+ def getPeriodicWarmupMetrics = {
+ [
+ submitted:
WarmupMetricsUtils.getClusterMetricSum(sqlRunner, dstCluster,
+
"file_cache_once_or_periodic_warm_up_submitted_segment_num"),
+ finished :
WarmupMetricsUtils.getClusterMetricSum(sqlRunner, dstCluster,
+
"file_cache_once_or_periodic_warm_up_finished_segment_num"),
+ ]
+ }
+
+ def waitForPeriodicWarmupFinish = { Map before, long timeoutMs ->
+ long deadline = System.currentTimeMillis() + timeoutMs
+ Map latest = getPeriodicWarmupMetrics()
+ while (System.currentTimeMillis() < deadline) {
+ latest = getPeriodicWarmupMetrics()
+ if (latest.submitted > before.submitted && latest.finished >
before.finished
+ && latest.finished >= latest.submitted) {
+ return latest
+ }
+ sleep(2000)
+ }
+ logger.warn("periodic warmup metrics did not advance after
${timeoutMs}ms, "
+ + "before=${before}, latest=${latest}")
+ return latest
+ }
+
+ sql """use @${srcCluster}"""
+ sql """CREATE DATABASE IF NOT EXISTS ${dbName}"""
+ sql """use ${dbName}"""
+ sql """CREATE TABLE IF NOT EXISTS ${tableName} (
+ id INT,
+ value STRING
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("file_cache_ttl_seconds" = "3600")"""
+ for (int i = 0; i < 12; i++) {
+ sql """INSERT INTO ${tableName} VALUES (${i}, 'before_${i}')"""
+ }
+
+ def periodicJobId = sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ PROPERTIES (
+ "sync_mode" = "periodic",
+ "sync_interval_sec" = "1"
+ )
+ """)[0][0]
+
+ def cycles = WarmupMetricsUtils.waitForPeriodicCycles(sqlRunner,
periodicJobId, 1, 120000)
+ assertTrue(cycles.states.contains("RUNNING"))
+ assertTrue(cycles.states.any { it in ["PENDING", "WAITING"] })
+
+ def timelineBeforeRestart =
WarmupMetricsUtils.sampleJobTimeline(sqlRunner, periodicJobId, 5000, 1000)
+ assertTrue(timelineBeforeRestart*.status.any { it in ["RUNNING",
"PENDING", "WAITING"] })
+
+ restartMasterFe()
+ sql """use @${srcCluster}"""
+ sql """use ${dbName}"""
+
+ def timelineAfterRestart =
WarmupMetricsUtils.sampleJobTimeline(sqlRunner, periodicJobId, 8000, 1000)
+ assertTrue(timelineAfterRestart*.status.any { it in ["RUNNING",
"PENDING", "WAITING"] },
+ "periodic job should continue after FE restart,
timeline=${timelineAfterRestart}")
+
+ def beforeMetrics = getPeriodicWarmupMetrics()
+ for (int i = 12; i < 16; i++) {
+ sql """INSERT INTO ${tableName} VALUES (${i},
'after_restart_${i}')"""
+ }
+ for (int i = 0; i < 1000; i++) {
+ sql """SELECT * FROM ${tableName}"""
+ }
+ def afterMetrics = waitForPeriodicWarmupFinish(beforeMetrics, 120000)
+ assertTrue(afterMetrics.finished > beforeMetrics.finished,
+ "periodic job should trigger another round after restart,
before=${beforeMetrics}, after=${afterMetrics}")
+
+ sql """CANCEL WARM UP JOB WHERE ID = ${periodicJobId}"""
+ def cancelledRow = WarmupMetricsUtils.showWarmupJob(sqlRunner,
periodicJobId)
+ assertEquals("CANCELLED", cancelledRow.status)
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_mixed_cluster_change_and_failover.groovy
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_mixed_cluster_change_and_failover.groovy
new file mode 100644
index 00000000000..cf509bb7701
--- /dev/null
+++
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_mixed_cluster_change_and_failover.groovy
@@ -0,0 +1,117 @@
+// 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 groovy.json.JsonSlurper
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.WarmupMetricsUtils
+
+suite('test_warm_up_mixed_cluster_change_and_failover', 'docker') {
+ def options = new ClusterOptions()
+ options.feNum = 3
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'fetch_cluster_cache_hotspot_interval_ms=1000',
+ 'cloud_warm_up_table_filter_refresh_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
+ options.beNum = 1
+
+ docker(options) {
+ Closure sqlRunner = { String q -> sql(q) }
+ def srcCluster = "warmup_change_src"
+ def dstCluster = "warmup_change_dst"
+ def unrelatedDst = "warmup_change_unrelated_dst"
+ def renamedDst = "warmup_change_dst_renamed"
+ def dbName = "test_warmup_mixed_cluster_change_db"
+ def tableName = "change_tbl"
+ def jobIds = []
+ def metaService = cluster.getAllMetaservices().get(0)
+ def jsonSlurper = new JsonSlurper()
+
+ cluster.addBackend(1, srcCluster)
+ cluster.addBackend(1, dstCluster)
+ cluster.addBackend(1, unrelatedDst)
+
+ def getClusterId = { String clusterName ->
+ def tag = getCloudBeTagByName(clusterName)
+ jsonSlurper.parseText(tag).compute_group_id.toString()
+ }
+
+ sql """use @${srcCluster}"""
+ sql """CREATE DATABASE IF NOT EXISTS ${dbName}"""
+ sql """use ${dbName}"""
+ sql """CREATE TABLE IF NOT EXISTS ${tableName} (
+ id INT,
+ value STRING
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("file_cache_ttl_seconds" = "3600")"""
+
+ def periodicJobId = sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ PROPERTIES ("sync_mode" = "periodic", "sync_interval_sec" = "1")
+ """)[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 unrelatedJobId = sql("""
+ WARM UP CLUSTER ${unrelatedDst} WITH CLUSTER ${srcCluster}
+ PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load")
+ """)[0][0]
+ jobIds.addAll([periodicJobId, eventJobId, unrelatedJobId])
+ WarmupMetricsUtils.waitForJobStatus(sqlRunner, periodicJobId,
["RUNNING", "PENDING", "WAITING"], 60000)
+ WarmupMetricsUtils.waitForJobStatus(sqlRunner, eventJobId, ["RUNNING",
"PENDING"], 60000)
+ WarmupMetricsUtils.waitForJobStatus(sqlRunner, unrelatedJobId,
["RUNNING", "PENDING"], 60000)
+
+ def affectedBefore =
WarmupMetricsUtils.collectWarmupJobsByCluster(sqlRunner, dstCluster)*.jobId
+ assertTrue(affectedBefore.contains(periodicJobId.toString()))
+ assertTrue(affectedBefore.contains(eventJobId.toString()))
+
+ sql """ALTER SYSTEM RENAME COMPUTE GROUP ${dstCluster} ${renamedDst}"""
+ WarmupMetricsUtils.assertAffectedJobsCancelled(sqlRunner,
+ [periodicJobId.toString(), eventJobId.toString()], 120000)
+ WarmupMetricsUtils.assertUnrelatedJobsUnaffected(sqlRunner,
[unrelatedJobId.toString()])
+
+ sql """ALTER SYSTEM RENAME COMPUTE GROUP ${renamedDst} ${dstCluster}"""
+ def recreatedEventId = sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ ON TABLES (INCLUDE '${dbName}.${tableName}')
+ PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load")
+ """)[0][0]
+ jobIds << recreatedEventId
+
+ def unrelatedClusterId = getClusterId(unrelatedDst)
+ drop_cluster(unrelatedDst, unrelatedClusterId, metaService)
+ WarmupMetricsUtils.assertAffectedJobsCancelled(sqlRunner,
[unrelatedJobId.toString()], 120000)
+ WarmupMetricsUtils.assertUnrelatedJobsUnaffected(sqlRunner,
[recreatedEventId.toString()], ["RUNNING", "PENDING", "WAITING"])
+
+ jobIds.each { jid ->
+ try {
+ sql """CANCEL WARM UP JOB WHERE ID = ${jid}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_normal_queue_semantics.groovy
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_normal_queue_semantics.groovy
new file mode 100644
index 00000000000..0c6da79be07
--- /dev/null
+++
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_normal_queue_semantics.groovy
@@ -0,0 +1,146 @@
+// 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
+
+ def normalJobIds = [periodicJobId, onceTableJobId,
onceClusterJobId].collect { it.toString() }
+ WarmupMetricsUtils.waitForOnlyOneRunningNormalWarmup(sqlRunner,
dstCluster, 30000, normalJobIds)
+ 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, normalJobIds)
+ 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.finished >= beforeMetrics.finished + 1,
+ "event-driven warmup should finish after FE restart,
before=${beforeMetrics}, after=${afterMetrics}")
+ assertTrue(afterMetrics.finished + afterMetrics.failed >=
afterMetrics.submitted,
+ "event-driven warmup should drain submitted segments,
after=${afterMetrics}")
+ assertTrue(afterMetrics.requested > beforeMetrics.requested,
+ "event-driven warmup should continue after FE restart,
before=${beforeMetrics}, after=${afterMetrics}")
+
+ [onceTableJobId, onceClusterJobId, periodicJobId, eventJobId].each {
jid ->
+ try {
+ sql """CANCEL WARM UP JOB WHERE ID = ${jid}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/on_tables/test_warm_up_event_on_tables_overlap_semantics.groovy
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/on_tables/test_warm_up_event_on_tables_overlap_semantics.groovy
new file mode 100644
index 00000000000..a7c8ef1bb2a
--- /dev/null
+++
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/on_tables/test_warm_up_event_on_tables_overlap_semantics.groovy
@@ -0,0 +1,125 @@
+// 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_event_on_tables_overlap_semantics', 'docker') {
+ def options = new ClusterOptions()
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'cloud_warm_up_job_scheduler_interval_millisecond=100',
+ 'cloud_warm_up_table_filter_refresh_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
+ options.beNum = 1
+
+ docker(options) {
+ Closure sqlRunner = { String q -> sql(q) }
+
+ def srcCluster = "warmup_overlap_src"
+ def dstCluster = "warmup_overlap_dst"
+ def dbName = "test_warmup_overlap_semantics_db"
+ def tableName = "overlap_tbl"
+ def otherTable = "overlap_other_tbl"
+ def jobIds = []
+
+ cluster.addBackend(1, srcCluster)
+ cluster.addBackend(1, dstCluster)
+
+ sql """use @${srcCluster}"""
+ sql """CREATE DATABASE IF NOT EXISTS ${dbName}"""
+ sql """use ${dbName}"""
+ sql """CREATE TABLE IF NOT EXISTS ${tableName} (id INT, val STRING)
+ DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("file_cache_ttl_seconds" = "3600")"""
+ sql """CREATE TABLE IF NOT EXISTS ${otherTable} (id INT, val STRING)
+ DUPLICATE KEY(id) DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("file_cache_ttl_seconds" = "3600")"""
+
+ def preciseJobId = sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ ON TABLES (INCLUDE '${dbName}.${tableName}')
+ PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load")
+ """)[0][0]
+ def overlapJobId = sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ ON TABLES (
+ INCLUDE '${dbName}.*',
+ EXCLUDE '${dbName}.${otherTable}'
+ )
+ PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load")
+ """)[0][0]
+ def containerJobId = sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ ON TABLES (INCLUDE '${dbName}.*')
+ PROPERTIES ("sync_mode" = "event_driven", "sync_event" = "load")
+ """)[0][0]
+ jobIds.addAll([preciseJobId, overlapJobId, containerJobId])
+
+ assert WarmupMetricsUtils.waitForMatchedTables(sqlRunner, preciseJobId,
+ ["${dbName}.${tableName}".toString()] as Set,
+ ["${dbName}.${otherTable}".toString()] as Set) ==
+ ["${dbName}.${tableName}".toString()] as Set
+ assert WarmupMetricsUtils.waitForMatchedTables(sqlRunner, overlapJobId,
+ ["${dbName}.${tableName}".toString()] as Set,
+ ["${dbName}.${otherTable}".toString()] as Set)
+ .contains("${dbName}.${tableName}".toString())
+ assert WarmupMetricsUtils.waitForMatchedTables(sqlRunner,
containerJobId,
+ ["${dbName}.${tableName}".toString(),
"${dbName}.${otherTable}".toString()] as Set)
+ .containsAll(["${dbName}.${tableName}".toString(),
"${dbName}.${otherTable}".toString()] as Set)
+ jobIds.each { jobId ->
+ WarmupMetricsUtils.waitForJobStatus(sqlRunner, jobId, ["RUNNING"],
60000)
+ }
+ Thread.sleep(1000)
+
+ def baseMetrics = WarmupMetricsUtils.getWarmupMetrics(sqlRunner,
srcCluster, dstCluster)
+ for (int i = 0; i < 4; i++) {
+ sql """INSERT INTO ${tableName} VALUES (${i}, 'target_${i}')"""
+ sql """INSERT INTO ${otherTable} VALUES (${i}, 'other_${i}')"""
+ }
+ def finalMetrics = WarmupMetricsUtils.waitForWarmupFinish(sqlRunner,
srcCluster, dstCluster,
+ baseMetrics.finished + 1, 120000)
+ assertTrue(finalMetrics.requested > baseMetrics.requested,
+ "overlap jobs should all receive triggers,
before=${baseMetrics}, after=${finalMetrics}")
+
+ def preciseStats = WarmupMetricsUtils.waitForJobSyncStats(sqlRunner,
preciseJobId, {
+ it.seg_num.requested_5m >= 4
+ }, 120000)
+ def overlapStats = WarmupMetricsUtils.waitForJobSyncStats(sqlRunner,
overlapJobId, {
+ it.seg_num.requested_5m >= 4
+ }, 120000)
+ def containerStats = WarmupMetricsUtils.waitForJobSyncStats(sqlRunner,
containerJobId, {
+ it.seg_num.requested_5m >= 8
+ }, 120000)
+ assertTrue(preciseStats.seg_num.requested_5m >= 4)
+ assertTrue(overlapStats.seg_num.requested_5m >= 4)
+ assertTrue(containerStats.seg_num.requested_5m >= 8)
+
+ jobIds.each { jid ->
+ try {
+ sql """CANCEL WARM UP JOB WHERE ID = ${jid}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_allows_manual_periodic_jobs.groovy
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_allows_manual_periodic_jobs.groovy
new file mode 100644
index 00000000000..03f2f898c79
--- /dev/null
+++
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_allows_manual_periodic_jobs.groovy
@@ -0,0 +1,103 @@
+// 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 groovy.json.JsonOutput
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.WarmupMetricsUtils
+
+suite('test_vcg_warmup_allows_manual_periodic_jobs', 'docker') {
+ def options = new ClusterOptions()
+ options.feNum = 3
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'sys_log_verbose_modules=org',
+ 'fetch_cluster_cache_hotspot_interval_ms=1000',
+ ]
+ options.cloudMode = true
+
+ docker(options) {
+ Closure sqlRunner = { String q -> sql(q) }
+
+ def ms = cluster.getAllMetaservices().get(0)
+ def msHttpPort = ms.host + ":" + ms.httpPort
+ def instanceId = "default_instance_id"
+ def srcCluster = "vcg_periodic_src"
+ def dstCluster = "vcg_periodic_dst"
+ def extraDst = "vcg_periodic_extra"
+ def vcgName = "vcgWarmupPeriodicAllowed"
+ def vcgId = "vcgWarmupPeriodicAllowedId"
+ def createdJobIds = []
+
+ cluster.addBackend(1, srcCluster)
+ cluster.addBackend(1, dstCluster)
+ cluster.addBackend(1, extraDst)
+
+ def addClusterApi = { requestBody, Closure checkFunc ->
+ httpTest {
+ endpoint msHttpPort
+ uri "/MetaService/http/add_cluster?token=$token"
+ body requestBody
+ check checkFunc
+ }
+ }
+
+ def vcgBody = JsonOutput.toJson([
+ instance_id: instanceId,
+ cluster: [
+ cluster_name : vcgName,
+ cluster_id : vcgId,
+ type : "VIRTUAL",
+ cluster_names : [srcCluster, dstCluster],
+ cluster_policy : [type: "ActiveStandby",
active_cluster_name: srcCluster,
+ standby_cluster_names: [dstCluster]]
+ ]
+ ])
+
+ try {
+ addClusterApi(vcgBody) { respCode, body ->
+ def json = parseJson(body)
+ assertTrue(json.code.equalsIgnoreCase("OK"))
+ }
+
+ def autoRows =
WarmupMetricsUtils.waitForWarmupJobsByPair(sqlRunner, srcCluster, dstCluster,
2, 120000)
+ createdJobIds.addAll(autoRows*.jobId)
+ assertTrue(autoRows.any { it.syncMode.startsWith("PERIODIC") })
+ assertTrue(autoRows.any { it.syncMode.startsWith("EVENT_DRIVEN") })
+
+ def manualPeriodicId = sql("""
+ WARM UP CLUSTER ${extraDst} WITH CLUSTER ${srcCluster}
+ PROPERTIES ("sync_mode" = "periodic", "sync_interval_sec" =
"600")
+ """)[0][0]
+ createdJobIds << manualPeriodicId.toString()
+
+ def manualPeriodicRow =
WarmupMetricsUtils.showWarmupJob(sqlRunner, manualPeriodicId)
+ assertEquals(srcCluster, manualPeriodicRow.srcCluster)
+ assertEquals(extraDst, manualPeriodicRow.dstCluster)
+ assertTrue(manualPeriodicRow.syncMode.startsWith("PERIODIC"))
+
+ def originalAutoRows =
WarmupMetricsUtils.waitForWarmupJobsByPair(sqlRunner, srcCluster, dstCluster,
2, 30000)
+
assertTrue(originalAutoRows*.jobId.toSet().containsAll(autoRows*.jobId as Set))
+ } finally {
+ createdJobIds.each { jobId ->
+ try {
+ sql """CANCEL WARM UP JOB WHERE ID = ${jobId}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_failover_cancels_old_jobs.groovy
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_failover_cancels_old_jobs.groovy
new file mode 100644
index 00000000000..e2bb6a68876
--- /dev/null
+++
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_failover_cancels_old_jobs.groovy
@@ -0,0 +1,125 @@
+// 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 groovy.json.JsonOutput
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.WarmupMetricsUtils
+
+suite('test_vcg_warmup_failover_cancels_old_jobs', 'docker') {
+ def options = new ClusterOptions()
+ options.feNum = 3
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'sys_log_verbose_modules=org',
+ 'fetch_cluster_cache_hotspot_interval_ms=1000',
+ ]
+ options.cloudMode = true
+
+ docker(options) {
+ Closure sqlRunner = { String q -> sql(q) }
+
+ def ms = cluster.getAllMetaservices().get(0)
+ def msHttpPort = ms.host + ":" + ms.httpPort
+ def instanceId = "default_instance_id"
+ def clusterA = "vcg_failover_a"
+ def clusterB = "vcg_failover_b"
+ def vcgName = "vcgWarmupFailoverCancel"
+ def vcgId = "vcgWarmupFailoverCancelId"
+
+ cluster.addBackend(2, clusterA)
+ cluster.addBackend(2, clusterB)
+
+ def addClusterApi = { requestBody, Closure checkFunc ->
+ httpTest {
+ endpoint msHttpPort
+ uri "/MetaService/http/add_cluster?token=$token"
+ body requestBody
+ check checkFunc
+ }
+ }
+
+ def buildBody = { String active, String standby ->
+ JsonOutput.toJson([
+ instance_id: instanceId,
+ cluster: [
+ cluster_name : vcgName,
+ cluster_id : vcgId,
+ type : "VIRTUAL",
+ cluster_names : [clusterA, clusterB],
+ cluster_policy : [type: "ActiveStandby",
active_cluster_name: active,
+ standby_cluster_names:
[standby], failover_failure_threshold: 3]
+ ]
+ ])
+ }
+
+ try {
+ addClusterApi(buildBody(clusterA, clusterB)) { respCode, body ->
+ def json = parseJson(body)
+ assertTrue(json.code.equalsIgnoreCase("OK"))
+ }
+
+ def oldRows =
WarmupMetricsUtils.waitForWarmupJobsByPair(sqlRunner, clusterA, clusterB, 2,
120000)
+ def oldJobIds = oldRows*.jobId
+ assertTrue(oldRows.any { it.syncMode.startsWith("PERIODIC") })
+ assertTrue(oldRows.any { it.syncMode.startsWith("EVENT_DRIVEN") })
+
+ sql """USE @${vcgName}"""
+ sql """DROP TABLE IF EXISTS test_vcg_warmup_failover_cancel_tbl"""
+ sql """
+ CREATE TABLE test_vcg_warmup_failover_cancel_tbl (
+ k1 INT
+ )
+ DUPLICATE KEY(k1)
+ DISTRIBUTED BY HASH(k1) BUCKETS 1
+ PROPERTIES (
+ "replication_num" = "1"
+ )
+ """
+ sql """INSERT INTO test_vcg_warmup_failover_cancel_tbl VALUES
(1)"""
+
+ cluster.stopBackends(4, 5)
+ awaitUntil(50, 3) {
+ sql """USE @${vcgName}"""
+ sql """SELECT COUNT(*) FROM
test_vcg_warmup_failover_cancel_tbl"""
+ def groups = sql_return_maparray """SHOW COMPUTE GROUPS"""
+ def vcg = groups.find { it.Name == vcgName }
+ vcg != null &&
vcg.Policy.contains("\"activeComputeGroup\":\"${clusterB}\"")
+ }
+
+ WarmupMetricsUtils.assertHistoricalJobsCancelled(sqlRunner,
oldJobIds, 120000)
+ def newJobIds =
WarmupMetricsUtils.waitForVcgWarmupRecreated(sqlRunner, clusterB, clusterA,
+ oldJobIds, 2, 120000)
+ assertTrue(newJobIds.size() >= 2)
+ def newRows = newJobIds.collect {
WarmupMetricsUtils.showWarmupJob(sqlRunner, it) }
+ assertTrue(newRows.any { it.syncMode.startsWith("PERIODIC") })
+ assertTrue(newRows.any { it.syncMode.startsWith("EVENT_DRIVEN") })
+ } finally {
+ cluster.startBackends(4, 5)
+ WarmupMetricsUtils.showWarmupJobs(sqlRunner).findAll {
+ it.jobId != null && (
+ (it.srcCluster == clusterA && it.dstCluster ==
clusterB)
+ || (it.srcCluster == clusterB && it.dstCluster
== clusterA)
+ )
+ }.each { row ->
+ try {
+ sql """CANCEL WARM UP JOB WHERE ID = ${row.jobId}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_precedence_and_reconcile.groovy
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_precedence_and_reconcile.groovy
new file mode 100644
index 00000000000..9e416ec104d
--- /dev/null
+++
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_precedence_and_reconcile.groovy
@@ -0,0 +1,153 @@
+// 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 groovy.json.JsonOutput
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.WarmupMetricsUtils
+
+suite('test_vcg_warmup_precedence_and_reconcile', 'docker') {
+ def options = new ClusterOptions()
+ options.feNum = 3
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'sys_log_verbose_modules=org',
+ 'fetch_cluster_cache_hotspot_interval_ms=1000',
+ 'cloud_warm_up_table_filter_refresh_interval_ms=1000',
+ ]
+ options.cloudMode = true
+
+ docker(options) {
+ Closure sqlRunner = { String q -> sql(q) }
+
+ def ms = cluster.getAllMetaservices().get(0)
+ def msHttpPort = ms.host + ":" + ms.httpPort
+ def instanceId = "default_instance_id"
+ def srcCluster = "vcg_prec_src"
+ def dstCluster = "vcg_prec_dst"
+ def extraCluster = "vcg_prec_extra"
+ def vcgName = "vcgWarmupPrecedence"
+ def vcgId = "vcgWarmupPrecedenceId"
+ def dbName = "test_vcg_precedence_db"
+ def tableName = "vcg_tbl"
+
+ cluster.addBackend(1, srcCluster)
+ cluster.addBackend(1, dstCluster)
+ cluster.addBackend(1, extraCluster)
+
+ def addClusterApi = { requestBody, Closure checkFunc ->
+ httpTest {
+ endpoint msHttpPort
+ uri "/MetaService/http/add_cluster?token=$token"
+ body requestBody
+ check checkFunc
+ }
+ }
+
+ def vcgBody = {
+ def clusterPolicy = [type: "ActiveStandby", active_cluster_name:
srcCluster, standby_cluster_names: [dstCluster]]
+ def clusterMap = [cluster_name: vcgName, cluster_id: vcgId, type:
"VIRTUAL",
+ cluster_names: [srcCluster, dstCluster],
cluster_policy: clusterPolicy]
+ return JsonOutput.toJson([instance_id: instanceId, cluster:
clusterMap])
+ }
+
+ def knownJobIds = []
+ try {
+ sql """use @${srcCluster}"""
+ sql """CREATE DATABASE IF NOT EXISTS ${dbName}"""
+ sql """use ${dbName}"""
+ sql """CREATE TABLE IF NOT EXISTS ${tableName} (
+ id INT,
+ value STRING
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("file_cache_ttl_seconds" = "3600")"""
+
+ def tableFirstJobId = sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ ON TABLES (INCLUDE '${dbName}.${tableName}')
+ PROPERTIES ("sync_mode" = "event_driven", "sync_event" =
"load")
+ """)[0][0]
+ knownJobIds << tableFirstJobId.toString()
+
+ addClusterApi(vcgBody()) { respCode, body ->
+ def json = parseJson(body)
+ assertTrue(json.code.equalsIgnoreCase("OK"))
+ }
+
+ WarmupMetricsUtils.assertHistoricalJobsCancelled(sqlRunner,
[tableFirstJobId.toString()], 120000)
+ def cancelledTableJob =
WarmupMetricsUtils.showWarmupJob(sqlRunner, tableFirstJobId)
+ assertTrue(cancelledTableJob.errMsg.contains(
+ "vcg cancel table-level load-event warm up job before
rebuilding file cache jobs"))
+ def newAutoJobs =
WarmupMetricsUtils.waitForVcgWarmupRecreated(sqlRunner, srcCluster, dstCluster,
+ [tableFirstJobId.toString()], 2, 120000)
+ assertTrue(newAutoJobs.size() >= 2,
+ "vcg should recreate auto warmup jobs after table-level
cancellation, newJobs=${newAutoJobs}")
+ knownJobIds.addAll(newAutoJobs)
+ def autoRows = newAutoJobs.collect {
WarmupMetricsUtils.showWarmupJob(sqlRunner, it) }
+ assertTrue(autoRows.any { it.syncMode.startsWith("EVENT_DRIVEN")
&& it.type == "CLUSTER" })
+ assertTrue(autoRows.any { it.syncMode.startsWith("PERIODIC") &&
it.type == "CLUSTER" })
+
+ String clusterConflict =
WarmupMetricsUtils.expectCreateConflict(sqlRunner, {
+ sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ PROPERTIES ("sync_mode" = "event_driven", "sync_event" =
"load")
+ """)
+ })
+ WarmupMetricsUtils.assertConflictMessage(clusterConflict, ["Cannot
create warm up job", srcCluster, dstCluster])
+
+ String tableConflict =
WarmupMetricsUtils.expectCreateConflict(sqlRunner, {
+ sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ ON TABLES (INCLUDE '${dbName}.${tableName}')
+ PROPERTIES ("sync_mode" = "event_driven", "sync_event" =
"load")
+ """)
+ })
+ WarmupMetricsUtils.assertConflictMessage(tableConflict, ["Cannot
create warm up job", srcCluster, dstCluster])
+
+ def secondBody = JsonOutput.toJson([
+ instance_id: instanceId,
+ cluster: [
+ cluster_name : "vcgWarmupPrecedenceShared",
+ cluster_id : "vcgWarmupPrecedenceSharedId",
+ type : "VIRTUAL",
+ cluster_names : [srcCluster, extraCluster],
+ cluster_policy : [type: "ActiveStandby",
active_cluster_name: srcCluster,
+ standby_cluster_names:
[extraCluster]]
+ ]
+ ])
+ addClusterApi(secondBody) { respCode, body ->
+ def json = parseJson(body)
+ assertTrue(!json.code.equalsIgnoreCase("OK"))
+ }
+ } finally {
+ knownJobIds.each { jobId ->
+ try {
+ sql """CANCEL WARM UP JOB WHERE ID = ${jobId}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ try {
+ sql """use @${srcCluster}"""
+ sql """use ${dbName}"""
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ sql """DROP DATABASE IF EXISTS ${dbName}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_restart_master_fe.groovy
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_restart_master_fe.groovy
new file mode 100644
index 00000000000..a301e53ea27
--- /dev/null
+++
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_restart_master_fe.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 groovy.json.JsonOutput
+import groovy.json.JsonSlurper
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.WarmupMetricsUtils
+
+suite('test_vcg_warmup_restart_master_fe', 'docker') {
+ def options = new ClusterOptions()
+ options.feNum = 3
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'sys_log_verbose_modules=org',
+ 'fetch_cluster_cache_hotspot_interval_ms=1000',
+ ]
+ options.cloudMode = true
+
+ docker(options) {
+ Closure sqlRunner = { String q -> sql(q) }
+
+ def ms = cluster.getAllMetaservices().get(0)
+ def msHttpPort = ms.host + ":" + ms.httpPort
+ def instanceId = "default_instance_id"
+ def srcCluster = "vcg_restart_src"
+ def dstCluster = "vcg_restart_dst"
+ def vcgName = "vcgWarmupRestartMasterFe"
+ def vcgId = "vcgWarmupRestartMasterFeId"
+ def knownJobIds = []
+ def jsonSlurper = new JsonSlurper()
+
+ cluster.addBackend(1, srcCluster)
+ cluster.addBackend(1, dstCluster)
+
+ def addClusterApi = { requestBody, Closure checkFunc ->
+ httpTest {
+ endpoint msHttpPort
+ uri "/MetaService/http/add_cluster?token=$token"
+ body requestBody
+ check checkFunc
+ }
+ }
+
+ 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()
+ }
+
+ def getVcgWarmupJobIds = {
+ def groups = sql_return_maparray """SHOW COMPUTE GROUPS"""
+ def vcg = groups.find { it.Name == vcgName }
+ if (vcg == null) {
+ return []
+ }
+ def policy = jsonSlurper.parseText(vcg.Policy)
+ def rawJobIds = policy.cacheWarmupJobIds ?: "[]"
+ if (rawJobIds instanceof Collection) {
+ return rawJobIds.collect { it.toString() }
+ }
+ return jsonSlurper.parseText(rawJobIds.toString()).collect {
it.toString() }
+ }
+
+ def waitForVcgWarmupJobIds = { Collection<String> excludedJobIds ->
+ List<String> jobIds = []
+ awaitUntil(120, 1) {
+ jobIds = getVcgWarmupJobIds().findAll {
+ !excludedJobIds.contains(it)
+ }
+ jobIds.size() >= 2
+ }
+ return jobIds
+ }
+
+ def vcgBody = JsonOutput.toJson([
+ instance_id: instanceId,
+ cluster: [
+ cluster_name : vcgName,
+ cluster_id : vcgId,
+ type : "VIRTUAL",
+ cluster_names : [srcCluster, dstCluster],
+ cluster_policy : [type: "ActiveStandby",
active_cluster_name: srcCluster,
+ standby_cluster_names: [dstCluster]]
+ ]
+ ])
+
+ try {
+ addClusterApi(vcgBody) { respCode, body ->
+ def json = parseJson(body)
+ assertTrue(json.code.equalsIgnoreCase("OK"))
+ }
+
+ def beforeRestartRows =
WarmupMetricsUtils.waitForWarmupJobsByPair(sqlRunner, srcCluster, dstCluster,
2, 120000)
+ knownJobIds.addAll(beforeRestartRows*.jobId)
+ assertTrue(beforeRestartRows.any {
it.syncMode.startsWith("PERIODIC") })
+ assertTrue(beforeRestartRows.any {
it.syncMode.startsWith("EVENT_DRIVEN") })
+ assertTrue(getVcgWarmupJobIds().containsAll(knownJobIds))
+
+ restartMasterFe()
+
+ WarmupMetricsUtils.assertHistoricalJobsCancelled(sqlRunner,
knownJobIds, 120000)
+ def newJobIds = waitForVcgWarmupJobIds(knownJobIds)
+ assertTrue(newJobIds.size() >= 2)
+ def rowsAfterRestart = newJobIds.collect {
WarmupMetricsUtils.showWarmupJob(sqlRunner, it) }
+ assertTrue(rowsAfterRestart.any {
it.syncMode.startsWith("PERIODIC") && it.status in ["RUNNING", "PENDING",
"WAITING"] })
+ assertTrue(rowsAfterRestart.any {
it.syncMode.startsWith("EVENT_DRIVEN") && it.status in ["RUNNING", "PENDING"] })
+ knownJobIds.addAll(newJobIds)
+ } finally {
+ knownJobIds.each { jobId ->
+ try {
+ sql """CANCEL WARM UP JOB WHERE ID = ${jobId}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_restart_target_be.groovy
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_restart_target_be.groovy
new file mode 100644
index 00000000000..0f798b84d95
--- /dev/null
+++
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_restart_target_be.groovy
@@ -0,0 +1,130 @@
+// 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 groovy.json.JsonOutput
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.WarmupMetricsUtils
+
+suite('test_vcg_warmup_restart_target_be', 'docker') {
+ def options = new ClusterOptions()
+ options.feNum = 3
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'sys_log_verbose_modules=org',
+ '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 ms = cluster.getAllMetaservices().get(0)
+ def msHttpPort = ms.host + ":" + ms.httpPort
+ def instanceId = "default_instance_id"
+ def srcCluster = "vcg_be_restart_src"
+ def dstCluster = "vcg_be_restart_dst"
+ def vcgName = "vcgWarmupRestartBe"
+ def vcgId = "vcgWarmupRestartBeId"
+ def dbName = "test_vcg_warmup_restart_be_db"
+ def tableName = "be_restart_tbl"
+ def knownJobIds = []
+
+ cluster.addBackend(1, srcCluster)
+ def dstBeIndexes = cluster.addBackend(1, dstCluster)
+
+ def addClusterApi = { requestBody, Closure checkFunc ->
+ httpTest {
+ endpoint msHttpPort
+ uri "/MetaService/http/add_cluster?token=$token"
+ body requestBody
+ check checkFunc
+ }
+ }
+
+ def vcgBody = JsonOutput.toJson([
+ instance_id: instanceId,
+ cluster: [
+ cluster_name : vcgName,
+ cluster_id : vcgId,
+ type : "VIRTUAL",
+ cluster_names : [srcCluster, dstCluster],
+ cluster_policy : [type: "ActiveStandby",
active_cluster_name: srcCluster,
+ standby_cluster_names: [dstCluster]]
+ ]
+ ])
+
+ try {
+ sql """use @${srcCluster}"""
+ sql """CREATE DATABASE IF NOT EXISTS ${dbName}"""
+ sql """use ${dbName}"""
+ sql """CREATE TABLE IF NOT EXISTS ${tableName} (
+ id INT,
+ value STRING
+ )
+ DUPLICATE KEY(id)
+ DISTRIBUTED BY HASH(id) BUCKETS 1
+ PROPERTIES ("file_cache_ttl_seconds" = "3600")"""
+
+ addClusterApi(vcgBody) { respCode, body ->
+ def json = parseJson(body)
+ assertTrue(json.code.equalsIgnoreCase("OK"))
+ }
+
+ def beforeRestartRows =
WarmupMetricsUtils.waitForWarmupJobsByPair(sqlRunner, srcCluster, dstCluster,
2, 120000)
+ knownJobIds.addAll(beforeRestartRows*.jobId)
+ assertTrue(beforeRestartRows.any {
it.syncMode.startsWith("PERIODIC") })
+ assertTrue(beforeRestartRows.any {
it.syncMode.startsWith("EVENT_DRIVEN") })
+
+ cluster.restartBackends(dstBeIndexes[0] as int)
+ sleep(8000)
+ sql """use @${srcCluster}"""
+ sql """use ${dbName}"""
+
+ def afterRestartRows = knownJobIds.collect {
WarmupMetricsUtils.showWarmupJob(sqlRunner, it) }
+ assertTrue(afterRestartRows.any {
it.syncMode.startsWith("PERIODIC") && it.status in ["RUNNING", "PENDING",
"WAITING"] })
+ assertTrue(afterRestartRows.any {
it.syncMode.startsWith("EVENT_DRIVEN") && it.status in ["RUNNING", "PENDING"] })
+
+ def beforeMetrics = WarmupMetricsUtils.getWarmupMetrics(sqlRunner,
srcCluster, dstCluster)
+ for (int i = 0; i < 4; i++) {
+ sql """INSERT INTO ${tableName} VALUES (${i},
'after_be_restart_${i}')"""
+ }
+ def afterMetrics =
WarmupMetricsUtils.waitForWarmupFinish(sqlRunner, srcCluster, dstCluster,
+ beforeMetrics.finished + 1, 120000)
+ assertTrue(afterMetrics.requested > beforeMetrics.requested,
+ "vcg auto event-driven warmup should continue after target
BE restart, before=${beforeMetrics}, after=${afterMetrics}")
+ } finally {
+ knownJobIds.each { jobId ->
+ try {
+ sql """CANCEL WARM UP JOB WHERE ID = ${jobId}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ try {
+ sql """use @${srcCluster}"""
+ sql """use ${dbName}"""
+ sql """DROP TABLE IF EXISTS ${tableName}"""
+ sql """DROP DATABASE IF EXISTS ${dbName}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_shared_compute_group_conflict.groovy
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_shared_compute_group_conflict.groovy
new file mode 100644
index 00000000000..71cffefba8f
--- /dev/null
+++
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_shared_compute_group_conflict.groovy
@@ -0,0 +1,99 @@
+// 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 groovy.json.JsonOutput
+import org.apache.doris.regression.suite.ClusterOptions
+
+suite('test_vcg_warmup_shared_compute_group_conflict', 'docker') {
+ def options = new ClusterOptions()
+ options.feNum = 3
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'sys_log_verbose_modules=org',
+ ]
+ options.cloudMode = true
+
+ docker(options) {
+ def ms = cluster.getAllMetaservices().get(0)
+ def msHttpPort = ms.host + ":" + ms.httpPort
+ def instanceId = "default_instance_id"
+ def clusterA = "vcg_share_a"
+ def clusterB = "vcg_share_b"
+ def clusterC = "vcg_share_c"
+
+ cluster.addBackend(1, clusterA)
+ cluster.addBackend(1, clusterB)
+ cluster.addBackend(1, clusterC)
+
+ def addClusterApi = { requestBody, Closure checkFunc ->
+ httpTest {
+ endpoint msHttpPort
+ uri "/MetaService/http/add_cluster?token=$token"
+ body requestBody
+ check checkFunc
+ }
+ }
+
+ def firstBody = JsonOutput.toJson([
+ instance_id: instanceId,
+ cluster: [
+ cluster_name : "vcgShareFirst",
+ cluster_id : "vcgShareFirstId",
+ type : "VIRTUAL",
+ cluster_names : [clusterA, clusterB],
+ cluster_policy : [type: "ActiveStandby",
active_cluster_name: clusterA,
+ standby_cluster_names: [clusterB]]
+ ]
+ ])
+ addClusterApi(firstBody) { respCode, body ->
+ def json = parseJson(body)
+ assertTrue(json.code.equalsIgnoreCase("OK"))
+ }
+
+ def secondBody = JsonOutput.toJson([
+ instance_id: instanceId,
+ cluster: [
+ cluster_name : "vcgShareSecond",
+ cluster_id : "vcgShareSecondId",
+ type : "VIRTUAL",
+ cluster_names : [clusterA, clusterC],
+ cluster_policy : [type: "ActiveStandby",
active_cluster_name: clusterA,
+ standby_cluster_names: [clusterC]]
+ ]
+ ])
+ addClusterApi(secondBody) { respCode, body ->
+ def json = parseJson(body)
+ assertTrue(!json.code.equalsIgnoreCase("OK"))
+ }
+
+ def reverseSharedBody = JsonOutput.toJson([
+ instance_id: instanceId,
+ cluster: [
+ cluster_name : "vcgShareThird",
+ cluster_id : "vcgShareThirdId",
+ type : "VIRTUAL",
+ cluster_names : [clusterC, clusterB],
+ cluster_policy : [type: "ActiveStandby",
active_cluster_name: clusterC,
+ standby_cluster_names: [clusterB]]
+ ]
+ ])
+ addClusterApi(reverseSharedBody) { respCode, body ->
+ def json = parseJson(body)
+ assertTrue(!json.code.equalsIgnoreCase("OK"))
+ }
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_switch_and_drop_cancels_history.groovy
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_switch_and_drop_cancels_history.groovy
new file mode 100644
index 00000000000..d6410518fab
--- /dev/null
+++
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_switch_and_drop_cancels_history.groovy
@@ -0,0 +1,130 @@
+// 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 groovy.json.JsonOutput
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.WarmupMetricsUtils
+
+suite('test_vcg_warmup_switch_and_drop_cancels_history', 'docker') {
+ def options = new ClusterOptions()
+ options.feNum = 3
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'sys_log_verbose_modules=org',
+ 'fetch_cluster_cache_hotspot_interval_ms=1000',
+ ]
+ options.cloudMode = true
+
+ docker(options) {
+ Closure sqlRunner = { String q -> sql(q) }
+
+ def ms = cluster.getAllMetaservices().get(0)
+ def msHttpPort = ms.host + ":" + ms.httpPort
+ def instanceId = "default_instance_id"
+ def srcCluster = "vcg_switch_src"
+ def dstCluster = "vcg_switch_dst"
+ def vcgName = "vcgWarmupSwitchHistory"
+ def vcgId = "vcgWarmupSwitchHistoryId"
+
+ cluster.addBackend(1, srcCluster)
+ cluster.addBackend(1, dstCluster)
+
+ def logShowClusters = { String message ->
+ def rows = sql """SHOW CLUSTERS"""
+ log.info("{}: {}", message, rows)
+ }
+
+ def logWarmupJobs = { String message, Collection jobIds ->
+ jobIds.each { jobId ->
+ def rows = sql """SHOW WARM UP JOB WHERE ID = ${jobId}"""
+ log.info("{} jobId={}: {}", message, jobId, rows)
+ }
+ }
+
+ def addClusterApi = { requestBody, Closure checkFunc ->
+ httpTest {
+ endpoint msHttpPort
+ uri "/MetaService/http/add_cluster?token=$token"
+ body requestBody
+ check checkFunc
+ }
+ }
+
+ def alterClusterInfoApi = { requestBody, Closure checkFunc ->
+ httpTest {
+ endpoint msHttpPort
+ uri "/MetaService/http/alter_vcluster_info?token=$token"
+ body requestBody
+ check checkFunc
+ }
+ }
+
+ def buildVcgBody = { String active, String standby ->
+ def clusterPolicy = [type: "ActiveStandby", active_cluster_name:
active,
+ standby_cluster_names: [standby]]
+ def clusterMap = [cluster_name: vcgName, cluster_id: vcgId, type:
"VIRTUAL",
+ cluster_names: [srcCluster, dstCluster],
cluster_policy: clusterPolicy]
+ return JsonOutput.toJson([instance_id: instanceId, cluster:
clusterMap])
+ }
+
+ try {
+ addClusterApi(buildVcgBody(srcCluster, dstCluster)) { respCode,
body ->
+ def json = parseJson(body)
+ assertTrue(json.code.equalsIgnoreCase("OK"))
+ }
+ logShowClusters("show clusters after vcg create")
+
+ def initialRows =
WarmupMetricsUtils.waitForWarmupJobsByPair(sqlRunner, srcCluster, dstCluster,
2, 120000)
+ def initialJobIds = initialRows*.jobId
+ log.info("show warm up jobs after vcg create: {}", initialRows)
+ logWarmupJobs("show warm up job detail after vcg create",
initialJobIds)
+ assertTrue(initialRows.any { it.type == "CLUSTER" &&
it.syncMode.startsWith("PERIODIC") })
+ assertTrue(initialRows.any { it.type == "CLUSTER" &&
it.syncMode.startsWith("EVENT_DRIVEN") })
+
+ alterClusterInfoApi(buildVcgBody(dstCluster, srcCluster)) {
respCode, body ->
+ def json = parseJson(body)
+ assertTrue(json.code.equalsIgnoreCase("OK"))
+ }
+
+ WarmupMetricsUtils.assertHistoricalJobsCancelled(sqlRunner,
initialJobIds, 120000)
+ def switchedJobIds =
WarmupMetricsUtils.waitForVcgWarmupRecreated(sqlRunner, dstCluster, srcCluster,
+ initialJobIds, 2, 120000)
+ assertTrue(switchedJobIds.size() >= 2)
+ def switchedRows = switchedJobIds.collect {
WarmupMetricsUtils.showWarmupJob(sqlRunner, it) }
+ log.info("show warm up jobs after vcg switch: {}", switchedRows)
+ logWarmupJobs("show warm up job detail after vcg switch",
switchedJobIds)
+ assertTrue(switchedRows.any { it.type == "CLUSTER" &&
it.syncMode.startsWith("PERIODIC") })
+ assertTrue(switchedRows.any { it.type == "CLUSTER" &&
it.syncMode.startsWith("EVENT_DRIVEN") })
+
+ drop_cluster(vcgName, vcgId, ms)
+ WarmupMetricsUtils.assertHistoricalJobsCancelled(sqlRunner,
switchedJobIds, 120000)
+ logWarmupJobs("show warm up job detail after vcg drop",
switchedJobIds)
+ } finally {
+ WarmupMetricsUtils.showWarmupJobs(sqlRunner).findAll {
+ it.jobId != null && (
+ (it.srcCluster == srcCluster && it.dstCluster ==
dstCluster)
+ || (it.srcCluster == dstCluster &&
it.dstCluster == srcCluster)
+ )
+ }.each { row ->
+ try {
+ sql """CANCEL WARM UP JOB WHERE ID = ${row.jobId}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_waits_for_existing_cluster_jobs.groovy
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_waits_for_existing_cluster_jobs.groovy
new file mode 100644
index 00000000000..b9faa241b57
--- /dev/null
+++
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_waits_for_existing_cluster_jobs.groovy
@@ -0,0 +1,107 @@
+// 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 groovy.json.JsonOutput
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.WarmupMetricsUtils
+
+suite('test_vcg_warmup_waits_for_existing_cluster_jobs', 'docker') {
+ def options = new ClusterOptions()
+ options.feNum = 3
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'sys_log_verbose_modules=org',
+ 'fetch_cluster_cache_hotspot_interval_ms=1000',
+ ]
+ options.cloudMode = true
+
+ docker(options) {
+ Closure sqlRunner = { String q -> sql(q) }
+
+ def ms = cluster.getAllMetaservices().get(0)
+ def msHttpPort = ms.host + ":" + ms.httpPort
+ def instanceId = "default_instance_id"
+ def srcCluster = "vcg_wait_src"
+ def dstCluster = "vcg_wait_dst"
+ def vcgName = "vcgWarmupWaitsExisting"
+ def vcgId = "vcgWarmupWaitsExistingId"
+ def allKnownJobIds = []
+
+ cluster.addBackend(1, srcCluster)
+ cluster.addBackend(1, dstCluster)
+
+ def addClusterApi = { requestBody, Closure checkFunc ->
+ httpTest {
+ endpoint msHttpPort
+ uri "/MetaService/http/add_cluster?token=$token"
+ body requestBody
+ check checkFunc
+ }
+ }
+
+ def vcgBody = {
+ def clusterPolicy = [type: "ActiveStandby", active_cluster_name:
srcCluster,
+ standby_cluster_names: [dstCluster]]
+ def clusterMap = [cluster_name: vcgName, cluster_id: vcgId, type:
"VIRTUAL",
+ cluster_names: [srcCluster, dstCluster],
cluster_policy: clusterPolicy]
+ return JsonOutput.toJson([instance_id: instanceId, cluster:
clusterMap])
+ }
+
+ try {
+ def periodicJobId = sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ PROPERTIES ("sync_mode" = "periodic", "sync_interval_sec" =
"600")
+ """)[0][0]
+ def clusterEventJobId = sql("""
+ WARM UP CLUSTER ${dstCluster} WITH CLUSTER ${srcCluster}
+ PROPERTIES ("sync_mode" = "event_driven", "sync_event" =
"load")
+ """)[0][0]
+ allKnownJobIds.addAll([periodicJobId.toString(),
clusterEventJobId.toString()])
+
+ addClusterApi(vcgBody()) { respCode, body ->
+ def json = parseJson(body)
+ assertTrue(json.code.equalsIgnoreCase("OK"))
+ }
+
+ sleep(8000)
+ def rowsBeforeCancel =
WarmupMetricsUtils.waitForWarmupJobsByPair(sqlRunner, srcCluster, dstCluster,
2, 30000)
+ assertEquals(2, rowsBeforeCancel.size())
+ assertTrue(rowsBeforeCancel*.jobId.toSet() ==
[periodicJobId.toString(), clusterEventJobId.toString()] as Set,
+ "vcg should wait existing cluster-level jobs before
creating auto jobs, rows=${rowsBeforeCancel}")
+
+ sql """CANCEL WARM UP JOB WHERE ID = ${periodicJobId}"""
+ sql """CANCEL WARM UP JOB WHERE ID = ${clusterEventJobId}"""
+ WarmupMetricsUtils.assertHistoricalJobsCancelled(sqlRunner,
+ [periodicJobId.toString(), clusterEventJobId.toString()],
120000)
+
+ def recreatedJobIds =
WarmupMetricsUtils.waitForVcgWarmupRecreated(sqlRunner, srcCluster, dstCluster,
+ [periodicJobId.toString(), clusterEventJobId.toString()],
2, 120000)
+ allKnownJobIds.addAll(recreatedJobIds)
+ assertTrue(recreatedJobIds.size() >= 2)
+ def recreatedRows = recreatedJobIds.collect {
WarmupMetricsUtils.showWarmupJob(sqlRunner, it) }
+ assertTrue(recreatedRows.any { it.type == "CLUSTER" &&
it.syncMode.startsWith("PERIODIC") })
+ assertTrue(recreatedRows.any { it.type == "CLUSTER" &&
it.syncMode.startsWith("EVENT_DRIVEN") })
+ } finally {
+ allKnownJobIds.each { jobId ->
+ try {
+ sql """CANCEL WARM UP JOB WHERE ID = ${jobId}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_with_manual_once_queue_semantics.groovy
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_with_manual_once_queue_semantics.groovy
new file mode 100644
index 00000000000..7932d9d3681
--- /dev/null
+++
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_with_manual_once_queue_semantics.groovy
@@ -0,0 +1,111 @@
+// 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 groovy.json.JsonOutput
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.WarmupMetricsUtils
+
+suite('test_vcg_warmup_with_manual_once_queue_semantics', 'docker') {
+ def options = new ClusterOptions()
+ options.feNum = 3
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'sys_log_verbose_modules=org',
+ 'fetch_cluster_cache_hotspot_interval_ms=1000',
+ ]
+ options.cloudMode = true
+
+ docker(options) {
+ Closure sqlRunner = { String q -> sql(q) }
+
+ def ms = cluster.getAllMetaservices().get(0)
+ def msHttpPort = ms.host + ":" + ms.httpPort
+ def instanceId = "default_instance_id"
+ def vcgSrc = "vcg_queue_src"
+ def sharedDst = "vcg_queue_dst"
+ def manualSrc = "vcg_queue_manual_src"
+ def vcgName = "vcgWarmupManualOnceQueue"
+ def vcgId = "vcgWarmupManualOnceQueueId"
+ def createdJobIds = []
+
+ cluster.addBackend(1, vcgSrc)
+ cluster.addBackend(1, sharedDst)
+ cluster.addBackend(1, manualSrc)
+
+ def addClusterApi = { requestBody, Closure checkFunc ->
+ httpTest {
+ endpoint msHttpPort
+ uri "/MetaService/http/add_cluster?token=$token"
+ body requestBody
+ check checkFunc
+ }
+ }
+
+ def vcgBody = JsonOutput.toJson([
+ instance_id: instanceId,
+ cluster: [
+ cluster_name : vcgName,
+ cluster_id : vcgId,
+ type : "VIRTUAL",
+ cluster_names : [vcgSrc, sharedDst],
+ cluster_policy : [type: "ActiveStandby",
active_cluster_name: vcgSrc,
+ standby_cluster_names: [sharedDst]]
+ ]
+ ])
+
+ try {
+ addClusterApi(vcgBody) { respCode, body ->
+ def json = parseJson(body)
+ assertTrue(json.code.equalsIgnoreCase("OK"))
+ }
+
+ def autoRows =
WarmupMetricsUtils.waitForWarmupJobsByPair(sqlRunner, vcgSrc, sharedDst, 2,
120000)
+ createdJobIds.addAll(autoRows*.jobId)
+ assertTrue(autoRows.any { it.syncMode.startsWith("PERIODIC") })
+ assertTrue(autoRows.any { it.syncMode.startsWith("EVENT_DRIVEN") })
+
+ def manualOnceId = sql("""WARM UP CLUSTER ${sharedDst} WITH
CLUSTER ${manualSrc}""")[0][0]
+ createdJobIds << manualOnceId.toString()
+
+ def rowsByDst = WarmupMetricsUtils.showWarmupJobsByDst(sqlRunner,
sharedDst)
+ assertTrue(rowsByDst*.jobId.contains(manualOnceId.toString()))
+
+ def expectedNormalJobIds = (autoRows.findAll {
WarmupMetricsUtils.isNormalWarmupJob(it) }*.jobId
+ + [manualOnceId.toString()])
+ WarmupMetricsUtils.waitForOnlyOneRunningNormalWarmup(sqlRunner,
sharedDst, 30000, expectedNormalJobIds)
+ long runningNormal =
WarmupMetricsUtils.countRunningNormalWarmupByDst(sqlRunner, sharedDst)
+ assertTrue(runningNormal <= 1,
+ "vcg auto periodic plus manual once should still serialize
normal warmup on same dst, "
+ + "running=${runningNormal}")
+
+ def manualOnceRow = WarmupMetricsUtils.showWarmupJob(sqlRunner,
manualOnceId)
+ assertTrue(WarmupMetricsUtils.isNormalWarmupJob(manualOnceRow))
+ assertEquals(sharedDst, manualOnceRow.dstCluster)
+
+ def eventRows = rowsByDst.findAll {
it.syncMode.startsWith("EVENT_DRIVEN") }
+ assertTrue(eventRows.size() >= 1)
+ assertTrue(eventRows.every { it.status in ["RUNNING", "PENDING"] })
+ } finally {
+ createdJobIds.each { jobId ->
+ try {
+ sql """CANCEL WARM UP JOB WHERE ID = ${jobId}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+ }
+}
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_with_manual_periodic_queue_semantics.groovy
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_with_manual_periodic_queue_semantics.groovy
new file mode 100644
index 00000000000..185013a9e1b
--- /dev/null
+++
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/test_vcg_warmup_with_manual_periodic_queue_semantics.groovy
@@ -0,0 +1,110 @@
+// 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 groovy.json.JsonOutput
+import org.apache.doris.regression.suite.ClusterOptions
+import org.apache.doris.regression.util.WarmupMetricsUtils
+
+suite('test_vcg_warmup_with_manual_periodic_queue_semantics', 'docker') {
+ def options = new ClusterOptions()
+ options.feNum = 3
+ options.feConfigs += [
+ 'cloud_cluster_check_interval_second=1',
+ 'sys_log_verbose_modules=org',
+ 'fetch_cluster_cache_hotspot_interval_ms=1000',
+ ]
+ options.cloudMode = true
+
+ docker(options) {
+ Closure sqlRunner = { String q -> sql(q) }
+
+ def ms = cluster.getAllMetaservices().get(0)
+ def msHttpPort = ms.host + ":" + ms.httpPort
+ def instanceId = "default_instance_id"
+ def vcgSrc = "vcg_periodic_queue_src"
+ def sharedDst = "vcg_periodic_queue_dst"
+ def manualSrc = "vcg_periodic_queue_manual_src"
+ def vcgName = "vcgWarmupManualPeriodicQueue"
+ def vcgId = "vcgWarmupManualPeriodicQueueId"
+ def createdJobIds = []
+
+ cluster.addBackend(1, vcgSrc)
+ cluster.addBackend(1, sharedDst)
+ cluster.addBackend(1, manualSrc)
+
+ def addClusterApi = { requestBody, Closure checkFunc ->
+ httpTest {
+ endpoint msHttpPort
+ uri "/MetaService/http/add_cluster?token=$token"
+ body requestBody
+ check checkFunc
+ }
+ }
+
+ def vcgBody = JsonOutput.toJson([
+ instance_id: instanceId,
+ cluster: [
+ cluster_name : vcgName,
+ cluster_id : vcgId,
+ type : "VIRTUAL",
+ cluster_names : [vcgSrc, sharedDst],
+ cluster_policy : [type: "ActiveStandby",
active_cluster_name: vcgSrc,
+ standby_cluster_names: [sharedDst]]
+ ]
+ ])
+
+ try {
+ addClusterApi(vcgBody) { respCode, body ->
+ def json = parseJson(body)
+ assertTrue(json.code.equalsIgnoreCase("OK"))
+ }
+
+ def autoRows =
WarmupMetricsUtils.waitForWarmupJobsByPair(sqlRunner, vcgSrc, sharedDst, 2,
120000)
+ createdJobIds.addAll(autoRows*.jobId)
+ assertTrue(autoRows.any { it.syncMode.startsWith("PERIODIC") })
+ assertTrue(autoRows.any { it.syncMode.startsWith("EVENT_DRIVEN") })
+
+ def manualPeriodicId = sql("""
+ WARM UP CLUSTER ${sharedDst} WITH CLUSTER ${manualSrc}
+ PROPERTIES ("sync_mode" = "periodic", "sync_interval_sec" =
"600")
+ """)[0][0]
+ createdJobIds << manualPeriodicId.toString()
+
+ def rowsByDst = WarmupMetricsUtils.showWarmupJobsByDst(sqlRunner,
sharedDst)
+ assertTrue(rowsByDst*.jobId.contains(manualPeriodicId.toString()))
+
+ def expectedNormalJobIds = (autoRows.findAll {
WarmupMetricsUtils.isNormalWarmupJob(it) }*.jobId
+ + [manualPeriodicId.toString()])
+ WarmupMetricsUtils.waitForOnlyOneRunningNormalWarmup(sqlRunner,
sharedDst, 30000, expectedNormalJobIds)
+ long runningNormal =
WarmupMetricsUtils.countRunningNormalWarmupByDst(sqlRunner, sharedDst)
+ assertTrue(runningNormal <= 1,
+ "vcg auto periodic plus manual periodic should still
serialize normal warmup on same dst, "
+ + "running=${runningNormal}")
+
+ def manualPeriodicRow =
WarmupMetricsUtils.showWarmupJob(sqlRunner, manualPeriodicId)
+ assertTrue(manualPeriodicRow.syncMode.startsWith("PERIODIC"))
+ assertEquals(sharedDst, manualPeriodicRow.dstCluster)
+ } finally {
+ createdJobIds.each { jobId ->
+ try {
+ sql """CANCEL WARM UP JOB WHERE ID = ${jobId}"""
+ } catch (Throwable ignored) {
+ }
+ }
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
