github-actions[bot] commented on code in PR #66914:
URL: https://github.com/apache/doris/pull/66914#discussion_r3835770119
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -556,19 +556,33 @@ 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 {
- super.updateJobStatus(status);
- if (JobStatus.PAUSED.equals(getJobStatus())) {
- clearRunningStreamTask(status);
+ if ((JobStatus.PAUSED.equals(status) ||
JobStatus.STOPPED.equals(status))
+ && status != getJobStatus()) {
+ taskToCancel = runningStreamTask;
+ runningStreamTask = null;
}
+ super.updateJobStatus(status);
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) {
+ // 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);
Review Comment:
[P2] Preserve canceled-task accounting on this detached path. Because line
566 clears `runningStreamTask` before `super.updateJobStatus()`, the inherited
call to this class's `cancelAllTasks()` sees null and returns before the
active-state check and `canceledTaskCount.incrementAndGet()`. This direct
cancel therefore stops a RUNNING/PENDING task without updating the serialized,
TVF-visible `CanceledTaskCount`. Capture the detached task's active state and
increment exactly once while keeping the blocking wait outside the lock; please
cover both active and already-terminal PAUSE/STOP cases so the count is neither
lost nor doubled.
##########
fe/fe-core/src/main/java/org/apache/doris/service/arrowflight/sessions/FlightSqlConnectPoolMgr.java:
##########
@@ -57,6 +57,9 @@ public int registerConnection(ConnectContext ctx) {
@Override
public void unregisterConnection(ConnectContext ctx) {
+ // Reject new publications and wait for in-flight GetFlightInfo
publication to decide its
+ // outcome before destroying either local Arrow results or deferred
coordinators.
+ ctx.sealAndCloseFlightSqlDeferredExecutors();
Review Comment:
[P1] Keep the result channel alive through the last admitted publisher.
`sealAndCloseFlightSqlDeferredExecutors()` returns immediately while a
GetFlightInfo publisher is active, but teardown then closes the channel here. A
local result can already have allocated its vectors and be paused just before
`resultCache.put()`; this close's one-time invalidation misses it (and
allocator close reports outstanding bytes), then the publisher inserts the
entry afterward. The seal correctly makes GetFlightInfo fail, but its failure
path never invalidates the channel again, so the late entry's off-heap buffers
lose deterministic cleanup. Defer channel close until the final publisher
exits, or make a losing publisher remove and close every result it created, and
cover the allocation-before-put race with a latch test.
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -556,19 +556,33 @@ 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 {
- super.updateJobStatus(status);
- if (JobStatus.PAUSED.equals(getJobStatus())) {
- clearRunningStreamTask(status);
+ if ((JobStatus.PAUSED.equals(status) ||
JobStatus.STOPPED.equals(status))
+ && status != getJobStatus()) {
+ taskToCancel = runningStreamTask;
+ runningStreamTask = null;
Review Comment:
[P1] Keep the task visible until cancellation is published. This clears
`runningStreamTask` while holding the job lock, then unlocks and only calls
`taskToCancel.cancel(...)` at line 584. A `beforeCommitted()` callback already
queued on that lock can acquire it in between and immediately dereference
`runningStreamTask` at line 1351, producing an NPE instead of the intended
canceled-transaction failure. This is a follow-on to the prior lock/self-wait
fix: publish cancellation without waiting under the lock (or make the callback
capture/reject a detached task), and clear the exact task conditionally
afterward. Please add a latch test that lets `beforeCommitted()` enter after
detachment but before the out-of-lock cancel call.
--
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]
