924060929 commented on code in PR #68283:
URL: https://github.com/apache/doris/pull/68283#discussion_r4070870141
##########
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:
Fixed. The driver now takes the outer `StmtExecutor` and installs a sticky
cancellation handoff: it picks up a cancellation that landed before the driver
existed, registers a cancel delegate for the ones that arrive later, and
re-polls the sticky `StmtExecutor.getPendingCancelReason()` in the group wait
loop. On cancellation it cancels and drains the submitted groups. Source
registration and the final commit both run under a lock that the cancellation
handoff also takes, so a cancellation that wins the lock rolls back instead of
registering/committing, and one that arrives while the critical section runs is
linearized after the commit. Added
`outerCancellationBeforeCommitRollsBackWithoutRegisteringOrCommitting` (a group
completes, cancellation is observed exactly at the register/commit decision:
rollback, no register, no commit) and
`outerCancellationWhileWaitingDrainsLiveGroupsBeforeRollback`.
##########
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:
Fixed. `drain` now takes a single budget and computes one shared deadline
for all submitted groups instead of granting each a fresh `insert_timeout`, so
a timed-out rewrite cannot hold the shared transaction/task graph for `T +
G*T`. `ConnectorRewriteGroupTask.cancel()` also makes queued cancellation
conclusively terminal: a task the scheduler has not started writes nothing, so
it counts down its terminal latch immediately and a drain does not burn a full
timeout on it. The timeout/interruption/Nth-submission tests and the new
multi-task never-terminal and outer-cancel crossing tests cover this.
--
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]
