github-actions[bot] commented on code in PR #68283:
URL: https://github.com/apache/doris/pull/68283#discussion_r4070769435
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/execute/ConnectorRewriteDriver.java:
##########
@@ -224,30 +223,63 @@ public void onTaskFailed(Long taskId, Exception error) {
tasks.add(task);
}
+ List<ConnectorRewriteGroupTask> submitted = Lists.newArrayList();
try {
- for (TransientTaskExecutor task : tasks) {
+ for (ConnectorRewriteGroupTask task : tasks) {
Env.getCurrentEnv().getTransientTaskManager().addMemoryTask(task);
+ submitted.add(task);
}
} catch (JobException e) {
+ // Groups submitted before the failing call already bind the
shared transaction; drain them so the
+ // caller never rolls that transaction back while a live group
still reports into it.
+ drain(submitted);
throw new UserException("Failed to submit rewrite tasks: " +
e.getMessage(), e);
}
int maxWaitTime = ctx.getSessionVariable().getInsertTimeoutS();
try {
boolean completed = collector.await(maxWaitTime, TimeUnit.SECONDS);
if (!completed) {
+ // The owner gave up waiting: stop every live group and wait
for its terminal callback so no
+ // group is still reporting into the shared transaction when
the caller rolls it back.
+ drain(submitted);
throw new UserException("Rewrite tasks did not complete within
timeout");
}
if (collector.getFirstError() != null) {
throw new UserException("Some rewrite tasks failed: " +
collector.getFirstError().getMessage(),
collector.getFirstError());
}
} catch (InterruptedException e) {
+ // The interrupt flag was cleared by the exception, so the drain
below can still wait.
+ drain(submitted);
Thread.currentThread().interrupt();
throw new UserException("Wait for rewrite tasks completion was
interrupted", e);
}
}
+ /**
+ * Cancels every submitted group and waits (bounded) for its terminal
callback, so a shared transaction
+ * is never rolled back while a live group still has commit data flowing
into it.
+ */
+ private void drain(List<ConnectorRewriteGroupTask> submitted) {
+ for (ConnectorRewriteGroupTask task : submitted) {
+ try {
+ task.cancel();
+ } catch (Exception e) {
+ LOG.warn("Failed to cancel rewrite task {}: {}", task.getId(),
e.getMessage());
+ }
+ }
+ long waitSeconds = Math.max(1,
ctx.getSessionVariable().getInsertTimeoutS());
+ for (ConnectorRewriteGroupTask task : submitted) {
+ try {
+ task.awaitTerminal(waitSeconds, TimeUnit.SECONDS);
Review Comment:
[P1] Use one deadline for the whole drain. The caller has already spent one
full `insert_timeout` in `collector.await`, but this loop gives every submitted
group a fresh full timeout sequentially. If G groups remain nonterminal, a
timed-out rewrite can retain the shared connector transaction and task graph
for roughly `T + G*T` (the default T is four hours) before rollback and
response; interruption or submission failure can still add up to `G*T`. The new
tests stub every await to return immediately, so they miss the amplification.
Use one shared drain budget and make queued cancellation conclusively terminal
so rollback safety does not require another full timeout per task, with a
multi-task never-terminal test.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/execute/ConnectorRewriteDriver.java:
##########
@@ -224,30 +223,63 @@ public void onTaskFailed(Long taskId, Exception error) {
tasks.add(task);
}
+ List<ConnectorRewriteGroupTask> submitted = Lists.newArrayList();
try {
- for (TransientTaskExecutor task : tasks) {
+ for (ConnectorRewriteGroupTask task : tasks) {
Env.getCurrentEnv().getTransientTaskManager().addMemoryTask(task);
+ submitted.add(task);
Review Comment:
[P1] Propagate outer statement cancellation into the rewrite owner. A
TIMEOUT/KILL only reaches the outer `StmtExecutor` here, but
`ExecuteActionCommand` is neither `CancelableCommand` nor coordinator-backed,
this driver receives no terminal state, and every group runs under a fresh
context/executor. The groups can therefore finish after the outer statement is
cancelled, and source registration plus `txnManager.commit` can still commit
the shared rewrite. Install a sticky outer-executor-to-driver cancellation
handoff, cancel and drain submitted groups, and serialize cancellation with the
source-registration/final-commit decision, with an outer-cancel crossing 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]
