This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new 44e9789bb3f branch-4.2: [fix](regression) Stabilize MySQL streaming
job waits #69822 (#69873)
44e9789bb3f is described below
commit 44e9789bb3f4efc53af8ded2ddd3d2b1c4930c92
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Fri Oct 30 18:24:30 2026 +0800
branch-4.2: [fix](regression) Stabilize MySQL streaming job waits #69822
(#69873)
Cherry-picked from #69822
Co-authored-by: wudi <[email protected]>
---
.../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]
