github-actions[bot] commented on code in PR #66914:
URL: https://github.com/apache/doris/pull/66914#discussion_r3838176267
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java:
##########
@@ -390,6 +402,39 @@ public void releaseRemoteReader() {
}
}
+ /** Wait for the BE to acknowledge reader release before allowing a
successor to reuse the source. */
+ boolean releaseRemoteReaderAndWait() {
+ if (runningBackendId <= 0) {
Review Comment:
[P1] Do not acknowledge release before the execution owner can publish its
backend. A worker can pass `run()`'s cancel check and block in
`resolveBackend()` while `runningBackendId` is still zero; PAUSE then returns
true here without an RPC. The worker can resume, publish the backend, and send
`/api/writeRecords` without another cancel check, starting a remote reader
after PAUSE has completed. Linearize backend/request publication with
cancellation or replay an acknowledged release after publication, and add a
latch test for this exact window.
##########
fe/fe-core/src/main/java/org/apache/doris/nereids/trees/plans/commands/execute/ConnectorRewriteGroupTask.java:
##########
@@ -155,10 +180,18 @@ public void cancel() throws JobException {
LOG.info("[Connector Rewrite Task] taskId: {} cancelled", taskId);
}
- private void executeGroup(ConnectContext taskConnectContext,
+ void awaitTerminal() throws InterruptedException {
+ terminal.await();
+ }
+
+ protected void executeGroup(ConnectContext taskConnectContext,
RewriteTableCommand taskLogicalPlan,
StatementBase taskParsedStmt) throws Exception {
stmtExecutor = new StmtExecutor(taskConnectContext, taskParsedStmt);
+ if (isCanceled.get()) {
Review Comment:
[P1] Linearize executor publication with cancellation. The worker
plain-writes `stmtExecutor` and can read `isCanceled=false`; a canceller that
writes true afterward has no happens-before edge for the preceding plain field
and may still read null, so both sides miss their only replay point and the
worker enters blocking plan/execute uncanceled. `cancelAndAwaitTasks()` then
waits on `terminal` without a bound, turning the rewrite timeout/interrupt path
into a hang. Publish/check under one monitor (or an atomic handoff) and add a
latch test for this window.
##########
fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectPoolMgr.java:
##########
@@ -57,6 +59,12 @@ public int registerConnection(ConnectContext ctx) {
@Override
public void unregisterConnection(ConnectContext ctx) {
+ // Reject new publications first, then signal the active query before
waiting for an admitted
+ // GetFlightInfo publisher. Waiting before cancellation can deadlock
KILL CONNECTION behind the
+ // publisher whose query must be canceled in order to leave
publication.
+ ctx.sealFlightSqlDeferredExecutors();
+ ctx.cancelQuery(new Status(TStatusCode.CANCELLED, "arrow flight
connection closed"));
Review Comment:
[P1] Guarantee local teardown when cancellation signaling fails. For a
forwarded query, `StmtExecutor.cancel()` turns a master RPC/journal failure
into `RuntimeException`; because this call is before every cleanup stage and
outside a `finally`, CloseSession/token expiry/KILL can leave the sealed
context registered, its Arrow allocator/results and transaction open, and its
counter/token entries stale. Treat cancel as best-effort (capture/log the
failure) and guarantee publisher drain, channel/txn close, and pool bookkeeping
before surfacing it; add a forwarded-cancel failure test.
##########
fe/fe-core/src/main/java/org/apache/doris/qe/MasterOpExecutor.java:
##########
@@ -59,12 +62,25 @@ public MasterOpExecutor(ConnectContext ctx) {
@Override
public void execute() throws Exception {
+ synchronized (executionAdmissionLock) {
+ if (cancellationRequested) {
+ ctx.getState().setError("forward operation cancelled");
+ return;
+ }
+ executionStarted = true;
+ }
super.execute();
waitOnReplaying();
}
@Override
public void cancel() throws Exception {
+ synchronized (executionAdmissionLock) {
Review Comment:
[P1] Keep the cancel reply separate from the forwarded statement result.
Once execution is admitted, this call can run concurrently with
`super.execute()`, and both inherited methods assign the same unsynchronized
`result` field. If the statement stores its packet/error and the cancel RPC
then stores a status-0 packet-less acknowledgement, `waitOnReplaying()` and the
later `StmtExecutor` getters observe the wrong response. Use a local cancel
result (including its journal ID) and reserve `result` for the statement; add
barrier tests for both completion orders.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -556,19 +557,145 @@ public void alterJob(AlterJobCommand alterJobCommand)
throws AnalysisException,
@Override
public void updateJobStatus(JobStatus status) throws JobException {
+ AbstractStreamingTask taskToCancel = null;
+ boolean waitForTask = JobStatus.PAUSED.equals(status);
lock.writeLock().lock();
try {
+ if ((JobStatus.PAUSED.equals(status) ||
JobStatus.STOPPED.equals(status))
+ && status != getJobStatus()) {
+ taskToCancel = runningStreamTask;
+ }
+ JobStatus previousStatus = getJobStatus();
super.updateJobStatus(status);
- if (JobStatus.PAUSED.equals(getJobStatus())) {
- clearRunningStreamTask(status);
+ if (previousStatus != getJobStatus()) {
+ statusEpoch++;
}
if (isFinalStatus()) {
Env.getCurrentGlobalTransactionMgr().getCallbackFactory().removeCallback(getJobId());
}
log.info("Streaming insert job {} update status to {}",
getJobId(), getJobStatus());
+ } catch (RuntimeException | JobException e) {
+ if (taskToCancel != null) {
+ runningStreamTask = taskToCancel;
+ }
+ throw e;
+ } finally {
+ lock.writeLock().unlock();
+ }
+ if (taskToCancel != null && waitForTask) {
+ // The task owner can need this job's write lock while finishing
transaction callbacks.
+ // Cancel and wait only after publishing the status and releasing
the job lock.
+ taskToCancel.cancel(waitForTask);
+ }
+ }
+
+ /**
+ * Applies a user-requested status transition as one job-lock operation.
The reason is published before
+ * PAUSED/PENDING becomes visible to the scheduler, while transition
validation happens before either field
+ * is changed. Blocking cancellation completion and reader release stay
outside the job lock.
+ */
+ public void updateManualJobStatus(JobStatus status, FailureReason reason)
throws JobException {
+ AbstractStreamingTask taskToWait = null;
+ AbstractStreamingTask taskToRelease = null;
+ JobStatus publishedStatus = JobStatus.RUNNING.equals(status) ?
JobStatus.PENDING : status;
+ lock.writeLock().lock();
+ try {
+ validateManualStatusTransition(status);
+ resetFailureInfo(reason);
+ if (JobStatus.PAUSED.equals(status) && runningStreamTask != null) {
Review Comment:
[P1] Preserve the remote owner on manual STOP. `taskToRelease` is
PAUSE-only, so STOP leaves `readerReleased=true`; once `/api/writeRecords` has
been admitted, the FE execution owner is already finished and this path clears
the task after a purely local `cancel()`. That cancel intentionally sends no
reader RPC, and `/api/close` runs only from later DROP cleanup, which now also
lost the exact snapshot-phase `runningBackendId`. A stopped job can therefore
keep polling/loading on the BE and a later DROP can route close to the wrong
backend. Please complete an acknowledged terminal `/api/close` (or equivalent)
on the exact runtime BE before discarding the task, with a nonzero-backend STOP
test.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingMultiTblTask.java:
##########
@@ -390,6 +402,39 @@ public void releaseRemoteReader() {
}
}
+ /** Wait for the BE to acknowledge reader release before allowing a
successor to reuse the source. */
+ boolean releaseRemoteReaderAndWait() {
+ if (runningBackendId <= 0) {
+ return true;
+ }
+ Backend backend =
Env.getCurrentSystemInfo().getBackend(runningBackendId);
+ if (backend == null) {
Review Comment:
[P1] Give an authoritatively removed backend a terminal handoff. Returning
false here retains the canceled predecessor; after RESUME the job is PENDING,
but `StreamingInsertJob.isReadyForScheduling()` requires `runningStreamTask ==
null`, so the timer never even creates the scheduler task that could retry
release or reach `resolveBoundBackend()` on another healthy BE. A
decommissioned/lost BE therefore wedges this job forever. Define a safe
removed-owner policy (the task-ID fences already reject zombie callbacks) and
test PAUSE/RESUME rebinding after backend removal.
--
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]
