924060929 commented on code in PR #66914:
URL: https://github.com/apache/doris/pull/66914#discussion_r3838731786
##########
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:
这条评论涉及通用查询、任务或 Flight/Streaming 生命周期，不属于本 PR 仅处理 Hudi/Iceberg 资源关闭与泄露的范围。当前
head 已撤回对应旁支改动，本 PR 忽略该问题。
##########
fe/fe-core/src/test/java/org/apache/doris/service/arrowflight/DorisFlightSqlProducerTest.java:
##########
@@ -207,4 +209,69 @@ public void
testGetFlightInfoFinalizesDeferredExecutorWhenSchemaFetchFails() thr
producer.close();
}
}
+
+ @Test
+ public void
testGetFlightInfoFailsWhenTeardownDrainsRegisteredQueryBeforePublication()
throws Exception {
+ assertTeardownPreventsFlightInfoPublication(true);
+ }
+
+ @Test
+ public void
testGetFlightInfoFailsWhenTeardownSealsBeforeQueryRegistration() throws
Exception {
+ assertTeardownPreventsFlightInfoPublication(false);
+ }
+
+ @Test
+ public void testPublicationCompletionAtomicallyObservesTerminalSeal() {
+ ConnectContext ctx = new ConnectContext();
+ StmtExecutor deferred = Mockito.mock(StmtExecutor.class);
+ Assert.assertTrue(ctx.beginFlightSqlResultPublication());
+ Assert.assertTrue(ctx.addFlightSqlDeferredExecutor(deferred));
+
+ ctx.sealAndCloseFlightSqlDeferredExecutors();
Review Comment:
这条评论涉及通用查询、任务或 Flight/Streaming 生命周期，不属于本 PR 仅处理 Hudi/Iceberg 资源关闭与泄露的范围。当前
head 已撤回对应旁支改动，本 PR 忽略该问题。
##########
fe/fe-core/src/main/java/org/apache/doris/qe/ConnectContext.java:
##########
@@ -1005,21 +1005,96 @@ public void clear() {
// 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:
这条评论涉及通用查询、任务或 Flight/Streaming 生命周期，不属于本 PR 仅处理 Hudi/Iceberg 资源关闭与泄露的范围。当前
head 已撤回对应旁支改动，本 PR 忽略该问题。
##########
fe/fe-core/src/test/java/org/apache/doris/qe/ConnectContextTest.java:
##########
@@ -811,4 +816,37 @@ public void
testCloseFlightSqlDeferredExecutorsFinalizesRemainingWhenOneFails()
Mockito.verify(failing, Mockito.times(1)).finalizeArrowFlightQuery();
Mockito.verify(healthy, Mockito.times(1)).finalizeArrowFlightQuery();
}
+
+ @Test
+ public void testSessionTeardownRejectsLateDeferredExecutorRegistration() {
+ ConnectContext ctx = new ConnectContext();
+ StmtExecutor late = Mockito.mock(StmtExecutor.class);
+
+ ctx.sealAndCloseFlightSqlDeferredExecutors();
+
+ Assert.assertFalse("an executor registered after the final teardown
drain must not become unreachable",
+ ctx.addFlightSqlDeferredExecutor(late));
+ ctx.closeFlightSqlDeferredExecutors();
+ Mockito.verifyNoInteractions(late);
+ }
+
+ @Test
+ public void testSessionSealWaitsForAdmittedResultPublisher() throws
Exception {
+ ConnectContext ctx = new ConnectContext();
+ Assert.assertTrue(ctx.beginFlightSqlResultPublication());
+ ExecutorService executor = Executors.newSingleThreadExecutor();
+ try {
+ Future<?> teardown =
executor.submit(ctx::sealAndCloseFlightSqlDeferredExecutors);
+ try {
+ teardown.get(100, TimeUnit.MILLISECONDS);
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]
