[Bug](pipeline) try fix the exchange sink buffer result error (apache…

…#27052)
924060929 · Nov 16, 2023 · 7ef1f7e · 7ef1f7e
1 parent 02f3762
commit 7ef1f7e
Showing 1 changed file with 16 additions and 11 deletions.
diff --git a/be/src/pipeline/task_scheduler.cpp b/be/src/pipeline/task_scheduler.cpp
@@ -333,24 +333,29 @@ void TaskScheduler::_do_work(size_t index) {
 void TaskScheduler::_try_close_task(PipelineTask* task, PipelineTaskState state,
                                     Status exec_status) {
     auto status = task->try_close(exec_status);
-    if (!status.ok() && state != PipelineTaskState::CANCELED) {
-        // Call `close` if `try_close` failed to make sure allocated resources are released
-        static_cast<void>(task->close(exec_status));
+    auto cancel = [&]() {
         task->query_context()->cancel(true, status.to_string(),
                                       Status::Cancelled(status.to_string()));
         state = PipelineTaskState::CANCELED;
-    } else if (task->is_pending_finish()) {
-        task->set_state(PipelineTaskState::PENDING_FINISH);
-        static_cast<void>(_blocked_task_scheduler->add_blocked_task(task));
-        return;
-    } else {
+    };
+
+    auto try_close_failed = !status.ok() && state != PipelineTaskState::CANCELED;
+    if (try_close_failed) {
+        cancel();
+        // Call `close` if `try_close` failed to make sure allocated resources are released
+        static_cast<void>(task->close(exec_status));
+    } else if (!task->is_pending_finish()) {
         status = task->close(exec_status);
         if (!status.ok() && state != PipelineTaskState::CANCELED) {
-            task->query_context()->cancel(true, status.to_string(),
-                                          Status::Cancelled(status.to_string()));
-            state = PipelineTaskState::CANCELED;
+            cancel();
         }
     }
+
+    if (task->is_pending_finish()) {
+        task->set_state(PipelineTaskState::PENDING_FINISH);
+        static_cast<void>(_blocked_task_scheduler->add_blocked_task(task));
+        return;
+    }
     task->set_state(state);
     task->set_close_pipeline_time();
     task->release_dependency();