924060929 commented on code in PR #66913:
URL: https://github.com/apache/doris/pull/66913#discussion_r3838732300
##########
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:
这条评论涉及通用查询、任务、Streaming、Flight、Hive 或其他非本 PR 生命周期问题，不属于本 PR 仅处理 Hudi/Iceberg
资源关闭与泄露的范围。当前 head 已撤回对应旁支改动，本 PR 忽略该问题。
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/AbstractStreamingTask.java:
##########
@@ -94,35 +98,76 @@ public long getRunningBackendId() {
}
public void execute() throws JobException {
- while (retryCount <= MAX_RETRY) {
- try {
- before();
- run();
- onSuccess();
- return;
- } catch (Exception e) {
+ synchronized (executionCompletion) {
+ executionStarted = true;
+ executionOwner = Thread.currentThread();
+ }
+ try {
+ while (retryCount <= MAX_RETRY) {
+ Exception attemptFailure = null;
+ try {
+ before();
+ run();
+ } catch (Exception e) {
+ attemptFailure = e;
+ } finally {
+ // Only the scheduler worker that created this attempt's
ConnectContext may tear it down.
+ // A cancelling thread waits for this handoff instead of
racing before() and clearing fields
+ // while planning is still publishing them.
+ try {
+ closeOrReleaseResources();
+ } catch (RuntimeException cleanupFailure) {
Review Comment:
这条评论涉及通用查询、任务、Streaming、Flight、Hive 或其他非本 PR 生命周期问题，不属于本 PR 仅处理 Hudi/Iceberg
资源关闭与泄露的范围。当前 head 已撤回对应旁支改动，本 PR 忽略该问题。
##########
fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java:
##########
@@ -984,21 +984,96 @@ public PlSqlOperation getPlSqlOperation() {
// with "Split source X is released". These executors are finalized when
the next query starts
// on this connection, or when the connection is torn down. See #62259.
private final List<StmtExecutor> flightSqlDeferredExecutors = new
ArrayList<>();
+ private boolean flightSqlDeferredExecutorsSealed;
+ private int flightSqlResultPublishers;
- public void addFlightSqlDeferredExecutor(StmtExecutor executor) {
+ public boolean addFlightSqlDeferredExecutor(StmtExecutor executor) {
synchronized (flightSqlDeferredExecutors) {
+ if (flightSqlDeferredExecutorsSealed) {
+ return false;
+ }
flightSqlDeferredExecutors.add(executor);
+ return true;
+ }
+ }
+
+ /** Linearizes GetFlightInfo publication with the terminal session seal. */
+ public boolean canPublishFlightSqlResult() {
+ synchronized (flightSqlDeferredExecutors) {
+ return !flightSqlDeferredExecutorsSealed;
+ }
+ }
+
+ public boolean beginFlightSqlResultPublication() {
+ synchronized (flightSqlDeferredExecutors) {
+ if (flightSqlDeferredExecutorsSealed) {
+ return false;
+ }
+ flightSqlResultPublishers++;
+ return true;
+ }
+ }
+
+ public boolean endFlightSqlResultPublication() {
+ List<StmtExecutor> toClose = null;
+ boolean published;
+ synchronized (flightSqlDeferredExecutors) {
+ published = !flightSqlDeferredExecutorsSealed;
+ if (--flightSqlResultPublishers == 0 &&
flightSqlDeferredExecutorsSealed) {
+ toClose = drainFlightSqlDeferredExecutors();
+ flightSqlDeferredExecutors.notifyAll();
+ }
}
+ finalizeFlightSqlDeferredExecutors(toClose);
+ return published;
}
public void closeFlightSqlDeferredExecutors() {
- List<StmtExecutor> toClose;
+ closeFlightSqlDeferredExecutors(false);
+ }
+
+ /** Prevents a session teardown race from accepting an executor after the
final drain. */
+ public void sealAndCloseFlightSqlDeferredExecutors() {
+ closeFlightSqlDeferredExecutors(true);
+ }
+
+ private void closeFlightSqlDeferredExecutors(boolean seal) {
+ List<StmtExecutor> toClose = null;
synchronized (flightSqlDeferredExecutors) {
- if (flightSqlDeferredExecutors.isEmpty()) {
- return;
+ if (seal) {
+ flightSqlDeferredExecutorsSealed = true;
+ // The result channel is destroyed immediately after this
method returns. Wait until every
+ // admitted publisher has either committed or observed the
seal, so a losing local-result
+ // publisher cannot insert Arrow buffers after the channel's
one-time invalidation.
+ boolean interrupted = false;
+ while (flightSqlResultPublishers != 0) {
Review Comment:
这条评论涉及通用查询、任务、Streaming、Flight、Hive 或其他非本 PR 生命周期问题，不属于本 PR 仅处理 Hudi/Iceberg
资源关闭与泄露的范围。当前 head 已撤回对应旁支改动，本 PR 忽略该问题。
##########
fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java:
##########
@@ -556,19 +557,141 @@ 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 validated manual transition while reason, status, and
cancellation share one 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) {
+ needRebuildReader = true;
+ taskToRelease = runningStreamTask;
+ }
+ if (JobStatus.PAUSED.equals(status) ||
JobStatus.STOPPED.equals(status)) {
+ taskToWait = runningStreamTask;
+ if (taskToWait != null) {
+ // Linearize manual termination with callbacks before
publishing the new job status.
+ // Task-specific RPCs and waits remain outside the job
lock.
+ taskToWait.publishCancellation();
+ }
+ }
+ // RESUME must re-enter PENDING so the scheduler creates a
successor before publishing RUNNING.
+ super.updateJobStatus(publishedStatus);
+ statusEpoch++;
+ if (isFinalStatus()) {
+
Env.getCurrentGlobalTransactionMgr().getCallbackFactory().removeCallback(getJobId());
+ }
+ log.info("Streaming insert job {} manually updated status to {}",
getJobId(), getJobStatus());
} finally {
lock.writeLock().unlock();
}
+ boolean readerReleased = true;
+ if (taskToRelease != null) {
+ readerReleased = !(taskToRelease instanceof StreamingMultiTblTask)
+ || ((StreamingMultiTblTask)
taskToRelease).releaseRemoteReaderAndWait();
+ if (!(taskToRelease instanceof StreamingMultiTblTask)) {
+ taskToRelease.releaseRemoteReader();
+ }
+ }
+ if (taskToWait != null) {
+ taskToWait.cancel(true);
+ if (readerReleased && taskToWait.canHandoffAfterCancellation()) {
Review Comment:
这条评论涉及通用查询、任务、Streaming、Flight、Hive 或其他非本 PR 生命周期问题，不属于本 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:
这条评论涉及通用查询、任务、Streaming、Flight、Hive 或其他非本 PR 生命周期问题，不属于本 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]
