fix(cluster): close flight channel in drop

src/query/service/src/api/rpc/exchange/exchange_sink_writer.rs

@@ -58,6 +58,12 @@ impl ExchangeWriterSink {
     }
 }
 
+impl Drop for ExchangeWriterSink {
+    fn drop(&mut self) {
+        self.flight_sender.close();
+    }
+}
+
 #[async_trait::async_trait]
 impl AsyncSink for ExchangeWriterSink {
     const NAME: &'static str = "ExchangeWriterSink";
@@ -85,9 +91,11 @@ impl AsyncSink for ExchangeWriterSink {
         for packet in serialize_meta.packet {
             bytes += packet.bytes_size();
             if let Err(error) = self.flight_sender.send(packet).await {
-                if error.code() == ErrorCode::ABORTED_QUERY {
-                    return Ok(true);
-                }
+                self.flight_sender.close();
+
+                // if error.code() == ErrorCode::ABORTED_QUERY {
+                //     return Ok(true);
+                // }
 
                 return Err(error);
             }

src/query/service/src/api/rpc/exchange/exchange_source_reader.rs

@@ -62,6 +62,12 @@ impl ExchangeSourceReader {
     }
 }
 
+impl Drop for ExchangeSourceReader {
+    fn drop(&mut self) {
+        self.flight_receiver.close();
+    }
+}
+
 #[async_trait::async_trait]
 impl Processor for ExchangeSourceReader {
     fn name(&self) -> String {

src/query/service/src/api/rpc/flight_client.rs

@@ -225,10 +225,6 @@ impl FlightSender {
         FlightSender { tx }
     }
 
-    pub fn is_closed(&self) -> bool {
-        self.tx.is_closed()
-    }
-
     #[async_backtrace::framed]
     pub async fn send(&self, data: DataPacket) -> Result<()> {
         if let Err(_cause) = self.tx.send(Ok(FlightData::try_from(data)?)).await {