This is an automated email from the ASF dual-hosted git repository.
deardeng pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/master by this push:
new 296a3bbbffb [fix](regression) Fix docker regression case assertions
(#65201)
296a3bbbffb is described below
commit 296a3bbbffb3634b66b207a76ae0a60e7ccd4992
Author: chunping <[email protected]>
AuthorDate: Fri Jul 10 14:01:56 2026 +0800
[fix](regression) Fix docker regression case assertions (#65201)
### What problem does this PR solve?
Issue Number: None
Related PR: None
Problem Summary: Fix several unstable docker regression cases. LRU
persist now validates restart and dump availability instead of comparing
dump md5 across BE stop/start. TeamCity reporting now resolves the
actual failing Groovy stack frame to avoid reporting failures against
the wrong running suite. Cloud/file cache and compaction cases now avoid
exact segment-count assumptions, poll stale rowset file cache state, and
print rowset, unused_rowsets_count, and file cache diagnostics on
timeout.
### Release note
None
### Check List (For Author)
- Test: Manual test
- Ran git diff --check for the synchronized regression files.
- Did not run full docker regression locally.
- Behavior changed: No
- Does this need documentation: No
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #xxx
Problem Summary:
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [x] Regression test
- [ ] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [x] No.
- [ ] Yes. <!-- Explain the behavior change -->
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
### Check List (For Reviewer who merge this PR)
- [ ] Confirm the release note
- [x] Confirm test cases
- [ ] Confirm document
- [ ] Add branch pick label <!-- Add branch pick label that this PR
should merge into -->
---
regression-test/README.md | 19 ++-
.../cache/ttl/ex06_ttl_restart_consistency.out | 8 +-
.../doris/regression/suite/SuiteCluster.groovy | 2 +-
.../suite/event/TeamcityEventListener.groovy | 8 +-
.../doris/regression/util/LoggerUtils.groovy | 42 +++++++
...est_balance_use_compute_group_properties.groovy | 19 +++
.../test_immediate_warmup_multi_segments.groovy | 24 ++--
...est_warm_up_cluster_event_cancel_expired.groovy | 15 ++-
...rm_up_cluster_event_compaction_sync_wait.groovy | 5 +-
...uster_event_compaction_sync_wait_timeout.groovy | 4 +-
...test_warm_up_cluster_event_schema_change.groovy | 73 ++++++++++-
...rm_up_cluster_periodic_restart_master_fe.groovy | 40 ++++++
.../cache/ttl/ex06_ttl_restart_consistency.groovy | 13 +-
.../cache/warm_up/test_warmup_table_docker.groovy | 63 +++++++---
.../default_vcg_auto_failover.groovy | 68 +++++------
.../test_clean_tablet_when_drop_force_table.groovy | 89 +++++++++-----
...e_compaction_and_read_stale_cloud_docker.groovy | 75 +++++++++---
...ultisegments_and_read_stale_cloud_docker.groovy | 135 ++++++++++++++++-----
.../suites/demo_p0/test_lru_persist.groovy | 31 +++--
.../test_cloud_concurrent_calc_dbm_task.groovy | 15 +--
...est_cloud_full_compaction_multi_segments.groovy | 17 +--
.../test_cloud_mow_retry_txn_interleave.groovy | 11 +-
22 files changed, 567 insertions(+), 209 deletions(-)
diff --git a/regression-test/README.md b/regression-test/README.md
index e28865d8e14..f5ad7cc9cda 100644
--- a/regression-test/README.md
+++ b/regression-test/README.md
@@ -94,9 +94,26 @@ under the License.
Also read the
[doris-compose](https://github.com/apache/doris/tree/master/docker/runtime/doris-compose)
readme.
+10. Do not use raw `Thread.start` or `Thread.startDaemon` to run SQL in a
case. Use the framework `thread` or `extraThread` helper and call
`future.get()` to wait for completion, so exceptions are reported to the case
that created the thread.
+
+ Problematic code:
+ ```
+ def t = Thread.start {
+ sql """insert into tbl values (1)"""
+ }
+ t.join()
+ ```
+
+ Correct code:
+ ```
+ def future = thread("insert-thread") {
+ sql """insert into tbl values (1)"""
+ }
+ future.get()
+ ```
+
## Compatibility case
Refers to the resources or rules created on the initial cluster during FE
testing or upgrade testing, which can still be used normally after the cluster
restart or upgrade, such as permissions, UDF, etc.
These cases need to be split into two files, `load.groovy` and `xxxx.groovy`,
placed in a folder, and tagged with the `restart_fe` group label,
[example](https://github.com/apache/doris/pull/37118).
-
diff --git
a/regression-test/data/cloud_p0/cache/ttl/ex06_ttl_restart_consistency.out
b/regression-test/data/cloud_p0/cache/ttl/ex06_ttl_restart_consistency.out
index c9f49789b71..9f8256b1c0e 100644
--- a/regression-test/data/cloud_p0/cache/ttl/ex06_ttl_restart_consistency.out
+++ b/regression-test/data/cloud_p0/cache/ttl/ex06_ttl_restart_consistency.out
@@ -2,11 +2,5 @@
-- !ex06_preheat --
300
--- !ex06_create_table_1 --
-ex06_ttl_restart_consistency CREATE TABLE `ex06_ttl_restart_consistency`
(\n `k1` bigint NOT NULL,\n `c1` varchar(64) NOT NULL\n)
ENGINE=OLAP\nDUPLICATE KEY(`k1`)\nDISTRIBUTED BY HASH(`k1`) BUCKETS
8\nPROPERTIES (\n"file_cache_ttl_seconds" = "0",\n"is_being_synced" =
"false",\n"storage_medium" = "hdd",\n"storage_format" =
"V2",\n"inverted_index_storage_format" = "V3",\n"light_schema_change" =
"true",\n"disable_auto_compaction" = "true",\n"group_commit_interval_ms" =
"10000",\n"group [...]
-
--- !ex06_create_table_2 --
-ex06_ttl_restart_consistency CREATE TABLE `ex06_ttl_restart_consistency`
(\n `k1` bigint NOT NULL,\n `c1` varchar(64) NOT NULL\n)
ENGINE=OLAP\nDUPLICATE KEY(`k1`)\nDISTRIBUTED BY HASH(`k1`) BUCKETS
8\nPROPERTIES (\n"file_cache_ttl_seconds" = "0",\n"is_being_synced" =
"false",\n"storage_medium" = "hdd",\n"storage_format" =
"V2",\n"inverted_index_storage_format" = "V3",\n"light_schema_change" =
"true",\n"disable_auto_compaction" = "true",\n"group_commit_interval_ms" =
"10000",\n"group [...]
-
-- !ex06_after_restart --
-300
\ No newline at end of file
+300
diff --git
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/SuiteCluster.groovy
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/SuiteCluster.groovy
index 899ba118f5e..689547f78fe 100644
---
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/SuiteCluster.groovy
+++
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/SuiteCluster.groovy
@@ -621,7 +621,7 @@ class SuiteCluster {
}
List<Integer> addFrontend(int num, boolean followerMode=false) throws
Exception {
- def result = add(0, num, '', false, null)
+ def result = add(num, 0, '', followerMode, null)
return result.first
}
diff --git
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/event/TeamcityEventListener.groovy
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/event/TeamcityEventListener.groovy
index 98768da1348..35fb5e5f2f9 100644
---
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/event/TeamcityEventListener.groovy
+++
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/suite/event/TeamcityEventListener.groovy
@@ -50,10 +50,12 @@ class TeamcityEventListener implements EventListener {
@Override
void onSuiteFailed(SuiteContext suiteContext, Throwable t) {
- def (Integer errorLine, String errorInfo) =
LoggerUtils.getErrorInfo(t, suiteContext.scriptContext.file)
+ File errorFile = LoggerUtils.findErrorFile(t, new
File(suiteContext.config.suitePath), suiteContext.scriptContext.file)
+ String errorFileName = new
File(suiteContext.config.suitePath).relativePath(errorFile)
+ def (Integer errorLine, String errorInfo) =
LoggerUtils.getErrorInfo(t, errorFile)
String errorMsg = errorInfo == null
- ? "Exception in ${suiteContext.scriptContext.name}:"
- : "Exception in ${suiteContext.scriptContext.name}(line
${errorLine}):\n\n${errorInfo}\n\nException:"
+ ? "Exception in ${errorFileName}:"
+ : "Exception in ${errorFileName}(line
${errorLine}):\n\n${errorInfo}\n\nException:"
def stackTrace = Throwables.getStackTraceAsString(t)
TeamcityUtils.testFailed(suiteContext, errorMsg, stackTrace)
}
diff --git
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/LoggerUtils.groovy
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/LoggerUtils.groovy
index 5635ef8b4ff..dd6fd86ec78 100644
---
a/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/LoggerUtils.groovy
+++
b/regression-test/framework/src/main/groovy/org/apache/doris/regression/util/LoggerUtils.groovy
@@ -20,6 +20,29 @@ package org.apache.doris.regression.util
import com.google.common.collect.Sets
class LoggerUtils {
+ static File findErrorFile(Throwable t, File suiteRoot, File fallbackFile) {
+ try {
+ def st = findRootGroovyStackTrace(t, Sets.newLinkedHashSet())
+ if (st.is(null) || Objects.equals(st.fileName, fallbackFile.name))
{
+ return fallbackFile
+ }
+
+ def candidates = []
+ suiteRoot.eachFileRecurse { file ->
+ if (file.isFile() && Objects.equals(file.name, st.fileName)) {
+ candidates.add(file)
+ }
+ }
+
+ candidates = candidates.findAll { file ->
+ st.lineNumber <= 0 || st.lineNumber <= file.readLines().size()
+ }
+ return candidates.size() == 1 ? candidates[0] as File :
fallbackFile
+ } catch (Exception e) {
+ return fallbackFile
+ }
+ }
+
static Tuple2<Integer, String> getErrorInfo(Throwable t, File file) {
try {
if (file.name.endsWith(".groovy")) {
@@ -70,4 +93,23 @@ class LoggerUtils {
}
return null
}
+
+ static StackTraceElement findRootGroovyStackTrace(Throwable t,
Set<Throwable> throwables) {
+ throwables.add(t)
+
+ def cause = t.getCause()
+ if (!cause.is(null) && !throwables.contains(cause)) {
+ def foundStackTrace = findRootGroovyStackTrace(cause, throwables)
+ if (!foundStackTrace.is(null)) {
+ return foundStackTrace
+ }
+ }
+
+ for (def st : t.getStackTrace()) {
+ if (st.fileName != null && st.fileName.endsWith(".groovy") &&
st.lineNumber > 0) {
+ return st
+ }
+ }
+ return null
+ }
}
diff --git
a/regression-test/suites/cloud_p0/balance/test_balance_use_compute_group_properties.groovy
b/regression-test/suites/cloud_p0/balance/test_balance_use_compute_group_properties.groovy
index 0fa780a6108..71ca7ebe5aa 100644
---
a/regression-test/suites/cloud_p0/balance/test_balance_use_compute_group_properties.groovy
+++
b/regression-test/suites/cloud_p0/balance/test_balance_use_compute_group_properties.groovy
@@ -186,6 +186,24 @@ suite('test_balance_use_compute_group_properties',
'docker') {
}
}
+ def waitUntilFirstMapKeysNotEqual = { clusterName, clusterRet ->
+ awaitUntil(20, 1) {
+ sql """ use @$clusterName """
+ sql """ select * from $table """
+ def currentGetFromFe = getTabletAndBeHostFromFe(table)
+ def keys = currentGetFromFe.keySet().toList()
+ logger.info("wait clusterName {} fe tablets {}", clusterName,
currentGetFromFe)
+ if (keys.size() < 2) {
+ return false
+ }
+ if (currentGetFromFe[keys[0]] != currentGetFromFe[keys[1]]) {
+ clusterRet[0] = currentGetFromFe
+ return true
+ }
+ return false
+ }
+ }
+
// check afterBalanceEveryClusterCache
// fe config cloud_warm_up_for_rebalance_type=sync_warmup
def global_config_cluster_ret =
afterBalanceEveryClusterCache[global_config_cluster]
@@ -201,6 +219,7 @@ suite('test_balance_use_compute_group_properties',
'docker') {
def async_warmup_cluster_ret =
afterBalanceEveryClusterCache[async_warmup_cluster]
logger.info("async_warmup_cluster_ret {}", async_warmup_cluster_ret)
// fe tablets has changed, due to task timeout
+ waitUntilFirstMapKeysNotEqual(async_warmup_cluster,
async_warmup_cluster_ret)
assertFirstMapKeys(async_warmup_cluster, async_warmup_cluster_ret,
false)
def sync_warmup_cluster_ret =
afterBalanceEveryClusterCache[sync_warmup_cluster]
diff --git
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_immediate_warmup_multi_segments.groovy
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_immediate_warmup_multi_segments.groovy
index 3fe09c7e882..6bd76d929b6 100644
---
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_immediate_warmup_multi_segments.groovy
+++
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_immediate_warmup_multi_segments.groovy
@@ -98,18 +98,19 @@ suite('test_immediate_warmup_multi_segments', 'docker') {
return getBrpcMetrics(ip, port, name)
}
- def injectS3FileReadSlow = {cluster, sleep_s ->
+ def injectS3FileReadSlow = {cluster, sleepSeconds ->
def backends = sql """SHOW BACKENDS"""
def cluster_bes = backends.findAll {
it[19].contains("""\"compute_group_name\" : \"${cluster}\"""") }
def injectName = 'S3FileReader::read_at_impl.io_slow'
+ def sleepMicros = sleepSeconds * 1000 * 1000
for (be in cluster_bes) {
def ip = be[1]
def port = be[4]
- GetDebugPoint().enableDebugPoint(ip, port as int, NodeType.BE,
injectName, [sleep:sleep_s, execute:1])
+ GetDebugPoint().enableDebugPoint(ip, port as int, NodeType.BE,
injectName, [sleep:sleepMicros, execute:1])
}
}
- def getTabletStatus = { cluster, tablet_id, rowsetIndex,
lastRowsetSegmentNum, enableAssert = false ->
+ def getRowsetSegmentNum = { cluster, tablet_id, rowsetIndex ->
def backends = sql """SHOW BACKENDS"""
def cluster_bes = backends.findAll {
it[19].contains("""\"compute_group_name\" : \"${cluster}\"""") }
assert cluster_bes.size() > 0, "No backend found for cluster
${cluster}"
@@ -146,11 +147,7 @@ suite('test_immediate_warmup_multi_segments', 'docker') {
int end_index = rowset.indexOf("DATA")
def segmentNumStr = rowset.substring(start_index + 1, end_index).trim()
logger.info("segmentNumStr: ${segmentNumStr}")
- if (enableAssert) {
- assertEquals(lastRowsetSegmentNum, Integer.parseInt(segmentNumStr))
- } else {
- return lastRowsetSegmentNum == Integer.parseInt(segmentNumStr);
- }
+ return Integer.parseInt(segmentNumStr)
}
docker(options) {
@@ -217,8 +214,9 @@ suite('test_immediate_warmup_multi_segments', 'docker') {
sql "sync"
def rowCount1 = sql """ select count() from ${testTable}; """
logger.info("rowCount1: ${rowCount1}")
- // check generate 3 segments
- getTabletStatus(clusterName1, tablet_id, 2, 3, true)
+ // Check it generates multiple segments. The exact number depends
on flush timing.
+ def segmentNum = getRowsetSegmentNum(clusterName1, tablet_id, 2)
+ assertTrue(segmentNum > 1, "Expect multi-segment rowset, actual
segment num: ${segmentNum}")
// switch to read cluster, trigger a sync rowset
injectS3FileReadSlow(clusterName2, 10)
@@ -229,11 +227,13 @@ suite('test_immediate_warmup_multi_segments', 'docker') {
}
sleep(1000)
assertEquals(1, getBrpcMetricsByCluster(clusterName2,
"file_cache_warm_up_rowset_triggered_by_sync_rowset_num"))
- assertEquals(2, getBrpcMetricsByCluster(clusterName2,
"file_cache_warm_up_segment_complete_num"))
+ def segmentCompleteNum = getBrpcMetricsByCluster(clusterName2,
"file_cache_warm_up_segment_complete_num")
+ assertTrue(segmentNum > segmentCompleteNum,
+ "Expect segment num: ${segmentNum} not warm up yet, but
actual file_cache_warm_up_segment_complete_num: ${segmentCompleteNum}")
assertEquals(0, getBrpcMetricsByCluster(clusterName2,
"file_cache_warm_up_rowset_complete_num"))
future.get()
- assertEquals(3, getBrpcMetricsByCluster(clusterName2,
"file_cache_warm_up_segment_complete_num"))
+ assertEquals(segmentNum, getBrpcMetricsByCluster(clusterName2,
"file_cache_warm_up_segment_complete_num"))
assertEquals(1, getBrpcMetricsByCluster(clusterName2,
"file_cache_warm_up_rowset_complete_num"))
} finally {
GetDebugPoint().clearDebugPointsForAllBEs()
diff --git
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_cancel_expired.groovy
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_cancel_expired.groovy
index 31c28cbf2b7..6114892039a 100644
---
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_cancel_expired.groovy
+++
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_cancel_expired.groovy
@@ -153,10 +153,19 @@ suite('test_warm_up_cluster_event_cancel_expired',
'docker') {
logger.info("waiting for FE to expire+remove cancelled job")
def removed = false
for (int i = 0; i < 60; i++) {
- def rows = sql """SHOW WARM UP JOB WHERE ID = ${jobId}"""
- if (rows.isEmpty()) {
+ try {
+ def rows = sql """SHOW WARM UP JOB WHERE ID = ${jobId}"""
+ if (rows.isEmpty()) {
+ removed = true
+ logger.info("job ${jobId} removed from FE after ${i}s")
+ break
+ }
+ } catch (Exception e) {
+ if (!e.getMessage().contains("cloud warm up with job ${jobId}
does not exist")) {
+ throw e
+ }
removed = true
- logger.info("job ${jobId} removed from FE after ${i}s")
+ logger.info("job ${jobId} removed from FE after ${i}s:
${e.getMessage()}")
break
}
sleep(1000)
diff --git
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_compaction_sync_wait.groovy
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_compaction_sync_wait.groovy
index 35266cf402f..6529f8b1901 100644
---
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_compaction_sync_wait.groovy
+++
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_compaction_sync_wait.groovy
@@ -304,8 +304,9 @@ suite('test_warm_up_cluster_event_compaction_sync_wait',
'docker') {
assertTrue(num_submitted >= 8)
assertEquals(num_finished, num_submitted)
- // inject slow io, which should cause the warmup takes longger than 10s
- GetDebugPoint().enableDebugPoint(be.ip, be.http_port as int,
NodeType.BE, "S3FileReader::read_at_impl.io_slow", [sleep:10])
+ // The debug point sleep unit is microseconds. Sleep 10s to keep
warmup in progress.
+ GetDebugPoint().enableDebugPoint(be.ip, be.http_port as int,
NodeType.BE,
+ "S3FileReader::read_at_impl.io_slow", [sleep:10000000])
// trigger and wait compaction async
def future = thread {
diff --git
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_compaction_sync_wait_timeout.groovy
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_compaction_sync_wait_timeout.groovy
index 8a096603120..92a0350c53d 100644
---
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_compaction_sync_wait_timeout.groovy
+++
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_compaction_sync_wait_timeout.groovy
@@ -304,8 +304,8 @@
suite('test_warm_up_cluster_event_compaction_sync_wait_timeout', 'docker') {
assertTrue(num_submitted >= 8)
assertEquals(num_finished, num_submitted)
- // inject slow io, which should cause the warmup takes longger than 10s
- GetDebugPoint().enableDebugPoint(be.ip, be.http_port as int,
NodeType.BE, "S3FileReader::read_at_impl.io_slow", [sleep:20])
+ // inject 20s slow io, which should cause the warmup takes longer than
10s
+ GetDebugPoint().enableDebugPoint(be.ip, be.http_port as int,
NodeType.BE, "S3FileReader::read_at_impl.io_slow", [sleep:20000000])
// trigger and wait compaction async
def future = thread {
diff --git
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_schema_change.groovy
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_schema_change.groovy
index 13a5435b104..4ce2669d633 100644
---
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_schema_change.groovy
+++
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_event_schema_change.groovy
@@ -103,6 +103,77 @@ suite('test_warm_up_cluster_event_schema_change',
'docker') {
}
}
+ def getWarmUpMetrics = { cluster ->
+ def backends = sql """SHOW BACKENDS"""
+ def cluster_bes = backends.findAll {
it[19].contains("""\"compute_group_name\" : \"${cluster}\"""") }
+ assertTrue(cluster_bes.size() > 0, "No backend found for cluster
${cluster}")
+
+ def metrics = [
+ downloadSubmitted: 0L,
+ downloadFinished: 0L,
+ downloadFailed: 0L,
+ submittedSegment: 0L,
+ finishedSegment: 0L,
+ failedSegment: 0L,
+ submittedIndex: 0L,
+ finishedIndex: 0L,
+ failedIndex: 0L,
+ ]
+ for (be in cluster_bes) {
+ def ip = be[1]
+ def port = be[5]
+ def beMetrics = [
+ downloadSubmitted: getBrpcMetrics(ip, port,
"file_cache_download_submitted_num"),
+ downloadFinished: getBrpcMetrics(ip, port,
"file_cache_download_finished_num"),
+ downloadFailed: getBrpcMetrics(ip, port,
"file_cache_download_failed_num"),
+ submittedSegment: getBrpcMetrics(ip, port,
"file_cache_event_driven_warm_up_submitted_segment_num"),
+ finishedSegment: getBrpcMetrics(ip, port,
"file_cache_event_driven_warm_up_finished_segment_num"),
+ failedSegment: getBrpcMetrics(ip, port,
"file_cache_event_driven_warm_up_failed_segment_num"),
+ submittedIndex: getBrpcMetrics(ip, port,
"file_cache_event_driven_warm_up_submitted_index_num"),
+ finishedIndex: getBrpcMetrics(ip, port,
"file_cache_event_driven_warm_up_finished_index_num"),
+ failedIndex: getBrpcMetrics(ip, port,
"file_cache_event_driven_warm_up_failed_index_num"),
+ ]
+ logger.info("${cluster} be ${ip}:${port}, warmup metrics
${beMetrics}")
+ beMetrics.each { key, value ->
+ metrics[key] += value
+ }
+ }
+ logger.info("${cluster} warmup metrics summary ${metrics}")
+ return metrics
+ }
+
+ def isWarmUpMetricsFinished = { metrics ->
+ return metrics.downloadSubmitted > 0
+ && metrics.downloadSubmitted == metrics.downloadFinished +
metrics.downloadFailed
+ && metrics.submittedSegment > 0
+ && metrics.submittedSegment == metrics.finishedSegment +
metrics.failedSegment
+ && metrics.submittedIndex == metrics.finishedIndex +
metrics.failedIndex
+ && metrics.downloadFailed == 0
+ && metrics.failedSegment == 0
+ && metrics.failedIndex == 0
+ }
+
+ def waitWarmUpFinished = { cluster, timeoutMs = 120000, intervalMs = 2000,
stableRounds = 3 ->
+ def lastMetrics = null
+ int stableCount = 0
+ for (int elapsed = 0; elapsed <= timeoutMs; elapsed += intervalMs) {
+ def metrics = getWarmUpMetrics(cluster)
+ if (isWarmUpMetricsFinished(metrics)) {
+ stableCount = (lastMetrics != null && metrics == lastMetrics)
? stableCount + 1 : 1
+ if (stableCount >= stableRounds) {
+ logger.info("${cluster} warmup finished and stable after
${elapsed} ms, metrics=${metrics}")
+ return
+ }
+ } else {
+ stableCount = 0
+ }
+ lastMetrics = metrics
+ sleep(intervalMs)
+ }
+ def finalMetrics = getWarmUpMetrics(cluster)
+ throw new RuntimeException("${cluster} warmup did not finish in
${timeoutMs} ms, metrics=${finalMetrics}")
+ }
+
def getTTLCacheSize = { ip, port ->
return getBrpcMetrics(ip, port, "ttl_cache_size")
}
@@ -246,8 +317,8 @@ suite('test_warm_up_cluster_event_schema_change', 'docker')
{
sleep(20000)
run_schema_change();
- sleep(15000)
+ waitWarmUpFinished(clusterName2)
logFileCacheDownloadMetrics(clusterName2)
logWarmUpRowsetMetrics(clusterName2)
checkTTLCacheSizeSumEqual(clusterName1, clusterName2)
diff --git
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_periodic_restart_master_fe.groovy
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_periodic_restart_master_fe.groovy
index 9d40d4efe8b..837c9ef559a 100644
---
a/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_periodic_restart_master_fe.groovy
+++
b/regression-test/suites/cloud_p0/cache/multi_cluster/warm_up/cluster/test_warm_up_cluster_periodic_restart_master_fe.groovy
@@ -84,6 +84,39 @@ suite('test_warm_up_cluster_periodic_restart_master_fe',
'docker') {
}
}
+ def logFileCacheQueueMetrics = { label, cluster ->
+ def metricNames = [
+ "file_cache_cache_size",
+ "file_cache_ttl_cache_size",
+ "file_cache_ttl_cache_lru_queue_size",
+ "file_cache_ttl_cache_lru_queue_element_count",
+ "file_cache_normal_queue_cache_size",
+ "file_cache_normal_queue_element_count",
+ "file_cache_index_queue_cache_size",
+ "file_cache_index_queue_element_count",
+ "file_cache_disposable_queue_cache_size",
+ "file_cache_disposable_queue_element_count",
+ ]
+
+ 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 values = metricNames.collectEntries { metricName ->
+ [(metricName): getBrpcMetrics(ip, port, metricName)]
+ }
+ logger.info("${label}, ${cluster} be ${ip}:${port}, file cache
queue metrics: ${values}")
+ }
+ }
+
+ def logFileCacheQueueMetricsOnClusters = { label, clusters ->
+ for (cluster in clusters) {
+ logFileCacheQueueMetrics(label, cluster)
+ logFileCacheDownloadMetrics(cluster)
+ }
+ }
+
def logWarmUpRowsetMetrics = { cluster ->
def backends = sql """SHOW BACKENDS"""
def cluster_bes = backends.findAll {
it[19].contains("""\"compute_group_name\" : \"${cluster}\"""") }
@@ -109,6 +142,8 @@ suite('test_warm_up_cluster_periodic_restart_master_fe',
'docker') {
}
def checkTTLCacheSizeSumEqual = { cluster1, cluster2 ->
+ logFileCacheQueueMetricsOnClusters("before ttl cache size equality
check", [cluster1, cluster2])
+
def backends = sql """SHOW BACKENDS"""
def srcBes = backends.findAll {
it[19].contains("""\"compute_group_name\" : \"${cluster1}\"""") }
@@ -195,13 +230,18 @@ suite('test_warm_up_cluster_periodic_restart_master_fe',
'docker') {
// Clear hotspot statistics
sql """truncate table __internal_schema.cloud_cache_hotspot;"""
+ logFileCacheQueueMetricsOnClusters("after restart master fe, before
clear cache",
+ [clusterName1, clusterName2])
clearFileCacheOnAllBackends()
+ logFileCacheQueueMetricsOnClusters("after clear cache", [clusterName1,
clusterName2])
for (int i = 0; i < 1000; i++) {
sql """SELECT * FROM customer"""
}
+ logFileCacheQueueMetricsOnClusters("after querying source cluster",
[clusterName1, clusterName2])
sleep(5000)
+ logFileCacheQueueMetricsOnClusters("after waiting periodic warm up",
[clusterName1, clusterName2])
def hotspot = sql """select * from
__internal_schema.cloud_cache_hotspot;"""
logger.info("hotspot: {}", hotspot)
diff --git
a/regression-test/suites/cloud_p0/cache/ttl/ex06_ttl_restart_consistency.groovy
b/regression-test/suites/cloud_p0/cache/ttl/ex06_ttl_restart_consistency.groovy
index 26a6325803d..f31f6f1e336 100644
---
a/regression-test/suites/cloud_p0/cache/ttl/ex06_ttl_restart_consistency.groovy
+++
b/regression-test/suites/cloud_p0/cache/ttl/ex06_ttl_restart_consistency.groovy
@@ -80,6 +80,15 @@ suite("ex06_ttl_restart_consistency", "docker") {
assertTrue(false, "Timeout waiting type=${expectedType},
tablets=${tabletIds}")
}
+ def assertTtlDisabledInShowCreate = {
+ def rows = sql """show create table ${tableName};"""
+ assertTrue(!rows.isEmpty(), "No show create table result for
${tableName}")
+ def createTableSql = rows[0][1].toString()
+ logger.info("show create table ${tableName}:
${createTableSql}")
+ assertTrue(createTableSql.contains("\"file_cache_ttl_seconds\"
= \"0\""),
+ "file_cache_ttl_seconds is not 0 in show create table
result: ${createTableSql}")
+ }
+
def values = (0..<300).collect { i -> "(${i}, 'restart_${i}')"
}.join(",")
sql """insert into ${tableName} values ${values}"""
qt_ex06_preheat """select count(*) from ${tableName} where c1 like
'restart_%'"""
@@ -90,9 +99,9 @@ suite("ex06_ttl_restart_consistency", "docker") {
// EX-06: TTL 转换窗口内重启，重启后状态应一致收敛到 normal。
sql """alter table ${tableName} set
("file_cache_ttl_seconds"="0")"""
- qt_ex06_create_table_1 """show create table ${tableName};"""
+ assertTtlDisabledInShowCreate.call()
cluster.restartBackends()
- qt_ex06_create_table_2 """show create table ${tableName};"""
+ assertTtlDisabledInShowCreate.call()
sleep(10000)
qt_ex06_after_restart """select count(*) from ${tableName} where
c1 like 'restart_%'"""
diff --git
a/regression-test/suites/cloud_p0/cache/warm_up/test_warmup_table_docker.groovy
b/regression-test/suites/cloud_p0/cache/warm_up/test_warmup_table_docker.groovy
index 1ac63a63c26..e93560b389e 100644
---
a/regression-test/suites/cloud_p0/cache/warm_up/test_warmup_table_docker.groovy
+++
b/regression-test/suites/cloud_p0/cache/warm_up/test_warmup_table_docker.groovy
@@ -28,6 +28,9 @@ suite('test_warmup_table_docker', 'docker') {
'file_cache_enter_disk_resource_limit_mode_percent=99',
'enable_evict_file_cache_in_advance=false',
'enable_only_warm_up_idx=true',
+ 'file_cache_background_gc_interval_ms=10',
+ 'file_cache_background_monitor_interval_ms=100',
+ 'file_cache_background_block_lru_update_interval_ms=100',
]
options.cloudMode = true
options.beNum = 1
@@ -46,17 +49,17 @@ suite('test_warmup_table_docker', 'docker') {
}
}
- def clearFileCacheOnAllBackends = {
+ def getBackendsByCluster = { clusterName ->
def backends = sql """SHOW BACKENDS"""
+ return backends.findAll { it[19].contains("""\"compute_group_name\" :
\"${clusterName}\"""") }
+ }
+ def clearFileCacheOnBackends = { backends ->
for (be in backends) {
def ip = be[1]
def port = be[4]
clearFileCache(ip, port)
}
-
- // clear file cache is async, wait it done
- sleep(5000)
}
def getBrpcMetrics = {ip, port, name ->
@@ -70,6 +73,36 @@ suite('test_warmup_table_docker', 'docker') {
}
}
+ def getClusterFileCacheSize = { clusterName ->
+ def backends = getBackendsByCluster(clusterName)
+ assertTrue(backends.size() > 0, "No backend found for cluster
${clusterName}")
+
+ long sum = 0
+ for (be in backends) {
+ def ip = be[1]
+ def port = be[5]
+ def size = getBrpcMetrics(ip, port, "file_cache_cache_size")
+ sum += size
+ logger.info("BE ${ip}:${port} cluster=${clusterName}
file_cache_cache_size=${size}")
+ }
+ return sum
+ }
+
+ def waitClusterFileCacheCleared = { clusterName, timeoutMs = 60000,
intervalMs = 1000 ->
+ long start = System.currentTimeMillis()
+ long currentSize = getClusterFileCacheSize(clusterName)
+ while (currentSize != 0 && System.currentTimeMillis() - start <
timeoutMs) {
+ logger.info("Wait file cache clear for cluster ${clusterName},
current size=${currentSize}")
+ sleep(intervalMs)
+ currentSize = getClusterFileCacheSize(clusterName)
+ }
+ if (currentSize != 0) {
+ throw new RuntimeException("File cache on cluster ${clusterName}
is not cleared in ${timeoutMs} ms, " +
+ "current size=${currentSize}")
+ }
+ logger.info("File cache on cluster ${clusterName} has been cleared")
+ }
+
def updateBeConf = {cluster, key, value ->
def backends = sql """SHOW BACKENDS"""
def cluster_bes = backends.findAll {
it[19].contains("""\"compute_group_name\" : \"${cluster}\"""") }
@@ -84,12 +117,11 @@ suite('test_warmup_table_docker', 'docker') {
docker(options) {
def clusterName = "warmup_cluster"
- // Add one cluster
- cluster.addBackend(1, clusterName)
-
+ // rename default cluster， avoid add new cluster and stream load use
wrong be
+ sql """ALTER SYSTEM RENAME COMPUTE GROUP compute_cluster
${clusterName}"""
// Ensure we are in the cluster
sql """use @${clusterName}"""
-
+
try {
sql "set global enable_audit_plugin = false"
} catch (Exception e) {
@@ -142,19 +174,20 @@ suite('test_warmup_table_docker', 'docker') {
}
sql "sync"
- def backends = sql """SHOW BACKENDS"""
+ def backends = getBackendsByCluster(clusterName)
+ assertTrue(backends.size() > 0, "No backend found for cluster
${clusterName}")
def ip = backends[0][1]
def brpcPort = backends[0][5]
sleep(3000)
- def cache_size_after_load = getBrpcMetrics(ip, brpcPort,
"cache_cache_size")
+ def cache_size_after_load = getClusterFileCacheSize(clusterName)
// Clear file cache to ensure warm up actually does something
- clearFileCacheOnAllBackends()
+ clearFileCacheOnBackends(backends)
- sleep(3000)
- def cache_size_after_clear = getBrpcMetrics(ip, brpcPort,
"cache_cache_size")
- assertEquals(cache_size_after_clear, 0)
+ waitClusterFileCacheCleared(clusterName)
+ def cache_size_after_clear = getClusterFileCacheSize(clusterName)
+ assertEquals(0, cache_size_after_clear)
// Set enable_only_warm_up_idx = true
updateBeConf(clusterName, "enable_only_warm_up_idx", "true")
@@ -184,7 +217,7 @@ suite('test_warmup_table_docker', 'docker') {
assertTrue(waitJobFinished(id), "Warm up job ${id} did not finish in
time")
sleep(3000)
- def cache_size_after_warm = getBrpcMetrics(ip, brpcPort,
"cache_cache_size")
+ def cache_size_after_warm = getClusterFileCacheSize(clusterName)
logger.info("Cache size after load: ${cache_size_after_load}, after
clear: ${cache_size_after_clear}, after warm up: ${cache_size_after_warm}")
assertTrue(cache_size_after_warm < cache_size_after_load);
diff --git
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/default_vcg_auto_failover.groovy
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/default_vcg_auto_failover.groovy
index f1107d2a58e..fa7f8dfeffc 100644
---
a/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/default_vcg_auto_failover.groovy
+++
b/regression-test/suites/cloud_p0/multi_cluster/virtual_compute_group/default_vcg_auto_failover.groovy
@@ -23,8 +23,8 @@ import groovy.json.JsonOutput
// 2 create vcg, c1, c2 are sub compute group of vcg, adn c1 is active cg
// 3 set vcg as default cluster
// 4 stop 2 bes of c1
-// 5 start 2 bes of c1
-// 6 long-term stop 2 bes of c1
+// 5 start 2 bes of c1, vcg does not actively switch back and writes still go
to c2
+// 6 stop the current serving c2, writes go back to c1
suite('default_vcg_auto_failover', 'multi_cluster,docker') {
def options = new ClusterOptions()
@@ -66,11 +66,11 @@ suite('default_vcg_auto_failover', 'multi_cluster,docker') {
def clusterName1 = "newcluster1"
// add cluster newcluster1
- cluster.addBackend(2, clusterName1)
+ def cluster1BeIndexes = cluster.addBackend(2, clusterName1)
def clusterName2 = "newcluster2"
// add cluster newcluster2
- cluster.addBackend(2, clusterName2)
+ def cluster2BeIndexes = cluster.addBackend(2, clusterName2)
// add vcluster
def instance_id = "default_instance_id"
@@ -189,7 +189,7 @@ suite('default_vcg_auto_failover', 'multi_cluster,docker') {
sql """ set enable_profile = true """
- cluster.stopBackends(4, 5)
+ cluster.stopBackends(cluster1BeIndexes[0], cluster1BeIndexes[1])
def before_cluster2_be0_load_rows = get_be_metric(cluster2Ips[0],
"8040", "load_rows");
log.info("before_cluster2_be0_load_rows :
${before_cluster2_be0_load_rows}".toString())
@@ -262,7 +262,7 @@ suite('default_vcg_auto_failover', 'multi_cluster,docker') {
log.info("policy {}", vcgInShow.Policy)
assertTrue(vcgInShow.Policy.contains('"activeComputeGroup":"newcluster1","standbyComputeGroup":"newcluster2"'))
- cluster.startBackends(4, 5)
+ cluster.startBackends(cluster1BeIndexes[0], cluster1BeIndexes[1])
def before_cluster1_be0_load_rows = get_be_metric(cluster1Ips[0],
"8040", "load_rows");
log.info("before_cluster1_be0_load_rows :
${before_cluster1_be0_load_rows}".toString())
@@ -335,15 +335,15 @@ suite('default_vcg_auto_failover',
'multi_cluster,docker') {
after_cluster2_be1_flush = get_be_metric(cluster2Ips[1], "8040",
"memtable_flush_total");
log.info("after_cluster2_be1_flush :
${after_cluster2_be1_flush}".toString())
- assertTrue(before_cluster1_be0_load_rows <
after_cluster1_be0_load_rows || before_cluster1_be1_load_rows <
after_cluster1_be1_load_rows)
- assertTrue(before_cluster1_be0_flush < after_cluster1_be0_flush ||
before_cluster1_be1_flush < after_cluster1_be1_flush)
+ assertTrue(before_cluster1_be0_load_rows ==
after_cluster1_be0_load_rows)
+ assertTrue(before_cluster1_be0_flush == after_cluster1_be0_flush)
+ assertTrue(before_cluster1_be1_load_rows ==
after_cluster1_be1_load_rows)
+ assertTrue(before_cluster1_be1_flush == after_cluster1_be1_flush)
- assertTrue(before_cluster2_be0_load_rows ==
after_cluster2_be0_load_rows)
- assertTrue(before_cluster2_be0_flush == after_cluster2_be0_flush)
- assertTrue(before_cluster2_be1_load_rows ==
after_cluster2_be1_load_rows)
- assertTrue(before_cluster2_be1_flush == after_cluster2_be1_flush)
+ assertTrue(before_cluster2_be0_load_rows <
after_cluster2_be0_load_rows || before_cluster2_be1_load_rows <
after_cluster2_be1_load_rows)
+ assertTrue(before_cluster2_be0_flush < after_cluster2_be0_flush ||
before_cluster2_be1_flush < after_cluster2_be1_flush)
- addrSet = [cluster1Ips[0] + ":" + "8060", cluster1Ips[1] + ":" +
"8060"] as Set
+ addrSet = [cluster2Ips[0] + ":" + "8060", cluster2Ips[1] + ":" +
"8060"] as Set
sql """ select count(k2) AS theCount, k3 from test_all_vcluster
group by k3 order by theCount limit 1 """
if (options.connectToFollower) {
checkProfileNew.call(cluster.getOneFollowerFe(), addrSet)
@@ -351,17 +351,17 @@ suite('default_vcg_auto_failover',
'multi_cluster,docker') {
checkProfileNew.call(cluster.getMasterFe(), addrSet)
}
- cluster.stopBackends(4, 5)
+ cluster.stopBackends(cluster2BeIndexes[0], cluster2BeIndexes[1])
- before_cluster2_be0_load_rows = get_be_metric(cluster2Ips[0],
"8040", "load_rows");
- log.info("before_cluster2_be0_load_rows :
${before_cluster2_be0_load_rows}".toString())
- before_cluster2_be0_flush = get_be_metric(cluster2Ips[0], "8040",
"memtable_flush_total");
- log.info("before_cluster2_be0_flush :
${before_cluster2_be0_flush}".toString())
+ before_cluster1_be0_load_rows = get_be_metric(cluster1Ips[0],
"8040", "load_rows");
+ log.info("before_cluster1_be0_load_rows :
${before_cluster1_be0_load_rows}".toString())
+ before_cluster1_be0_flush = get_be_metric(cluster1Ips[0], "8040",
"memtable_flush_total");
+ log.info("before_cluster1_be0_flush :
${before_cluster1_be0_flush}".toString())
- before_cluster2_be1_load_rows = get_be_metric(cluster2Ips[1],
"8040", "load_rows");
- log.info("before_cluster2_be1_load_rows :
${before_cluster2_be1_load_rows}".toString())
- before_cluster2_be1_flush = get_be_metric(cluster2Ips[1], "8040",
"memtable_flush_total");
- log.info("before_cluster2_be1_flush :
${before_cluster2_be1_flush}".toString())
+ before_cluster1_be1_load_rows = get_be_metric(cluster1Ips[1],
"8040", "load_rows");
+ log.info("before_cluster1_be1_load_rows :
${before_cluster1_be1_load_rows}".toString())
+ before_cluster1_be1_flush = get_be_metric(cluster1Ips[1], "8040",
"memtable_flush_total");
+ log.info("before_cluster1_be1_flush :
${before_cluster1_be1_flush}".toString())
txnId = -1;
streamLoad {
@@ -394,20 +394,20 @@ suite('default_vcg_auto_failover',
'multi_cluster,docker') {
order_qt_all11 "SELECT count(*) FROM ${tableName}" // 20
order_qt_all12 "SELECT count(*) FROM ${tableName} where k1 <= 10"
// 11
- after_cluster2_be0_load_rows = get_be_metric(cluster2Ips[0],
"8040", "load_rows");
- log.info("after_cluster2_be0_load_rows :
${after_cluster2_be0_load_rows}".toString())
- after_cluster2_be0_flush = get_be_metric(cluster2Ips[0], "8040",
"memtable_flush_total");
- log.info("after_cluster2_be0_flush :
${after_cluster2_be0_flush}".toString())
+ after_cluster1_be0_load_rows = get_be_metric(cluster1Ips[0],
"8040", "load_rows");
+ log.info("after_cluster1_be0_load_rows :
${after_cluster1_be0_load_rows}".toString())
+ after_cluster1_be0_flush = get_be_metric(cluster1Ips[0], "8040",
"memtable_flush_total");
+ log.info("after_cluster1_be0_flush :
${after_cluster1_be0_flush}".toString())
- after_cluster2_be1_load_rows = get_be_metric(cluster2Ips[1],
"8040", "load_rows");
- log.info("after_cluster2_be1_load_rows :
${after_cluster2_be1_load_rows}".toString())
- after_cluster2_be1_flush = get_be_metric(cluster2Ips[1], "8040",
"memtable_flush_total");
- log.info("after_cluster2_be1_flush :
${after_cluster2_be1_flush}".toString())
+ after_cluster1_be1_load_rows = get_be_metric(cluster1Ips[1],
"8040", "load_rows");
+ log.info("after_cluster1_be1_load_rows :
${after_cluster1_be1_load_rows}".toString())
+ after_cluster1_be1_flush = get_be_metric(cluster1Ips[1], "8040",
"memtable_flush_total");
+ log.info("after_cluster1_be1_flush :
${after_cluster1_be1_flush}".toString())
- assertTrue(before_cluster2_be0_load_rows <
after_cluster2_be0_load_rows || before_cluster2_be1_load_rows <
after_cluster2_be1_load_rows)
- assertTrue(before_cluster2_be0_flush < after_cluster2_be0_flush ||
before_cluster2_be1_flush < after_cluster2_be1_flush)
+ assertTrue(before_cluster1_be0_load_rows <
after_cluster1_be0_load_rows || before_cluster1_be1_load_rows <
after_cluster1_be1_load_rows)
+ assertTrue(before_cluster1_be0_flush < after_cluster1_be0_flush ||
before_cluster1_be1_flush < after_cluster1_be1_flush)
- addrSet = [cluster2Ips[0] + ":" + "8060", cluster2Ips[1] + ":" +
"8060"] as Set
+ addrSet = [cluster1Ips[0] + ":" + "8060", cluster1Ips[1] + ":" +
"8060"] as Set
sql """ select count(k2) AS theCount, k3 from test_all_vcluster
group by k3 order by theCount limit 1 """
if (options.connectToFollower) {
checkProfileNew.call(cluster.getOneFollowerFe(), addrSet)
@@ -425,7 +425,7 @@ suite('default_vcg_auto_failover', 'multi_cluster,docker') {
log.info("show compute group {}", showComputeGroup)
vcgInShow = showComputeGroup.find { it.Name == normalVclusterName }
assertNotNull(vcgInShow)
-
assertTrue(vcgInShow.Policy.contains('"activeComputeGroup":"newcluster2","standbyComputeGroup":"newcluster1"'))
+
assertTrue(vcgInShow.Policy.contains('"activeComputeGroup":"newcluster1","standbyComputeGroup":"newcluster2"'))
}
// connect to follower, run again
options.connectToFollower = true
diff --git
a/regression-test/suites/cloud_p0/tablets/test_clean_tablet_when_drop_force_table.groovy
b/regression-test/suites/cloud_p0/tablets/test_clean_tablet_when_drop_force_table.groovy
index 5ee69f9ee47..7ff1e3a5c2b 100644
---
a/regression-test/suites/cloud_p0/tablets/test_clean_tablet_when_drop_force_table.groovy
+++
b/regression-test/suites/cloud_p0/tablets/test_clean_tablet_when_drop_force_table.groovy
@@ -40,10 +40,17 @@ suite('test_clean_tablet_when_drop_force_table', 'docker') {
'write_buffer_size=10240',
'write_buffer_size_for_agg=10240',
'sys_log_verbose_modules=task_worker_pool',
+ 'file_cache_background_gc_interval_ms=10',
+ 'file_cache_remove_block_qps_limit=10000',
"enable_packed_file=${enablePackedFile}",
'enable_packed_file=false',
'disable_auto_compaction=true',
]
+ options.recycleConfigs += [
+ 'recycler_sleep_before_scheduling_seconds=0',
+ 'recycle_interval_seconds=1',
+ 'retention_seconds=0',
+ ]
options.setFeNum(3)
options.setBeNum(3)
options.cloudMode = true
@@ -80,6 +87,54 @@ suite('test_clean_tablet_when_drop_force_table', 'docker') {
assertTrue(false, "Tablet cache info still exists for tablet ids
${stillPresent}")
}
}
+
+ def collectCacheSubDirs = { File dir ->
+ def subDirs = []
+ def collectDirs
+ collectDirs = { File currentDir ->
+ if (currentDir.exists()) {
+ currentDir.eachDir { subDir ->
+ subDirs << subDir.name
+ collectDirs(subDir)
+ }
+ }
+ }
+ collectDirs(dir)
+ subDirs
+ }
+
+ def getRemainingCacheDirs = { Map beforeGetFromFe, Map mergedCacheDir ->
+ def remaining = []
+ beforeGetFromFe.each {
+ def backendId = it.Value[0]
+ def backendHost = it.Value[1]
+ def be = cluster.getBeByBackendId(backendId.toLong())
+ def dataPath = new File("${be.path}/storage/file_cache")
+ def subDirs = collectCacheSubDirs.call(dataPath)
+ def cacheDir = mergedCacheDir[backendHost] ?: []
+ def matched = cacheDir.findAll { hashFile ->
+ subDirs.any { subDir -> subDir.startsWith(hashFile) }
+ }
+ if (!matched.isEmpty()) {
+ remaining << [backendHost: backendHost, hashFiles: matched,
subDirs: subDirs]
+ }
+ logger.info("BE {} file_cache subdirs: {}", backendHost, subDirs)
+ }
+ remaining
+ }
+
+ def waitForCacheDirsCleaned = { Map beforeGetFromFe, Map mergedCacheDir,
long timeoutMs = 90000L, long intervalMs = 1000L ->
+ long start = System.currentTimeMillis()
+ def remaining = []
+ while (System.currentTimeMillis() - start < timeoutMs) {
+ remaining = getRemainingCacheDirs.call(beforeGetFromFe,
mergedCacheDir)
+ if (remaining.isEmpty()) {
+ return
+ }
+ sleep(intervalMs)
+ }
+ assertTrue(false, "File cache directories were not cleaned before
timeout: ${remaining}")
+ }
def testCase = { tableName, waitTime, useDp=false->
def ms = cluster.getAllMetaservices().get(0)
@@ -202,38 +257,8 @@ suite('test_clean_tablet_when_drop_force_table', 'docker')
{
futrue.get()
}
- sleep(25 * 1000)
-
- // check cache file has been deleted
- beforeGetFromFe.each {
- def tabletId = it.Key
- def backendId = it.Value[0]
- def backendHost = it.Value[1]
- def be = cluster.getBeByBackendId(backendId.toLong())
- def dataPath = new File("${be.path}/storage/file_cache")
- def subDirs = []
-
- def collectDirs
- collectDirs = { File dir ->
- if (dir.exists()) {
- dir.eachDir { subDir ->
- subDirs << subDir.name
- collectDirs(subDir)
- }
- }
- }
-
- collectDirs(dataPath)
- logger.info("BE {} file_cache subdirs: {}", backendHost, subDirs)
- def cacheDir = mergedCacheDir[backendHost]
-
- // add check
- cacheDir.each { hashFile ->
- assertFalse(subDirs.any { subDir ->
subDir.startsWith(hashFile) },
- "Found unexpected cache file pattern ${hashFile} in BE
${backendHost}'s file_cache directory. " +
- "Matching subdir found in: ${subDirs}")
- }
- }
+ // Wait until async file-cache GC removes the physical cache
directories.
+ waitForCacheDirsCleaned.call(beforeGetFromFe, mergedCacheDir, 90000L,
1000L)
String beLogPath = cluster.getBeByIndex(1).getLogFilePath()
checkBeLog(beLogPath)
diff --git
a/regression-test/suites/compaction/test_filecache_compaction_and_read_stale_cloud_docker.groovy
b/regression-test/suites/compaction/test_filecache_compaction_and_read_stale_cloud_docker.groovy
index c4dc38f0cd9..97d8d7dfae9 100644
---
a/regression-test/suites/compaction/test_filecache_compaction_and_read_stale_cloud_docker.groovy
+++
b/regression-test/suites/compaction/test_filecache_compaction_and_read_stale_cloud_docker.groovy
@@ -356,26 +356,66 @@
suite("test_filecache_compaction_and_read_stale_cloud_docker", "docker") {
tablet_status = getTabletStatus(tablet)
def final_rowsets = tablet_status["rowsets"]
- // sleep for vacuum_stale_rowsets_interval_s=10 seconds to wait
for unused rowsets are deleted
- sleep(21000)
-
def be_host = backendId_to_backendIP[tablet.BackendId]
def be_http_port = backendId_to_backendHttpPort[tablet.BackendId]
- for (int i = 0; i < all_history_stale_rowsets.size(); i++) {
- def rowsetStr = all_history_stale_rowsets[i]
- // [12-12] 1 DATA NONOVERLAPPING
02000000000000124843c92c13625daa8296c20957119893 1011.00 B
- def start_version = rowsetStr.split(" ")[0].replace('[',
'').replace(']', '').split("-")[0].toInteger()
- def end_version = rowsetStr.split(" ")[0].replace('[',
'').replace(']', '').split("-")[1].toInteger()
- def rowset_id = rowsetStr.split(" ")[4]
- if (start_version == 0 || start_version != end_version) {
- continue
+ def be_brpc_port = backendId_to_backendBrpcPort[tablet.BackendId]
+ def listHistoryStaleRowsetCache = {
+ def cachedRowsets = []
+ int i = 0
+ for (def rowsetStr : all_history_stale_rowsets) {
+ // [12-12] 1 DATA NONOVERLAPPING
02000000000000124843c92c13625daa8296c20957119893 1011.00 B
+ def start_version = rowsetStr.split(" ")[0].replace('[',
'').replace(']', '').split("-")[0].toInteger()
+ def end_version = rowsetStr.split(" ")[0].replace('[',
'').replace(']', '').split("-")[1].toInteger()
+ def rowset_id = rowsetStr.split(" ")[4]
+ if (start_version != 0 && start_version == end_version) {
+ def data =
Http.GET("http://${be_host}:${be_http_port}/api/file_cache?op=list_cache&value=${rowset_id}_0.dat";,
true)
+ if (data.size() > 0) {
+ cachedRowsets.add([
+ index: i,
+ version: "${start_version}-${end_version}",
+ rowset_id: rowset_id,
+ cache_files: data
+ ])
+ }
+ }
+ i++
+ }
+ return cachedRowsets
+ }
+ def getBeVars = {
+ def (code, out, err) = curl("GET",
"http://${be_host}:${be_brpc_port}/vars";)
+ logger.info("get be vars: code=${code}, err=${err}")
+ if (code != 0) {
+ return "failed to get /vars, code=${code}, err=${err}"
+ }
+ return out.readLines().findAll {
+ it.contains("file_cache") || it.contains("unused_rowsets")
+ }.join("\n")
+ }
+ def getUnusedRowsetsCount = {
+ def (code, out, err) = curl("GET",
"http://${be_host}:${be_brpc_port}/vars/unused_rowsets_count";)
+ logger.info("unused_rowsets_count: code=${code}, out=${out},
err=${err}")
+ if (code != 0) {
+ return -1
}
+ return out.trim().split(":")[1].trim().toInteger()
+ }
- logger.info("rowset ${i}, start: ${start_version}, end:
${end_version}, id: ${rowset_id}")
- def data =
Http.GET("http://${be_host}:${be_http_port}/api/file_cache?op=list_cache&value=${rowset_id}_0.dat";,
true)
- logger.info("file cache data: ${data}")
- assertTrue(data.size() == 0)
+ def staleRowsetCache = []
+ for (int i = 0; i < 100; i++) {
+ staleRowsetCache = listHistoryStaleRowsetCache()
+ if (staleRowsetCache.isEmpty()) {
+ break
+ }
+ logger.info("stale rowset file cache is not recycled yet,
try=${i}, cached=${staleRowsetCache}")
+ sleep(500)
+ }
+ if (!staleRowsetCache.isEmpty()) {
+ logger.info("final stale rowset file cache remains:
${staleRowsetCache}")
+ logger.info("final unused_rowsets_count:
${getUnusedRowsetsCount()}")
+ logger.info("final file cache vars:\n${getBeVars()}")
}
+ assertTrue(staleRowsetCache.isEmpty(), "stale rowset file cache is
not recycled: ${staleRowsetCache}")
for (int i = 0; i < final_rowsets.size(); i++) {
def rowsetStr = final_rowsets[i]
@@ -392,10 +432,7 @@
suite("test_filecache_compaction_and_read_stale_cloud_docker", "docker") {
assertTrue(data.size() > 0)
}
- def (code_0, out_0, err_0) = curl("GET",
"http://${be_host}:${backendId_to_backendBrpcPort[tablet.BackendId]}/vars/unused_rowsets_count";)
- logger.info("out_0: ${out_0}")
- def unusedRowsetsCount =
out_0.trim().split(":")[1].trim().toInteger()
- assertEquals(0, unusedRowsetsCount)
+ assertEquals(0, getUnusedRowsetsCount())
} finally {
GetDebugPoint().clearDebugPointsForAllBEs()
diff --git
a/regression-test/suites/compaction/test_filecache_compaction_multisegments_and_read_stale_cloud_docker.groovy
b/regression-test/suites/compaction/test_filecache_compaction_multisegments_and_read_stale_cloud_docker.groovy
index 63f97bde051..956a70c382e 100644
---
a/regression-test/suites/compaction/test_filecache_compaction_multisegments_and_read_stale_cloud_docker.groovy
+++
b/regression-test/suites/compaction/test_filecache_compaction_multisegments_and_read_stale_cloud_docker.groovy
@@ -85,6 +85,27 @@
suite("test_filecache_compaction_multisegments_and_read_stale_cloud_docker", "do
}
}
+ def checkTabletSegmentNumAtLeast = { tablet, rowsetIndex, minSegmentNum,
outputRowsets = null ->
+ String compactionUrl = tablet["CompactionStatus"]
+ def (code, out, err) = curl("GET", compactionUrl)
+ logger.info("Show tablets status: code=" + code + ", out=" + out + ",
err=" + err)
+ assertEquals(code, 0)
+ def tabletJson = parseJson(out.trim())
+ assert tabletJson.rowsets instanceof List
+ if (outputRowsets != null) {
+ outputRowsets.addAll(tabletJson.rowsets)
+ }
+
+ assertTrue(tabletJson.rowsets.size() >= rowsetIndex)
+ def rowset = tabletJson.rowsets.get(rowsetIndex - 1)
+ logger.info("rowset: ${rowset}")
+ int start_index = rowset.indexOf("]")
+ int end_index = rowset.indexOf("DATA")
+ def segmentNum = rowset.substring(start_index + 1,
end_index).trim().toInteger()
+ logger.info("segmentNum: ${segmentNum}")
+ assertTrue(segmentNum >= minSegmentNum)
+ }
+
def waitForCompaction = { tablet ->
String tablet_id = tablet.TabletId
String trigger_backend_id = tablet.BackendId
@@ -240,8 +261,8 @@
suite("test_filecache_compaction_multisegments_and_read_stale_cloud_docker", "do
sql "sync"
def rowCount1 = sql """ select count() from ${testTable}; """
logger.info("rowCount1: ${rowCount1}")
- // check generate 3 segments
- getTabletStatus(tablet, 2, 3, true, all_history_stale_rowsets)
+ // check generate multiple segments
+ checkTabletSegmentNumAtLeast(tablet, 2, 2,
all_history_stale_rowsets)
// trigger compaction
GetDebugPoint().enableDebugPointForAllBEs("CloudSizeBasedCumulativeCompactionPolicy::pick_input_rowsets.set_input_rowsets",
@@ -251,7 +272,7 @@
suite("test_filecache_compaction_multisegments_and_read_stale_cloud_docker", "do
assertEquals(code, 0)
def compactJson = parseJson(out.trim())
logger.info("compact json: " + compactJson)
- // check generate 1 segments
+ // check compaction output has one segment
for (int i = 0; i < 20; i++) {
if (getTabletStatus(tablet, 2, 1, false,
all_history_stale_rowsets)) {
break
@@ -282,8 +303,8 @@
suite("test_filecache_compaction_multisegments_and_read_stale_cloud_docker", "do
sql "sync"
def rowCount2 = sql """ select count() from ${testTable}; """
logger.info("rowCount2: ${rowCount2}")
- // check generate 3 segments
- getTabletStatus(tablet, 3, 6, false, all_history_stale_rowsets)
+ // check generate multiple segments
+ checkTabletSegmentNumAtLeast(tablet, 3, 2,
all_history_stale_rowsets)
def local_dm = getLocalDeleteBitmapStatus(tablet)
logger.info("local delete bitmap 1: " + local_dm)
@@ -296,7 +317,7 @@
suite("test_filecache_compaction_multisegments_and_read_stale_cloud_docker", "do
compactJson = parseJson(out.trim())
logger.info("compact json: " + compactJson)
waitForCompaction(tablet)
- // check generate 1 segments
+ // check compaction output has one segment
for (int i = 0; i < 20; i++) {
if (getTabletStatus(tablet, 3, 1, false,
all_history_stale_rowsets)) {
break
@@ -311,41 +332,89 @@
suite("test_filecache_compaction_multisegments_and_read_stale_cloud_docker", "do
logger.info("local delete bitmap 2: " + local_dm)
assertEquals(1, local_dm["delete_bitmap_count"])
-
- // sleep for vacuum_stale_rowsets_interval_s=10 seconds to wait
for unused rowsets are deleted
- sleep(21000)
-
def be_host = backendId_to_backendIP[tablet.BackendId]
def be_http_port = backendId_to_backendHttpPort[tablet.BackendId]
- logger.info("be_host: ${be_host}, be_http_port: ${be_http_port},
BrpcPort: ${backendId_to_backendBrpcPort[tablet.BackendId]}")
-
- for (int i = 0; i < all_history_stale_rowsets.size(); i++) {
- def rowsetStr = all_history_stale_rowsets[i]
- // [12-12] 1 DATA NONOVERLAPPING
02000000000000124843c92c13625daa8296c20957119893 1011.00 B
- def start_version = rowsetStr.split(" ")[0].replace('[',
'').replace(']', '').split("-")[0].toInteger()
- def end_version = rowsetStr.split(" ")[0].replace('[',
'').replace(']', '').split("-")[1].toInteger()
- def rowset_id = rowsetStr.split(" ")[4]
- if (start_version == 0) {
- continue
- }
+ def be_brpc_port = backendId_to_backendBrpcPort[tablet.BackendId]
+ logger.info("be_host: ${be_host}, be_http_port: ${be_http_port},
BrpcPort: ${be_brpc_port}")
+ def parseHistoryRowset = { int index, String rowsetStr ->
int start_index = rowsetStr.indexOf("]")
int end_index = rowsetStr.indexOf("DATA")
- def segmentNum = rowsetStr.substring(start_index + 1,
end_index).trim().toInteger()
-
- logger.info("rowset ${i}, start: ${start_version}, end:
${end_version}, id: ${rowset_id}, segment: ${segmentNum}")
- def data =
Http.GET("http://${be_host}:${be_http_port}/api/file_cache?op=list_cache&value=${rowset_id}_0.dat";,
true)
- logger.info("file cache data: ${data}")
- if (segmentNum <= 1) {
- assertTrue(data.size() > 0)
- } else {
- assertTrue(data.size() == 0)
+ return [
+ index: index,
+ version: rowsetStr.split(" ")[0].replace('[',
'').replace(']', ''),
+ start_version: rowsetStr.split(" ")[0].replace('[',
'').replace(']', '').split("-")[0].toInteger(),
+ end_version: rowsetStr.split(" ")[0].replace('[',
'').replace(']', '').split("-")[1].toInteger(),
+ rowset_id: rowsetStr.split(" ")[4],
+ segment_num: rowsetStr.substring(start_index + 1,
end_index).trim().toInteger(),
+ rowset: rowsetStr
+ ]
+ }
+ def listUnexpectedHistoryRowsetCache = {
+ def unexpected = [
+ cached_multi_segment_stale_rowsets: [],
+ missing_single_segment_stale_rowsets: []
+ ]
+ int i = 0
+ for (def rowsetStr : all_history_stale_rowsets) {
+ def rowset = parseHistoryRowset(i, rowsetStr)
+ if (rowset.start_version != 0 && rowset.start_version ==
rowset.end_version) {
+ def data =
Http.GET("http://${be_host}:${be_http_port}/api/file_cache?op=list_cache&value=${rowset.rowset_id}_0.dat";,
true)
+ if (rowset.segment_num > 1 && data.size() > 0) {
+ rowset.cache_files = data
+
unexpected.cached_multi_segment_stale_rowsets.add(rowset)
+ } else if (rowset.segment_num <= 1 && data.size() ==
0) {
+
unexpected.missing_single_segment_stale_rowsets.add(rowset)
+ }
+ }
+ i++
}
+ return unexpected
+ }
+ def getBeVars = {
+ def (code_0, out_0, err_0) = curl("GET",
"http://${be_host}:${be_brpc_port}/vars";)
+ logger.info("get be vars: code=${code_0}, err=${err_0}")
+ if (code_0 != 0) {
+ return "failed to get /vars, code=${code_0}, err=${err_0}"
+ }
+ return out_0.readLines().findAll {
+ it.contains("file_cache") || it.contains("unused_rowsets")
+ }.join("\n")
+ }
+ def getUnusedRowsetsCount = {
+ def (code_0, out_0, err_0) = curl("GET",
"http://${be_host}:${be_brpc_port}/vars/unused_rowsets_count";)
+ logger.info("unused_rowsets_count: code=${code_0},
out=${out_0}, err=${err_0}")
+ if (code_0 != 0) {
+ return -1
+ }
+ return out_0.trim().split(":")[1].trim().toInteger()
}
- def (code_0, out_0, err_0) = curl("GET",
"http://${be_host}:${backendId_to_backendBrpcPort[tablet.BackendId]}/vars/unused_rowsets_count";)
- logger.info("out_0: ${out_0}")
- def unusedRowsetsCount =
out_0.trim().split(":")[1].trim().toInteger()
+ def unexpectedRowsetCache = [:]
+ for (int i = 0; i < 100; i++) {
+ unexpectedRowsetCache = listUnexpectedHistoryRowsetCache()
+ if
(unexpectedRowsetCache.cached_multi_segment_stale_rowsets.isEmpty()
+ &&
unexpectedRowsetCache.missing_single_segment_stale_rowsets.isEmpty()) {
+ break
+ }
+ logger.info("history stale rowset file cache is not in
expected state yet, try=${i}, "
+ + "unexpected=${unexpectedRowsetCache}")
+ sleep(500)
+ }
+ if
(!unexpectedRowsetCache.cached_multi_segment_stale_rowsets.isEmpty()
+ ||
!unexpectedRowsetCache.missing_single_segment_stale_rowsets.isEmpty()) {
+ logger.info("final unexpected history stale rowset file cache:
${unexpectedRowsetCache}")
+ logger.info("final unused_rowsets_count:
${getUnusedRowsetsCount()}")
+ logger.info("final file cache vars:\n${getBeVars()}")
+ }
+
assertTrue(unexpectedRowsetCache.cached_multi_segment_stale_rowsets.isEmpty(),
+ "multi-segment stale rowset file cache is not recycled: "
+ +
"${unexpectedRowsetCache.cached_multi_segment_stale_rowsets}")
+
assertTrue(unexpectedRowsetCache.missing_single_segment_stale_rowsets.isEmpty(),
+ "single-segment stale rowset file cache is missing: "
+ +
"${unexpectedRowsetCache.missing_single_segment_stale_rowsets}")
+
+ def unusedRowsetsCount = getUnusedRowsetsCount()
assertEquals(0, unusedRowsetsCount)
} finally {
GetDebugPoint().clearDebugPointsForAllBEs()
diff --git a/regression-test/suites/demo_p0/test_lru_persist.groovy
b/regression-test/suites/demo_p0/test_lru_persist.groovy
index a8347254d88..7d8c84e6f55 100644
--- a/regression-test/suites/demo_p0/test_lru_persist.groovy
+++ b/regression-test/suites/demo_p0/test_lru_persist.groovy
@@ -79,20 +79,33 @@ suite('test_lru_persist', 'docker') {
sleep(15000);
cluster.stopBackends(1)
- def normalBefore = "md5sum
${cachePath}/lru_dump_normal.tail".execute().text.trim().split()[0]
- logger.info("normalBefore: ${normalBefore}")
+ def normalDumpBeforeRestart = new
File("${cachePath}/lru_dump_normal.tail")
+ assert normalDumpBeforeRestart.exists()
+ assert normalDumpBeforeRestart.length() > 0
+ def normalDumpLastModifiedBeforeRestart =
normalDumpBeforeRestart.lastModified()
+ logger.info("normal dump before restart size:
${normalDumpBeforeRestart.length()}, "
+ + "last modified: ${normalDumpLastModifiedBeforeRestart}")
cluster.startBackends(1)
sleep(10000);
- cluster.stopBackends(1)
+ def show_backend_after_restart = sql '''show backends'''
+ logger.info("Backend details after restart:
${show_backend_after_restart.toString()}")
+ assert show_backend_after_restart[0][9].toString() == "true"
+
+ def count_after_restart = sql '''select count(*) from tb1'''
+ assert count_after_restart[0][0] == 12
- // check md5sum again after be restart
- def normalAfter = "md5sum
${cachePath}/lru_dump_normal.tail".execute().text.trim().split()[0]
- logger.info("normalAfter: ${normalAfter}")
+ cluster.stopBackends(1)
- sql '''show data'''
- assert normalBefore == normalAfter
+ // The dump file may be rewritten during restart by
restore/replay/dump background tasks.
+ // Verify the new dump is generated instead of requiring byte-for-byte
equality.
+ def normalDump = new File("${cachePath}/lru_dump_normal.tail")
+ logger.info("normal dump after restart exists: ${normalDump.exists()},
size: ${normalDump.length()}, "
+ + "last modified: ${normalDump.lastModified()}")
+ assert normalDump.exists()
+ assert normalDump.length() > 0
+ assert normalDump.lastModified() > normalDumpLastModifiedBeforeRestart
// remove dump file
def rm_dump_ret = "rm -rf
${cachePath}/lru_dump_normal.tail".execute().text.trim()
@@ -128,4 +141,4 @@ suite('test_lru_persist', 'docker') {
sql '''select count(*) from tb1'''
}
-}
\ No newline at end of file
+}
diff --git
a/regression-test/suites/fault_injection_p0/cloud/test_cloud_concurrent_calc_dbm_task.groovy
b/regression-test/suites/fault_injection_p0/cloud/test_cloud_concurrent_calc_dbm_task.groovy
index 777795c8650..011e5705cd7 100644
---
a/regression-test/suites/fault_injection_p0/cloud/test_cloud_concurrent_calc_dbm_task.groovy
+++
b/regression-test/suites/fault_injection_p0/cloud/test_cloud_concurrent_calc_dbm_task.groovy
@@ -93,13 +93,6 @@ suite("test_cloud_concurrent_calc_dbm_task",
"multi_cluster,docker") {
sql """use @${cluster1.cluster}"""
qt_base_cluster1 "select * from ${table1} order by k1;"
- def newThreadInDocker = { Closure actionSupplier ->
- def connInfo = context.threadLocalConn.get()
- return Thread.start {
- connect(connInfo.username, connInfo.password,
connInfo.conn.getMetaData().getURL(), actionSupplier)
- }
- }
-
GetDebugPoint().enableDebugPointForAllFEs("CloudGlobalTransactionMgr.getDeleteBitmapUpdateLock.enable_spin_wait")
GetDebugPoint().enableDebugPointForAllFEs("CloudGlobalTransactionMgr.getDeleteBitmapUpdateLock.block")
@@ -108,7 +101,7 @@ suite("test_cloud_concurrent_calc_dbm_task",
"multi_cluster,docker") {
// partial update load 1 on cluster0
- def t1 = newThreadInDocker {
+ def t1 = thread("partial-update-load-1-cluster0") {
sql """use @${cluster0.cluster}"""
sql "set enable_unique_key_partial_update=true;"
sql "insert into ${table1}(k1,c1) values(1,999);"
@@ -117,7 +110,7 @@ suite("test_cloud_concurrent_calc_dbm_task",
"multi_cluster,docker") {
Thread.sleep(1000)
// partial update load 2 on cluster1
- def t2 = newThreadInDocker {
+ def t2 = thread("partial-update-load-2-cluster1") {
sql """use @${cluster1.cluster}"""
sql "set enable_unique_key_partial_update=true;"
sql "insert into ${table1}(k1,c2) values(1,888);"
@@ -127,7 +120,7 @@ suite("test_cloud_concurrent_calc_dbm_task",
"multi_cluster,docker") {
// let partial update load 1 succeed and wait for it
GetDebugPoint().disableDebugPointForAllFEs("CloudGlobalTransactionMgr.getDeleteBitmapUpdateLock.enable_spin_wait")
GetDebugPoint().disableDebugPointForAllFEs("CloudGlobalTransactionMgr.getDeleteBitmapUpdateLock.block")
- t1.join()
+ t1.get()
sql """use @${cluster0.cluster}"""
qt_upadte1_cluster0 "select * from ${table1} order by k1;"
@@ -140,7 +133,7 @@ suite("test_cloud_concurrent_calc_dbm_task",
"multi_cluster,docker") {
GetDebugPoint().disableDebugPoint(backend1.Host, backend1.HttpPort
as int, NodeType.BE,
"BaseTablet::update_delete_bitmap.after.create_transient_rs_writer")
// wait for partial update load 2 finish
- t2.join()
+ t2.get()
sql """use @${cluster1.cluster}"""
qt_upadte2_cluster1 "select * from ${table1} order by k1;"
diff --git
a/regression-test/suites/fault_injection_p0/cloud/test_cloud_full_compaction_multi_segments.groovy
b/regression-test/suites/fault_injection_p0/cloud/test_cloud_full_compaction_multi_segments.groovy
index 4ab7fd088fc..8fb00991db4 100644
---
a/regression-test/suites/fault_injection_p0/cloud/test_cloud_full_compaction_multi_segments.groovy
+++
b/regression-test/suites/fault_injection_p0/cloud/test_cloud_full_compaction_multi_segments.groovy
@@ -70,7 +70,7 @@
suite("test_cloud_full_compaction_multi_segments","multi_cluster,docker") {
}
logger.info("tablet ${tabletId} on backend ${tabletBackend.Host} with
backendId=${tabletBackend.BackendId}");
- def checkSegmentNum = { rowsetNum, lastRowsetSegmentNum ->
+ def checkSegmentNum = { rowsetNum, minLastRowsetSegmentNum ->
def tablets = sql_return_maparray """ show tablets from
${tableName}; """
logger.info("tablets: ${tablets}")
String compactionUrl = tablets[0]["CompactionStatus"]
@@ -86,7 +86,7 @@
suite("test_cloud_full_compaction_multi_segments","multi_cluster,docker") {
int end_index = rowset.indexOf("DATA")
def segmentNumStr = rowset.substring(start_index + 1,
end_index).trim()
logger.info("segmentNumStr: ${segmentNumStr}")
- assert lastRowsetSegmentNum == Integer.parseInt(segmentNumStr)
+ assert Integer.parseInt(segmentNumStr) >= minLastRowsetSegmentNum
}
def loadMultiSegmentData = { rows->
@@ -122,14 +122,7 @@
suite("test_cloud_full_compaction_multi_segments","multi_cluster,docker") {
GetDebugPoint().enableDebugPointForAllFEs("CloudGlobalTransactionMgr.getDeleteBitmapUpdateLock.block")
GetDebugPoint().enableDebugPointForAllBEs("CloudFullCompaction::modify_rowsets.block")
- def newThreadInDocker = { Closure actionSupplier ->
- def connInfo = context.threadLocalConn.get()
- return Thread.start {
- connect(connInfo.username, connInfo.password,
connInfo.conn.getMetaData().getURL(), actionSupplier)
- }
- }
-
- def t1 = newThreadInDocker {
+ def t1 = thread("load-multi-segment") {
loadMultiSegmentData(4096)
}
@@ -147,7 +140,7 @@
suite("test_cloud_full_compaction_multi_segments","multi_cluster,docker") {
// let the load publish
GetDebugPoint().disableDebugPointForAllFEs("CloudGlobalTransactionMgr.getDeleteBitmapUpdateLock.enable_spin_wait")
GetDebugPoint().disableDebugPointForAllFEs("CloudGlobalTransactionMgr.getDeleteBitmapUpdateLock.block")
- t1.join()
+ t1.get()
Thread.sleep(1000)
@@ -171,8 +164,6 @@
suite("test_cloud_full_compaction_multi_segments","multi_cluster,docker") {
qt_dup_key_count "select count() from (select k, count(*) from
${tableName} group by k having count(*) > 1) t"
qt_sql "select count() from ${tableName};"
checkSegmentNum(2, 3)
- } catch (Exception e) {
- logger.info(e.getMessage())
} finally {
GetDebugPoint().clearDebugPointsForAllBEs()
GetDebugPoint().clearDebugPointsForAllFEs()
diff --git
a/regression-test/suites/fault_injection_p0/cloud/test_cloud_mow_retry_txn_interleave.groovy
b/regression-test/suites/fault_injection_p0/cloud/test_cloud_mow_retry_txn_interleave.groovy
index c6bccbc5c19..27a6a55015f 100644
---
a/regression-test/suites/fault_injection_p0/cloud/test_cloud_mow_retry_txn_interleave.groovy
+++
b/regression-test/suites/fault_injection_p0/cloud/test_cloud_mow_retry_txn_interleave.groovy
@@ -96,13 +96,6 @@ suite("test_cloud_mow_retry_txn_interleave",
"multi_cluster,docker") {
sql """use @${cluster1.cluster}"""
qt_sql "select * from ${table1} order by k1;"
- def newThreadInDocker = { Closure actionSupplier ->
- def connInfo = context.threadLocalConn.get()
- return Thread.start {
- connect(connInfo.username, connInfo.password,
connInfo.conn.getMetaData().getURL(), actionSupplier)
- }
- }
-
// let load 1's calc task A response halt before report to FE on
cluster0
GetDebugPoint().enableDebugPoint(backend0.Host, backend0.HttpPort
as int, NodeType.BE, "CloudCalcDbmTask.handle.return.block",
[tablet_id: "${tablet_id}"])
@@ -113,7 +106,7 @@ suite("test_cloud_mow_retry_txn_interleave",
"multi_cluster,docker") {
// load 1 on cluster0
- def t1 = newThreadInDocker {
+ def t1 = thread("load-1-cluster0") {
sql """use @${cluster0.cluster}"""
sql "insert into ${table1} values(1,999,999);"
}
@@ -133,7 +126,7 @@ suite("test_cloud_mow_retry_txn_interleave",
"multi_cluster,docker") {
// let FE recieve task A's response
GetDebugPoint().disableDebugPoint(backend0.Host, backend0.HttpPort
as int, NodeType.BE, "CloudCalcDbmTask.handle.return.block")
// wait for load 1 finish
- t1.join()
+ t1.get()
// force it read delete bitmaps from MS rather than BE's cache
GetDebugPoint().enableDebugPointForAllBEs("CloudTxnDeleteBitmapCache::get_delete_bitmap.cache_miss")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
