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 ab529c9e0ad [fix](regression) Stabilize streaming job CDC cases
(#66978)
ab529c9e0ad is described below
commit ab529c9e0ada98d5d1cc0f61d25d6be022bc26ca
Author: wudi <[email protected]>
AuthorDate: Tue Aug 25 07:52:34 2026 +0800
[fix](regression) Stabilize streaming job CDC cases (#66978)
### What problem does this PR solve?
Issue Number: N/A
Related PR: N/A
Problem Summary:
The PostgreSQL TVF publication case used a fixed five-second delay
before manually dropping a user-provided replication slot. The TVF
connection closes automatically, but the slot could still be active when
the delay expired.
The MySQL special-offset case waited for two successful tasks even
though the first task had already committed all available data.
Scheduling the second task therefore depended on unrelated binlog
traffic from concurrent cases.
This change waits until the PostgreSQL slot becomes inactive, waits only
for the committed MySQL CREATE task, and pauses the MySQL job before
recording the ALTER offset and inserting the rows consumed after resume.
---
.../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]
