Copilot commented on code in PR #68168:
URL: https://github.com/apache/doris/pull/68168#discussion_r4056461046
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -128,7 +128,10 @@ public StreamingResponseBody
fetchRecordStream(FetchRecordRequest fetchReq) thro
LOG.info("Generated meta for job {}: {}", fetchReq.getJobId(),
meta);
}
- sourceReader = Env.getCurrentEnv().getReader(fetchReq,
!isLong(fetchReq.getJobId()));
+ sourceReader =
+ isJobDrivenTvf(fetchReq.getJobId())
+ ? Env.getCurrentEnv().getReaderAndClaim(fetchReq,
fetchReq.getTaskId())
+ : Env.getCurrentEnv().getReader(fetchReq, true);
readResult = sourceReader.prepareAndSubmitSplit(fetchReq);
Review Comment:
If `prepareAndSubmitSplit` fails after this call creates or replaces a job
reader, `fetchRecordStream` exits through its catch before returning a
`StreamingResponseBody`, so the cleanup finally block at lines 150-160 is never
installed. The reader remains in `Env` with its executor and any partially
opened fetcher, and there is no cleanup in `ClientController`; release the
captured reader and remove its context on preparation failure.
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/common/Env.java:
##########
@@ -140,12 +141,15 @@ public SourceReader getReaderAndClaim(JobBaseConfig
jobConfig, String taskId) {
try {
JobContext context = jobContexts.get(jobId);
if (context != null
- && jobConfig instanceof WriteRecordRequest
- && ((WriteRecordRequest) jobConfig).isRebuildReader()) {
- // FE declared the previous task abnormal: swap in a fresh
reader instance so the
- // old task's thread can never reach the new fetcher.
+ && (jobConfig instanceof FetchRecordRequest
+ || (jobConfig instanceof WriteRecordRequest
+ && ((WriteRecordRequest)
jobConfig).isRebuildReader()))) {
Review Comment:
The added EnvTest covers only WriteRecordRequest reuse/rebuild; it never
exercises this new FetchRecordRequest branch. The request-scoped replacement
and stale-task ownership checks are the core of the TVF pause/resume fix, so
this suite cannot detect regressions in that lifecycle. Add deterministic
coverage with two FetchRecordRequests/task IDs that verifies a fresh reader and
that releasing the first request cannot affect the successor, and run the
pause/resume regression.
--
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]
