This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new a22611dff2d branch-4.1: [fix](regression) Stabilize streaming job CDC
cases #66978 (#67083)
a22611dff2d is described below
commit a22611dff2da063ee87c243d2c9e61958cb37637
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Wed Aug 26 10:39:53 2026 +0800
branch-4.1: [fix](regression) Stabilize streaming job CDC cases #66978
(#67083)
Cherry-picked from #66978
Co-authored-by: wudi <[email protected]>
---
.../cdc/test_streaming_mysql_job_special_offset.groovy | 15 +++++++--------
.../cdc/test_streaming_postgres_job_publication.groovy | 9 ++++++++-
.../cdc/tvf/test_cdc_stream_tvf_publication.groovy | 10 +++++++++-
3 files changed, 24 insertions(+), 10 deletions(-)
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_special_offset.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_special_offset.groovy
index c4234913661..fa3df0f913f 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_special_offset.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_mysql_job_special_offset.groovy
@@ -130,14 +130,18 @@ suite("test_streaming_mysql_job_special_offset",
"p0,external,mysql,external_doc
// Verify data after CREATE with specific offset
qt_select_after_create """ SELECT * FROM ${currentDb}.${table1} ORDER
BY id """
- // Wait for current task to complete (commit offset successfully)
before PAUSE,
- // otherwise PAUSE may race with a running task and cause commit
offset failure.
+ // Wait for the CREATE task to commit before changing its offset.
Awaitility.await().atMost(300, SECONDS).pollInterval(2,
SECONDS).until({
def cnt = sql """select SucceedTaskCount from
jobs("type"="insert") where Name='${jobName}' and ExecuteType='STREAMING'"""
- return cnt.size() == 1 && (cnt.get(0).get(0) as int) >= 2
+ return cnt.size() == 1 && (cnt.get(0).get(0) as int) >= 1
})
// Step 2: Get a new binlog position (different from CREATE), insert
data, ALTER to it
+ sql "PAUSE JOB where jobname = '${jobName}'"
+ Awaitility.await().atMost(30, SECONDS).pollInterval(1, SECONDS).until({
+ def jobStatus = sql """select status from jobs("type"="insert")
where Name='${jobName}'"""
+ return jobStatus[0][0] == "PAUSED"
+ })
def alterBinlogFile = ""
def alterBinlogPos = ""
connect("root", "123456",
"jdbc:mysql://${externalEnvIp}:${mysql_port}") {
@@ -148,11 +152,6 @@ suite("test_streaming_mysql_job_special_offset",
"p0,external,mysql,external_doc
sql """INSERT INTO ${mysqlDb}.${table1} VALUES (20, 'alter1')"""
sql """INSERT INTO ${mysqlDb}.${table1} VALUES (21, 'alter2')"""
}
- sql "PAUSE JOB where jobname = '${jobName}'"
- Awaitility.await().atMost(30, SECONDS).pollInterval(1, SECONDS).until({
- def jobStatus = sql """select status from jobs("type"="insert")
where Name='${jobName}'"""
- return jobStatus[0][0] == "PAUSED"
- })
def alterOffsetJson =
"""{"file":"${alterBinlogFile}","pos":"${alterBinlogPos}"}"""
log.info("ALTER to new offset: ${alterOffsetJson}")
sql """ALTER JOB ${jobName}
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_publication.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_publication.groovy
index d38d07d8c4e..052475c03e4 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_publication.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/test_streaming_postgres_job_publication.groovy
@@ -236,7 +236,14 @@ suite("test_streaming_postgres_job_publication",
"p0,external,pg,external_docker
// Drop job: Doris must NOT touch user-provided slot or publication.
sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
- sleep(5000)
+ Awaitility.await().atMost(60, SECONDS).pollInterval(1, SECONDS).until({
+ boolean inactive = false
+ connect("${pgUser}", "${pgPassword}",
"jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
+ def slot = sql """SELECT active FROM pg_replication_slots
WHERE slot_name = '${userSlot}'"""
+ inactive = slot.size() == 1 && slot[0][0] == false
+ }
+ inactive
+ })
connect("${pgUser}", "${pgPassword}",
"jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
def pubAfter = sql """SELECT COUNT(1) FROM pg_publication WHERE
pubname = '${userPub}'"""
assert pubAfter[0][0] == 1 : "User-provided publication must be
preserved after job deletion"
diff --git
a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_publication.groovy
b/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_publication.groovy
index 2a3a163cef5..2d4f582649f 100644
---
a/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_publication.groovy
+++
b/regression-test/suites/job_p0/streaming_job/cdc/tvf/test_cdc_stream_tvf_publication.groovy
@@ -185,7 +185,15 @@ suite("test_cdc_stream_tvf_publication",
"p0,external,pg,external_docker,externa
// Drop job: Doris must NOT touch user-provided slot or publication.
sql """DROP JOB IF EXISTS where jobname = '${jobName}'"""
- sleep(5000)
+ // The TVF reader closes its connection automatically, so wait for the
user slot to become inactive.
+ Awaitility.await().atMost(60, SECONDS).pollInterval(1, SECONDS).until({
+ boolean inactive = false
+ connect("${pgUser}", "${pgPassword}",
"jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
+ def slot = sql """SELECT active FROM pg_replication_slots
WHERE slot_name = '${userSlot}'"""
+ inactive = slot.size() == 1 && slot[0][0] == false
+ }
+ inactive
+ })
connect("${pgUser}", "${pgPassword}",
"jdbc:postgresql://${externalEnvIp}:${pg_port}/${pgDB}") {
def pubAfter = sql """SELECT COUNT(1) FROM pg_publication WHERE
pubname = '${userPub}'"""
assert pubAfter[0][0] == 1 : "user-provided publication must be
preserved after job deletion"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
