github-actions[bot] commented on code in PR #68168:
URL: https://github.com/apache/doris/pull/68168#discussion_r4070143442
##########
fs_brokers/cdc_client/src/main/java/org/apache/doris/cdcclient/service/PipelineCoordinator.java:
##########
@@ -144,10 +157,27 @@ public StreamingResponseBody
fetchRecordStream(FetchRecordRequest fetchReq) thro
fetchReq.getTaskId(),
ex);
throw new StreamException(ex);
+ } finally {
+ closeTvfReader(fetchReq, sourceReader);
}
};
}
+ private void closeTvfReader(FetchRecordRequest request, SourceReader
sourceReader) {
+ Env env = Env.getCurrentEnv();
+ if (isJobDrivenTvf(request.getJobId())) {
+ env.detachReaderIfOwner(request.getJobId(), request.getTaskId());
Review Comment:
[P1] Confirm the old reader is quiescent before resuming. This removes local
ownership before `sourceReader.release()` has stopped the captured engine.
Manual PAUSE sends asynchronous BE cancellation and waits at most ten seconds,
so an immediate RESUME can schedule task B while task A's body is still
unwinding. On the same BE, B then sees no context and can prepare against A's
still-active PostgreSQL slot or MySQL/OceanBase server ID; on another scan BE,
the independent loopback cdc_client cannot observe even a local releasing
marker. That can fail the fresh task during the exact pause/resume overlap this
patch targets. Please require a teardown-confirmed handoff across the prior
scan BE(s), or pin the successor and retain a local releasing barrier until
identity quiescence, and cover both placements with a deterministic latch-based
test.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertTask.java:
##########
@@ -92,6 +93,7 @@ public StreamingInsertTask(long jobId,
this.originTvfProps = originTvfProps;
this.cloudCluster = cloudCluster;
this.auditEnabled =
S3TableValuedFunction.NAME.equalsIgnoreCase(offsetProvider.getSourceType());
+ this.noRetry =
CdcStreamTableValuedFunction.NAME.equalsIgnoreCase(offsetProvider.getSourceType());
Review Comment:
[P2] Bound offsets abandoned by no-retry tasks. A CDC body can finish and
put this task's offset into the cdc_client's unbounded `taskOffsetCache`, then
the load/transaction can fail before `beforeCommitted` fetches and removes it.
Although exhausting the old retry budget could already leak one key, a
transient first-attempt failure previously retried under this task ID,
overwrote the key, and let a later success consume it. With `noRetry`, that
recoverable failure immediately abandons the ID; recurring
fail-once/resume-success incidents can therefore strand unbounded distinct
entries. Please bound/expire this cache and invalidate failed-task and
closed-job entries, with a regression where the scan completes but fails before
offset fetch.
--
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]
