924060929 commented on code in PR #66914:
URL: https://github.com/apache/doris/pull/66914#discussion_r3838732247
##########
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:
这条评论涉及通用查询、任务或 Flight/Streaming 生命周期，不属于本 PR 仅处理 Hudi/Iceberg 资源关闭与泄露的范围。当前
head 已撤回对应旁支改动，本 PR 忽略该问题。
##########
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:
这条评论涉及通用查询、任务或 Flight/Streaming 生命周期，不属于本 PR 仅处理 Hudi/Iceberg 资源关闭与泄露的范围。当前
head 已撤回对应旁支改动，本 PR 忽略该问题。
##########
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:
这条评论涉及通用查询、任务或 Flight/Streaming 生命周期，不属于本 PR 仅处理 Hudi/Iceberg 资源关闭与泄露的范围。当前
head 已撤回对应旁支改动，本 PR 忽略该问题。
##########
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:
这条评论涉及通用查询、任务或 Flight/Streaming 生命周期，不属于本 PR 仅处理 Hudi/Iceberg 资源关闭与泄露的范围。当前
head 已撤回对应旁支改动，本 PR 忽略该问题。
##########
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:
这条评论涉及通用查询、任务或 Flight/Streaming 生命周期，不属于本 PR 仅处理 Hudi/Iceberg 资源关闭与泄露的范围。当前
head 已撤回对应旁支改动，本 PR 忽略该问题。
##########
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:
这条评论涉及通用查询、任务或 Flight/Streaming 生命周期，不属于本 PR 仅处理 Hudi/Iceberg 资源关闭与泄露的范围。当前
head 已撤回对应旁支改动，本 PR 忽略该问题。
--
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]
