fix(cluster): fix the possibility of connection leak when the cluster state is broken.

[zhang2014]

I hereby agree to the terms of the CLA available at: https://docs.databend.com/dev/policies/cla/

Summary

fix(cluster): fix the possibility of connection leak when the cluster state is broken.

Tests

• Unit Test
• Logic Test
• Benchmark Test
• No Test - Almost impossible state (caused by bugs during the development process)

Type of change

• Bug Fix (non-breaking change which fixes an issue)
• New Feature (non-breaking change which adds functionality)
• Breaking Change (fix or feature that could cause existing functionality not to work as expected)
• Documentation Update
• Refactoring
• Performance Improvement
• Other (please describe):

---

This change is 

[what-the-diff]

PR Summary

• Enhanced Error Handling in Pipeline Building
  The error management process in building distributed pipelines was overhauled. This change propagates errors more effectively, leading to smoother operations and quicker identification of issues.

• Updated "on finished query" Calls
  Updates were made to the calls to the "on finished query", enabling improved error reporting. This means any issues that occur will be documented more effectively, facilitating quicker fixes.

• Adaptable Shutdown Query Function
  The function to shut down queries was updated to accept possible errors. This recognition enables the shutdown program to identify the cause for any failures, leading to more effective troubleshooting.

• Flight Actions Sturdier Against Errors
  The error handling when initiating query environments and fragments, as well as starting prepared queries have been improved. It enables the system to report errors back to the Data Exchange Manager, again aiding troubleshooting and ensuring smoother operations.

• Logging and Shutdown Logic Adjustments in Exchange Manager
  Minor adjustments made to the logging and shutdown procedures in the Exchange Manager reflect the new robust error handling strategies. These adjustments improve the robustness of the system overall.

[dqhl76]

LGTM!