fix unstable mpp fail tests (#5709)

ref #5095, close #5708

dbms/src/DataStreams/TiRemoteBlockInputStream.h

@@ -241,7 +241,6 @@ class TiRemoteBlockInputStream : public IProfilingBlockInputStream
     void readSuffixImpl() override
     {
         LOG_FMT_DEBUG(log, "finish read {} rows from remote", total_rows);
-        remote_reader->close();
     }
 
     void appendInfo(FmtBuffer & buffer) const override

dbms/src/Flash/Mpp/MPPReceiverSet.cpp

@@ -39,10 +39,16 @@ ExchangeReceiverPtr MPPReceiverSet::getExchangeReceiver(const String & executor_
 void MPPReceiverSet::cancel()
 {
     for (auto & it : exchange_receiver_map)
-    {
         it.second->cancel();
-    }
     for (auto & cop_reader : coprocessor_readers)
         cop_reader->cancel();
 }
+
+void MPPReceiverSet::close()
+{
+    for (auto & it : exchange_receiver_map)
+        it.second->close();
+    for (auto & cop_reader : coprocessor_readers)
+        cop_reader->close();
+}
 } // namespace DB

dbms/src/Flash/Mpp/MPPReceiverSet.h

@@ -29,6 +29,7 @@ class MPPReceiverSet
     void addCoprocessorReader(const CoprocessorReaderPtr & coprocessor_reader);
     ExchangeReceiverPtr getExchangeReceiver(const String & executor_id) const;
     void cancel();
+    void close();
 
 private:
     /// two kinds of receiver in MPP

dbms/src/Flash/Mpp/MPPTask.cpp

@@ -412,7 +412,11 @@ void MPPTask::runImpl()
         while (from->read())
             continue;
 
+        // finish DataStream
         from->readSuffix();
+        // finish receiver
+        receiver_set->close();
+        // finish MPPTunnel
         finishWrite();
 
         const auto & return_statistics = mpp_task_statistics.collectRuntimeStatistics();