De-flake MultiFragmentTest::exchangeStatsOnFailure (#11094)

Summary:

The test is flaky for two independent reasons:
a) The producer task is writing a large payload may take more than 3sec (timeout value) to finish. The solution is decrease the payload. 

b) The second reason for the flakiness is a bug in ExchangeClient. ExchangeClient::close() was not correctly closing all the ExchangeSources (it keeps shared_ptr of ExchangeSource in a queue). As a result, Task::Close()-->ExchangeClient::close() doesn't cleanup the memory if there is another shared_ptr of same ExchangeClient is alive. Another ExchangeClient ptr can be alive as  ExchangeClient::request() passes shared_from_this() to a future.

Reviewed By: xiaoxmeng

Differential Revision: D63374267

velox/exec/ExchangeClient.cpp

@@ -70,13 +70,17 @@ void ExchangeClient::noMoreRemoteTasks() {
 
 void ExchangeClient::close() {
   std::vector<std::shared_ptr<ExchangeSource>> sources;
+  std::queue<ProducingSource> producingSources;
+  std::queue<std::shared_ptr<ExchangeSource>> emptySources;
   {
     std::lock_guard<std::mutex> l(queue_->mutex());
     if (closed_) {
       return;
     }
     closed_ = true;
     sources = std::move(sources_);
+    producingSources = std::move(producingSources_);
+    emptySources = std::move(emptySources_);
   }
 
   // Outside of mutex.

velox/exec/tests/MultiFragmentTest.cpp

@@ -1891,7 +1891,7 @@ DEBUG_ONLY_TEST_F(MultiFragmentTest, exchangeStatsOnFailure) {
   });
 
   auto producerPlan = PlanBuilder()
-                          .values({data}, false, 100)
+                          .values({data}, false, 30)
                           .partitionedOutput({}, 1)
                           .planNode();