fix: Fix flaky test BroadcastTest.endToEndWithMultipleWriteNodes (#26461)

Summary:

Fix flaky stress test buck2 test //github/presto-trunk/presto-native-execution/presto_cpp/main/operators/tests:presto_operators_test -- 'BroadcastTest.endToEndWithMultipleWriteNodes/zstd' --stress-runs 100

BroadcastTestParam::TearDown() inherited from OperatorTestBase https://www.internalfb.com/code/fbsource/[37f968940832]/fbcode/velox/exec/tests/utils/OperatorTestBase.cpp?lines=153 ,

We have an existing exchange source lifecycle protection for LocalExchangeSource https://fburl.com/code/aoz55crb.  But our BroadcastTest..endToEndWithMultipleWriteNodes uses BroadcastExchangeSource which lacks this protection.

Since it's just a flaky test, the most straightforward way is just inject a latency to handle a race condition in async cleanup.

Differential Revision: D85685549

Author: duxiao1212

presto-native-execution/presto_cpp/main/operators/tests/BroadcastTest.cpp

@@ -66,6 +66,15 @@ class BroadcastTest : public exec::test::OperatorTestBase,
         BroadcastExchangeSource::createExchangeSource);
   }
 
+  void TearDown() override {
+    velox::exec::test::waitForAllTasksToBeDeleted();
+    // Fix flaky stress-test: Add a delay to allow async destructors to fully complete.
+    std::this_thread::sleep_for(std::chrono::milliseconds(100));
+    pool_.reset();
+    rootPool_.reset();
+    resetMemory();
+  }
+
   std::unique_ptr<VectorSerde::Options> getVectorSerdeOptions(
       common::CompressionKind compressionKind) {
     std::unique_ptr<VectorSerde::Options> options = std::make_unique<
@@ -303,9 +312,9 @@ TEST_P(BroadcastTest, endToEndWithMultipleWriteNodes) {
 
   // Read back result.
   auto [taskCursorReadNode, broadcastReadResults] = executeBroadcastRead(
-      asRowType(dataVector[0]->type()),
-      tempDirectoryPath->getPath(),
-      broadcastFilePaths);
+    asRowType(dataVector[0]->type()),
+    tempDirectoryPath->getPath(),
+    broadcastFilePaths);
 
   // Validate BroadcastExchange reads back output of both writes.
   velox::exec::test::assertEqualResults(dataVector, broadcastReadResults);