This is an automated email from the ASF dual-hosted git repository.
liaoxin01 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 476ca43e6f2 [chore](test) Add explicit snapshot recycling checks to
cloud Docker suites (#69057)
476ca43e6f2 is described below
commit 476ca43e6f2c0d7871f6db7ebe58afbfa2574df3
Author: Yixuan Wang <[email protected]>
AuthorDate: Fri Oct 16 18:50:12 2026 +0800
[chore](test) Add explicit snapshot recycling checks to cloud Docker suites
(#69057)
---
.../suite/SnapshotRecyclerPluginTest.groovy | 233 +++++++++++++++++++++
.../plugins/snapshot_recycler_plugin.groovy | 170 +++++++++++++++
2 files changed, 403 insertions(+)
diff --git
a/regression-test/framework/src/test/groovy/org/apache/doris/regression/suite/SnapshotRecyclerPluginTest.groovy
b/regression-test/framework/src/test/groovy/org/apache/doris/regression/suite/SnapshotRecyclerPluginTest.groovy
new file mode 100644
index 00000000000..f2658c85b77
--- /dev/null
+++
b/regression-test/framework/src/test/groovy/org/apache/doris/regression/suite/SnapshotRecyclerPluginTest.groovy
@@ -0,0 +1,233 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+package org.apache.doris.regression.suite
+
+import groovy.json.JsonOutput
+import org.apache.doris.regression.Config
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.Test
+import org.slf4j.LoggerFactory
+
+import static org.junit.jupiter.api.Assertions.*
+
+class SnapshotRecyclerPluginTest {
+ private Suite loadPlugin(Closure request) {
+ def plugin = [new File("../plugins/snapshot_recycler_plugin.groovy"),
+ new
File("regression-test/plugins/snapshot_recycler_plugin.groovy")].find {
it.isFile() }
+ assertNotNull(plugin)
+ new GroovyShell(getClass().classLoader,
+ new Binding(logger:
LoggerFactory.getLogger(getClass()))).evaluate(plugin)
+ Suite.metaClass.httpTest = { Closure action ->
+ def httpCall = [method: 'POST']
+ def http = new Expando(
+ endpoint: { String value -> httpCall.endpoint = value },
+ uri: { String value -> httpCall.uri = value },
+ op: { String value -> httpCall.method =
value.toUpperCase() },
+ body: { String value -> httpCall.body = value },
+ check: { Closure value -> httpCall.check = value })
+ action.delegate = http
+ action.resolveStrategy = Closure.DELEGATE_FIRST
+ action.call(http)
+ def (code, response) = request(httpCall.method,
+ "http://${httpCall.endpoint}${httpCall.uri}";,
httpCall.body)
+ httpCall.check.call(code, response)
+ }
+ Suite.metaClass.sleep = { long millis -> }
+ def config = new Config()
+ config.suitePath = new File('regression-test/suites').absolutePath
+ config.dataPath = new File('regression-test/data').absolutePath
+ config.realDataPath = new
File('regression-test/real-data').absolutePath
+ config.defaultDb = 'regression_test'
+ def file = new File(config.suitePath, 'snapshot_recycler_test.groovy')
+ def scriptContext = new ScriptContext(file, null, null, config, [], {
true })
+ def cluster = new SuiteCluster('snapshot_recycler_test', config)
+ def context = new SuiteContext(file, 'snapshot_recycler_test', 'p0',
scriptContext,
+ cluster, null, null, config)
+ return new Suite('snapshot_recycler_test', 'p0', context, cluster)
+ }
+
+ @AfterEach
+ void cleanup() {
+ GroovySystem.metaClassRegistry.removeMetaClass(Suite)
+ GroovySystem.metaClassRegistry.removeMetaClass(SuiteCluster)
+ }
+
+ private static String job(String status, long finish, long success) {
+ JsonOutput.toJson([status: status, last_ctime_ms: "1",
+ last_finish_time_ms: "${finish}",
last_success_time_ms: "${success}"])
+ }
+
+ @Test
+ void waitsForIdleAndRetriesTheWholeRoundUntilBothJobsSucceed() {
+ def triggers = []
+ def polls = [recycle: 0, check: 0]
+ def runs = [recycle: 0, check: 0]
+ def suite = loadPlugin { method, url, body ->
+ String kind = url.contains('/recycle_') ? 'recycle' : 'check'
+ if (url.contains('_job_info?')) {
+ polls[kind]++
+ // Both tasks are initially busy. An old successful round is
insufficient.
+ if (polls[kind] == 1) return [200, job('BUSY', 10, 10), '']
+ long finish = 10 + runs[kind]
+ // First recycle fails, second check fails, third round
succeeds.
+ long success = ((kind == 'recycle' && runs[kind] == 1)
+ || (kind == 'check' && runs[kind] == 2)) ? finish - 1
: finish
+ return [200, job('IDLE', finish, success), '']
+ }
+ assertTrue(polls[kind] >= 2, 'must wait for IDLE before
triggering')
+ assertEquals(kind == 'recycle' ? 'POST' : 'GET', method)
+ if (kind == 'recycle')
assertEquals('{"instance_ids":["test_instance"]}', body)
+ runs[kind]++
+ triggers.add(kind)
+ return [200, 'OK', '']
+ }
+ suite.recycleAndCheckSnapshot('127.0.0.1:5000', 'test_instance', 3, 3)
+ assertEquals(['recycle', 'check', 'recycle', 'check', 'recycle',
'check'], triggers)
+ }
+
+ @Test
+ void busyJobTimesOutWithoutTriggering() {
+ int triggers = 0
+ def suite = loadPlugin { method, url, body ->
+ if (!url.contains('_job_info?')) triggers++
+ [200, job('BUSY', 10, 10), '']
+ }
+ def error = assertThrows(AssertionError) {
+ suite.recycleAndCheckSnapshot('127.0.0.1:5000', 'busy_instance',
2, 2)
+ }
+ assertTrue(error.message.contains('busy_instance'))
+ assertEquals(0, triggers)
+ }
+
+ @Test
+ void staleSuccessDoesNotCountAsCompletion() {
+ int triggers = 0
+ def suite = loadPlugin { method, url, body ->
+ if (url.contains('_job_info?')) return [200, job('IDLE', 10, 10),
'']
+ triggers++
+ [200, 'OK', '']
+ }
+ assertThrows(AssertionError) {
+ suite.recycleAndCheckSnapshot('127.0.0.1:5000', 'stale_instance',
2, 2)
+ }
+ assertEquals(1, triggers, 'must not retrigger while awaiting
completion')
+ }
+
+ @Test
+ void failsAfterTheRoundLimit() {
+ def runs = [recycle: 0, check: 0]
+ def suite = loadPlugin { method, url, body ->
+ String kind = url.contains('/recycle_') ? 'recycle' : 'check'
+ if (url.contains('_job_info?')) return [200, job('IDLE', 10 +
runs[kind], 0), '']
+ runs[kind]++
+ [200, 'OK', '']
+ }
+ def error = assertThrows(AssertionError) {
+ suite.recycleAndCheckSnapshot('127.0.0.1:5000', 'failed_instance',
2, 2)
+ }
+ assertTrue(error.message.contains('2 rounds'))
+ assertEquals([recycle: 2, check: 2], runs)
+ }
+
+ @Test
+ void rejectsFailedTriggerResponse() {
+ def suite = loadPlugin { method, url, body ->
+ url.contains('_job_info?') ? [200, job('IDLE', 0, 0), ''] : [200,
'checker not enabled', '']
+ }
+ assertThrows(AssertionError) {
+ suite.recycleAndCheckSnapshot('127.0.0.1:5000', 'test_instance',
2, 2)
+ }
+ }
+
+ @Test
+ void expiryAndDeletionWaitsOnlyTriggerIdleJobsOutsideTheInterval() {
+ def info = [status: 'BUSY', last_ctime_ms: '0']
+ int triggers = 0
+ def suite = loadPlugin { method, url, body ->
+ if (url.contains('_job_info?')) return [200,
JsonOutput.toJson(info), '']
+ triggers++
+ assertEquals('POST', method)
+ assertEquals('{"instance_ids":["test_instance"]}', body)
+ return [200, 'OK', '']
+ }
+ assertFalse(suite.triggerSnapshotRecycleWhenIdle('127.0.0.1:5000',
'test_instance'))
+ info = [status: 'IDLE', last_ctime_ms:
System.currentTimeMillis().toString()]
+ assertFalse(suite.triggerSnapshotRecycleWhenIdle('127.0.0.1:5000',
'test_instance'))
+ assertEquals(0, triggers)
+ info = [status: 'IDLE', last_ctime_ms: '0']
+ assertTrue(suite.triggerSnapshotRecycleWhenIdle('127.0.0.1:5000',
'test_instance'))
+ assertEquals(1, triggers)
+ }
+
+ private Suite dataSuite(List afterRows, boolean dropTable = false) {
+ def suite = loadPlugin { method, url, body -> throw new
AssertionError('unexpected HTTP call') }
+ boolean recycled = false
+ suite.cluster.metaClass.getAllRecyclers = { boolean alive ->
+ [new Recycler(host: '127.0.0.1', httpPort: 5000, alive: true)]
+ }
+ Suite.metaClass.connectWithDockerCluster = { SuiteCluster cluster,
Closure action -> action() }
+ Suite.metaClass.recycleAndCheckSnapshot = { String address, String id
->
+ assertEquals('test_instance', id)
+ recycled = true
+ }
+ Suite.metaClass.sql = { String query ->
+ if (query.contains('information_schema.tables')) {
+ assertTrue(query.contains("ENGINE = 'Doris'"))
+ return recycled && dropTable ? [] : [['test_db', 'test_table']]
+ }
+ assertTrue(query.contains('enable_sql_cache=false'))
+ assertTrue(query.contains('enable_query_cache=false'))
+ assertTrue(query.contains('enable_file_cache=false'))
+ assertTrue(query.contains('enable_page_cache=false'))
+ assertTrue(query.contains('`test_db`.`test_table`'))
+ return recycled ? afterRows : [[1, 'a'], [1, 'a'], [2, 'b']]
+ }
+ return suite
+ }
+
+ @Test
+ void dataComparisonIgnoresOrderButPreservesDuplicates() {
+ def suite = dataSuite([[2, 'b'], [1, 'a'], [1, 'a']])
+ suite.checkSnapshotRecycle(suite.cluster, [test_instance:
suite.cluster])
+ }
+
+ @Test
+ void dataComparisonRejectsMissingExtraAndChangedRows() {
+ for (rows in [[[1, 'a'], [2, 'b']],
+ [[1, 'a'], [1, 'a'], [1, 'a'], [2, 'b']],
+ [[1, 'a'], [1, 'a'], [2, 'changed']]]) {
+ try {
+ def suite = dataSuite(rows)
+ def error = assertThrows(AssertionError) {
+ suite.checkSnapshotRecycle(suite.cluster, [test_instance:
suite.cluster])
+ }
+ assertTrue(error.message.contains('Data changed after
recycling'))
+ } finally {
+ cleanup()
+ }
+ }
+ }
+
+ @Test
+ void dataComparisonRejectsMissingTable() {
+ def suite = dataSuite([], true)
+ assertThrows(AssertionError) {
+ suite.checkSnapshotRecycle(suite.cluster, [test_instance:
suite.cluster])
+ }
+ }
+
+}
diff --git a/regression-test/plugins/snapshot_recycler_plugin.groovy
b/regression-test/plugins/snapshot_recycler_plugin.groovy
new file mode 100644
index 00000000000..702e29043cc
--- /dev/null
+++ b/regression-test/plugins/snapshot_recycler_plugin.groovy
@@ -0,0 +1,170 @@
+// 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.Suite
+import org.apache.doris.regression.suite.SuiteCluster
+
+def recyclerRequest = { Suite suite, String address, String method, String
path, String payload = null ->
+ String out
+ suite.httpTest {
+ endpoint address
+ uri "/RecyclerService/http/${path}".toString()
+ if (method == 'GET') {
+ op 'get'
+ } else {
+ body payload
+ }
+ check { code, response ->
+ assert code == 200 : "Recycler request failed: path=${path},
status=${code}, body=${response}"
+ assert response?.trim() : "Empty recycler response: path=${path}"
+ out = response.trim()
+ }
+ }
+ return out
+}
+
+// The snapshot Docker suites set both scheduling intervals to 10 seconds.
+Suite.metaClass.recycleAndCheckSnapshot = { String recyclerAddress, String
instanceId,
+ int maxRounds = 10, int maxPolls =
60,
+ boolean requireCheck = true ->
+ Suite suite = delegate as Suite
+ assert maxRounds > 0 && maxPolls > 0
+ String token = "greedisgood9999"
+ String encodedInstanceId = URLEncoder.encode(instanceId, 'UTF-8')
+ def lastJobs = [:]
+
+ def request = { String method, String path, String payload = null ->
+ // A successful trigger only enqueues a job.
+ return recyclerRequest(suite, recyclerAddress, method, path, payload)
+ }
+ def waitJob = { String kind, Closure ready ->
+ for (int poll = 0; poll < maxPolls; poll++) {
+ def info = new JsonSlurper().parseText(request('GET',
+
"${kind}_job_info?token=${token}&instance_id=${encodedInstanceId}".toString()))
+ assert info.status in ['IDLE', 'BUSY'] : "Invalid ${kind} job:
instance=${instanceId}, ${info}"
+ lastJobs[kind] = info
+ if (ready(info)) return info
+ suite.sleep(1000L)
+ }
+ assert false : "Timed out waiting for ${kind}: instance=${instanceId},
jobs=${lastJobs}"
+ }
+ def runJob = { String kind ->
+ def before = waitJob(kind) { info ->
+ // Manual requests also obey the scheduling interval. These suites
disable automatic
+ // scheduling with recycle_whitelist=dummy, so a rejected trigger
would never run.
+ info.status == 'IDLE' && System.currentTimeMillis() -
+ (info.last_ctime_ms ?: '0').toString().toLong() > 10000L
+ }
+ long previousFinish = (before.last_finish_time_ms ?:
'0').toString().toLong()
+ String response = kind == 'recycle'
+ ? request('POST', "recycle_instance?token=${token}".toString(),
+ JsonOutput.toJson([instance_ids: [instanceId]]))
+ : request('GET',
"check_instance?token=${token}&instance_id=${encodedInstanceId}".toString())
+ assert response.equalsIgnoreCase('OK') : "Failed to trigger ${kind}:
instance=${instanceId}, ${response}"
+ def finished = waitJob(kind) { info ->
+ info.status == 'IDLE' &&
+ (info.last_finish_time_ms ?: '0').toString().toLong() >
previousFinish
+ }
+ return finished.last_success_time_ms != null &&
+ finished.last_finish_time_ms.toString().toLong() ==
finished.last_success_time_ms.toString().toLong()
+ }
+
+ for (int round = 1; round <= maxRounds; round++) {
+ suite.getLogger().info("Starting snapshot recycle/check:
instance=${instanceId}, round=${round}/${maxRounds}")
+ boolean recycled = runJob('recycle')
+ // Check cloned instances through SQL below: their inherited snapshot
files do not
+ // have snapshot keys in the derived instance, so the recycler checker
rejects them.
+ boolean checked = !requireCheck || runJob('check')
+ suite.getLogger().info("Snapshot recycle/check:
instance=${instanceId}, round=${round}/${maxRounds}, "
+ + "recycled=${recycled}, checked=${checked},
checkRequired=${requireCheck}, jobs=${lastJobs}")
+ if (recycled && checked) return
+ }
+ assert false : "Recycle/check failed after ${maxRounds} rounds:
instance=${instanceId}, jobs=${lastJobs}"
+}
+
+// Used only inside bounded waits for snapshot expiry or complete instance
deletion.
+// A fully deleted instance has no checker job to run; callers assert its
metadata disappears.
+Suite.metaClass.triggerSnapshotRecycleWhenIdle = { String recyclerAddress,
String instanceId ->
+ Suite suite = delegate as Suite
+ String token = "greedisgood9999"
+ String encodedInstanceId = URLEncoder.encode(instanceId, 'UTF-8')
+ String out = recyclerRequest(suite, recyclerAddress, 'GET',
+
"recycle_job_info?token=${token}&instance_id=${encodedInstanceId}".toString())
+ def info = new JsonSlurper().parseText(out)
+ assert info.status in ['IDLE', 'BUSY'] : "Invalid recycle job for
${instanceId}: ${info}"
+ if (info.status != 'IDLE' || System.currentTimeMillis() -
+ (info.last_ctime_ms ?: '0').toString().toLong() <= 10000L) return
false
+ String triggerOut = recyclerRequest(suite, recyclerAddress, 'POST',
+ "recycle_instance?token=${token}".toString(),
JsonOutput.toJson([instance_ids: [instanceId]]))
+ assert triggerOut.equalsIgnoreCase('OK') : "Failed to trigger recycle for
${instanceId}: ${triggerOut}"
+ return true
+}
+
+// Record all remaining OLAP rows before recycling, then read them again
without caches.
+// Callers pass the current instance IDs after rollback and omit clusters
whose creation failed.
+// A null cluster allows checking a predecessor that still owns snapshots but
has no running FE.
+Suite.metaClass.checkSnapshotRecycle = { SuiteCluster recyclerCluster, Map
instanceClusters ->
+ Suite suite = delegate as Suite
+ assert !instanceClusters.isEmpty()
+ def recyclers = recyclerCluster.getAllRecyclers(true)
+ assert !recyclers.isEmpty() : "No live recycler in ${recyclerCluster.name}"
+ def (ip, port) = recyclers[0].getHttpAddress()
+ String recyclerAddress = "${ip}:${port}"
+ def readData = { SuiteCluster dataCluster ->
+ def data = [:]
+ suite.connectWithDockerCluster(dataCluster) {
+ def tables = suite.sql("""
+ SELECT TABLE_SCHEMA, TABLE_NAME FROM information_schema.tables
+ WHERE ENGINE = 'Doris' AND TABLE_SCHEMA NOT IN
('information_schema', 'mysql', '__internal_schema')
+ ORDER BY TABLE_SCHEMA, TABLE_NAME
+ """)
+ tables.each { table ->
+ String name = table.collect { '`' + it.toString().replace('`',
'``') + '`' }.join('.')
+ def rows = suite.sql("""SELECT /*+
SET_VAR(enable_sql_cache=false,
+ enable_query_cache=false, enable_file_cache=false,
enable_page_cache=false) */ * FROM ${name}""")
+ // Sort serialized rows instead of using a set: duplicate
multiplicities must survive.
+ data[name] = rows.collect { JsonOutput.toJson(it) }.sort()
+ }
+ }
+ return data
+ }
+ def expected = [:]
+ instanceClusters.each { instanceId, dataCluster ->
+ if (dataCluster != null) {
+ expected[instanceId] = readData(dataCluster)
+ suite.getLogger().info("Snapshot data baseline:
instance=${instanceId}, tables=${expected[instanceId].size()}, "
+ + "rows=${expected[instanceId].values().sum { it.size() }
?: 0}")
+ }
+ }
+ instanceClusters.each { instanceId, dataCluster ->
+ if (dataCluster != null && dataCluster != recyclerCluster) {
+ suite.recycleAndCheckSnapshot(recyclerAddress,
instanceId.toString(), 10, 60, false)
+ } else {
+ suite.recycleAndCheckSnapshot(recyclerAddress,
instanceId.toString())
+ }
+ }
+ instanceClusters.each { instanceId, dataCluster ->
+ if (dataCluster != null) {
+ def actual = readData(dataCluster)
+ assert actual == expected[instanceId] : "Data changed after
recycling: instance=${instanceId}, " +
+ "expected=${expected[instanceId]}, actual=${actual}"
+ }
+ }
+}
+
+logger.info("Added snapshot recycle/check helpers to Suite")
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
