This is an automated email from the ASF dual-hosted git repository.
JNSimba 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 46798454678 [fix](regression) Stabilize MySQL streaming job waits
(#69822)
46798454678 is described below
commit 46798454678d67c8dc47b048e9a6675ca79669bd
Author: wudi <[email protected]>
AuthorDate: Fri Oct 30 11:18:46 2026 +0800
[fix](regression) Stabilize MySQL streaming job waits (#69822)
### What problem does this PR solve?
Problem Summary:
The MySQL streaming regression cases can time out when run separately
against an idle source. The basic case waits for a third successful task
after its snapshot has already completed. The schema-change ignore case
reads the success count after a row is visible and waits for another
commit. Concurrent cases can advance the shared MySQL binlog and satisfy
these waits with unrelated heartbeat-only tasks, hiding the dependency.
Wait for each case's committed row statistics and target data instead,
retain the initial binlog-phase barrier for schema changes, and replace
fixed sleeps in the basic case with progress polling. Keep all existing
expected data and schema checks.
---
.../cdc/test_streaming_mysql_job.groovy | 23 ++++++++--------
...streaming_mysql_job_schema_change_ignore.groovy | 31 +++++++++-------------
2 files changed, 24 insertions(+), 30 deletions(-)
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job.groovy
index 29680f8506d..e899bbf96c3 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job.groovy
@@ -53,6 +53,13 @@ suite("test_streaming_mysql_job",
"p0,external,mysql,external_docker,external_do
String bucket = getS3BucketName()
String driver_url =
"https://${bucket}.${s3_endpoint}/regression/jdbc_driver/mysql-connector-j-8.4.0.jar";
+ def waitForScannedRows = { long expected ->
+ Awaitility.await().atMost(300, SECONDS).pollInterval(1,
SECONDS).until({
+ def rows = sql """SELECT LoadStatistic, Status FROM
jobs("type"="insert") WHERE Name='${jobName}'"""
+ rows[0][1] == "RUNNING" && parseJson(rows[0][0]).scannedRows
>= expected
+ })
+ }
+
// create test
connect("root", "123456",
"jdbc:mysql://${externalEnvIp}:${mysql_port}") {
sql """CREATE DATABASE IF NOT EXISTS ${mysqlDb}"""
@@ -115,17 +122,9 @@ suite("test_streaming_mysql_job",
"p0,external,mysql,external_docker,external_do
assert createTalInfo.contains("UNIQUE KEY(`name`)");
assert createTalInfo.contains("DISTRIBUTED BY HASH(`name`) BUCKETS
AUTO");
- // check job running
+ // Wait for committed snapshot rows; an idle source does not produce
more successful tasks.
try {
- Awaitility.await().atMost(300, SECONDS)
- .pollInterval(1, SECONDS).until(
- {
- def jobSuccendCount = sql """ select SucceedTaskCount
from jobs("type"="insert") where Name = '${jobName}' and
ExecuteType='STREAMING' """
- log.info("jobSuccendCount: " + jobSuccendCount)
- // check job status and succeed task count larger than
2
- jobSuccendCount.size() == 1 && '3' <=
jobSuccendCount.get(0).get(0)
- }
- )
+ waitForScannedRows(4)
} catch (Exception ex){
def showjob = sql """select * from jobs("type"="insert") where
Name='${jobName}'"""
def showtask = sql """select * from tasks("type"="insert") where
JobName='${jobName}'"""
@@ -146,7 +145,7 @@ suite("test_streaming_mysql_job",
"p0,external,mysql,external_docker,external_do
sql """DELETE FROM ${mysqlDb}.${table1} WHERE name = 'A1';"""
}
- sleep(60000); // wait for cdc incremental data
+ waitForScannedRows(7)
// check incremental data
qt_select_binlog_table1 """ SELECT * FROM ${table1} order by name asc
"""
@@ -165,7 +164,7 @@ suite("test_streaming_mysql_job",
"p0,external,mysql,external_docker,external_do
sql """INSERT INTO ${mysqlDb}.${table1} (name,age) VALUES
('Apache',40);"""
}
- sleep(60000); // wait for cdc incremental data
+ waitForScannedRows(8)
// check incremental data
qt_select_next_binlog_table1 """ SELECT * FROM ${table1} order by name
asc """
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_schema_change_ignore.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_schema_change_ignore.groovy
index f300f3f7922..f28e3d49b15 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_schema_change_ignore.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_schema_change_ignore.groovy
@@ -37,15 +37,12 @@ suite("test_streaming_mysql_job_schema_change_ignore",
"p0,external,mysql,extern
id INT PRIMARY KEY, name VARCHAR(50), removed VARCHAR(50))"""
sql "INSERT INTO test_cdc_db.mysql_schema_ignore VALUES (1, 'initial',
'old')"
}
- def succeedCount = {
- (sql """SELECT SucceedTaskCount FROM jobs("type"="insert") WHERE
Name='test_mysql_schema_ignore'""")[0][0].toLong()
- }
- def waitForCommit = { long previous ->
- Awaitility.await().atMost(180, SECONDS).pollInterval(1,
SECONDS).until({ succeedCount() > previous })
- }
- def waitForRow = { int id ->
+ def waitForCommittedRow = { int id ->
Awaitility.await().atMost(180, SECONDS).pollInterval(1,
SECONDS).until({
- (sql "SELECT COUNT(*) FROM mysql_schema_ignore WHERE
id=${id}")[0][0].toLong() == 1
+ def statistic = (sql """SELECT LoadStatistic FROM
jobs("type"="insert") WHERE Name='test_mysql_schema_ignore'""")[0][0]
+ // Row visibility precedes the source offset commit; statistics
advance with that commit.
+ parseJson(statistic).scannedRows >= id &&
+ (sql "SELECT COUNT(*) FROM mysql_schema_ignore WHERE
id=${id}")[0][0].toLong() == 1
})
}
sql """CREATE JOB test_mysql_schema_ignore PROPERTIES ("max_interval" =
"5") ON STREAMING
@@ -56,16 +53,18 @@ suite("test_streaming_mysql_job_schema_change_ignore",
"p0,external,mysql,extern
"include_tables" = "mysql_schema_ignore", "offset" = "initial",
"schema_change_behavior" = "ignore"
) TO DATABASE ${currentDb} ("table.create.properties.replication_num"
= "1")"""
- waitForRow(1)
- waitForCommit(Math.max(1L, succeedCount()))
+ waitForCommittedRow(1)
+ Awaitility.await().atMost(180, SECONDS).pollInterval(1, SECONDS).until({
+ def offset = parseJson((sql """SELECT CurrentOffset FROM
jobs("type"="insert") WHERE Name='test_mysql_schema_ignore'""")[0][0])
+ offset instanceof Map && offset.splitId == "binlog-split"
+ })
connect("root", "123456", "jdbc:mysql://${externalEnvIp}:${mysqlPort}") {
sql "ALTER TABLE test_cdc_db.mysql_schema_ignore ADD COLUMN skipped
INT"
sql "ALTER TABLE test_cdc_db.mysql_schema_ignore DROP COLUMN removed"
sql "ALTER TABLE test_cdc_db.mysql_schema_ignore MODIFY COLUMN name
VARCHAR(150)"
sql "INSERT INTO test_cdc_db.mysql_schema_ignore VALUES (2, 'ignored',
20)"
}
- waitForRow(2)
- waitForCommit(succeedCount())
+ waitForCommittedRow(2)
qt_ignored_columns """SELECT column_name FROM information_schema.columns
WHERE table_schema='${currentDb}' AND table_name='mysql_schema_ignore'
ORDER BY column_name"""
sql "PAUSE JOB WHERE jobname='test_mysql_schema_ignore'"
@@ -73,20 +72,16 @@ suite("test_streaming_mysql_job_schema_change_ignore",
"p0,external,mysql,extern
connect("root", "123456", "jdbc:mysql://${externalEnvIp}:${mysqlPort}") {
sql "INSERT INTO test_cdc_db.mysql_schema_ignore VALUES (3, 'rebuilt',
30)"
}
- waitForRow(3)
- waitForCommit(succeedCount())
+ waitForCommittedRow(3)
sql "PAUSE JOB WHERE jobname='test_mysql_schema_ignore'"
sql """ALTER JOB test_mysql_schema_ignore FROM MYSQL
("schema_change_behavior" = "evolve")
TO DATABASE ${currentDb}"""
- long beforeResume = succeedCount()
sql "RESUME JOB WHERE jobname='test_mysql_schema_ignore'"
- waitForCommit(beforeResume)
connect("root", "123456", "jdbc:mysql://${externalEnvIp}:${mysqlPort}") {
sql "ALTER TABLE test_cdc_db.mysql_schema_ignore ADD COLUMN new_value
INT"
sql "INSERT INTO test_cdc_db.mysql_schema_ignore VALUES (4, 'evolved',
40, 400)"
}
- waitForRow(4)
- waitForCommit(succeedCount())
+ waitForCommittedRow(4)
qt_evolved_columns """SELECT column_name FROM information_schema.columns
WHERE table_schema='${currentDb}' AND table_name='mysql_schema_ignore'
ORDER BY column_name"""
qt_evolved_rows "SELECT id, name, removed, new_value FROM
mysql_schema_ignore ORDER BY id"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
