DqPqReadActorTest has been fixed (#6785)

2 files changed, 10 insertions, 4 deletions

diff --git a/ydb/library/yql/providers/common/ut_helpers/dq_fake_ca.cpp b/ydb/library/yql/providers/common/ut_helpers/dq_fake_ca.cpp
index 8356bfd4df..17430bc285 100644
--- a/ydb/library/yql/providers/common/ut_helpers/dq_fake_ca.cpp
+++ b/ydb/library/yql/providers/common/ut_helpers/dq_fake_ca.cpp
@@ -55,7 +55,7 @@ void TFakeActor::InitAsyncInput(IDqComputeActorAsyncInput* dqAsyncInput, IActor*
     DqAsyncInputAsActor = dqAsyncInputAsActor;
 }
 
-void TFakeActor::Terminate() {
+void TFakeActor::Terminate(std::shared_ptr<std::atomic<bool>> done) {
     if (DqAsyncInputActorId) {
         DqAsyncInput->PassAway();
 
@@ -71,6 +71,7 @@ void TFakeActor::Terminate() {
         DqAsyncOutput = nullptr;
         DqAsyncOutputAsActor = nullptr;
     }
+    done->store(true);
 }
 
 TFakeActor::TAsyncOutputCallbacks& TFakeActor::GetAsyncOutputCallbacks() {
@@ -101,9 +102,14 @@ TFakeCASetup::TFakeCASetup()
 }
 
 TFakeCASetup::~TFakeCASetup() {
-    Execute([](TFakeActor& actor) {
-        actor.Terminate();
+    auto shouldStop = std::make_shared<std::atomic<bool>>(); 
+    Execute([shouldStop](TFakeActor& actor) {
+        actor.Terminate(shouldStop);
     });
+
+    while (!*shouldStop) {
+        Sleep(TDuration::MilliSeconds(200));
+    }
 }
 
 void TFakeCASetup::AsyncOutputWrite(const TWriteValueProducer valueProducer, TMaybe<NDqProto::TCheckpoint> checkpoint, bool finish) {
diff --git a/ydb/library/yql/providers/common/ut_helpers/dq_fake_ca.h b/ydb/library/yql/providers/common/ut_helpers/dq_fake_ca.h
index 8d50cff70b..c74c9bb201 100644
--- a/ydb/library/yql/providers/common/ut_helpers/dq_fake_ca.h
+++ b/ydb/library/yql/providers/common/ut_helpers/dq_fake_ca.h
@@ -119,7 +119,7 @@ public:
 
     void InitAsyncOutput(IDqComputeActorAsyncOutput* dqAsyncOutput, IActor* dqAsyncOutputAsActor);
     void InitAsyncInput(IDqComputeActorAsyncInput* dqAsyncInput, IActor* dqAsyncInputAsActor);
-    void Terminate();
+    void Terminate(std::shared_ptr<std::atomic<bool>> done);
 
     TAsyncOutputCallbacks& GetAsyncOutputCallbacks();
     NKikimr::NMiniKQL::THolderFactory& GetHolderFactory();