github-actions[bot] commented on code in PR #66978:
URL: https://github.com/apache/doris/pull/66978#discussion_r3826808294
##########
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.
Review Comment:
The same teardown race remains in the parallel
`test_streaming_postgres_job_publication.groovy` path: after dropping its
running job, lines 239-245 still `sleep(5000)` and then call
`pg_drop_replication_slot` on a user-owned slot. That connection can remain
active longer than five seconds for the same reason fixed here; the other
established PostgreSQL cleanup tests already poll `active=false`. Please apply
this state-based gate there as well, otherwise the PR leaves an equivalent CDC
ownership case flaky.
##########
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"
Review Comment:
Waiting for `status = PAUSED` does not prove the remote MySQL writer is
quiescent. The successor installed after the first success can still have
`runningBackendId == 0`, making `releaseRemoteReader()` a no-op, or it can race
past its cancellation check and dispatch asynchronous `/api/writeRecords` work
after PAUSE. Rows 20/21 can then be buffered or loaded by the canceled task
while its offset callback is skipped; ALTER/RESUME starts again from the
pre-insert position and can load them a second time into this DUPLICATE KEY
table. Please wait for an acknowledged reader/task drain before taking the new
position and inserting rows, analogous to the explicit probe/stability gate in
the PostgreSQL special-offset test.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
