Tests TxUsage::Sinks_*_WriteToTopicAndTable_* (#23813)

1 files changed, 19 insertions, 11 deletions

diff --git a/ydb/public/sdk/cpp/src/client/topic/ut/topic_to_table_ut.cpp b/ydb/public/sdk/cpp/src/client/topic/ut/topic_to_table_ut.cpp
index 3f8020447bc..d756034abdb 100644
--- a/ydb/public/sdk/cpp/src/client/topic/ut/topic_to_table_ut.cpp
+++ b/ydb/public/sdk/cpp/src/client/topic/ut/topic_to_table_ut.cpp
@@ -484,25 +484,33 @@ std::vector<TResultSet> TFixture::TTableSession::Execute(const std::string& quer
                                                          bool commit,
                                                          const TParams& params)
 {
-    auto txTable = dynamic_cast<NTable::TTransaction*>(tx);
-    auto txControl = NTable::TTxControl::Tx(*txTable).CommitTx(commit);
-
-    auto result = Session_.ExecuteDataQuery(query, txControl, params).GetValueSync();
-    UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
+    while (true) {
+        auto txTable = dynamic_cast<NTable::TTransaction*>(tx);
+        auto txControl = NTable::TTxControl::Tx(*txTable).CommitTx(commit);
 
-    return std::move(result).ExtractResultSets();
+        auto result = Session_.ExecuteDataQuery(query, txControl, params).GetValueSync();
+        if (result.GetStatus() != EStatus::SESSION_BUSY) {
+            UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
+            return std::move(result).ExtractResultSets();
+        }
+        std::this_thread::sleep_for(100ms);
+    }
 }
 
 TFixture::ISession::TExecuteInTxResult TFixture::TTableSession::ExecuteInTx(const std::string& query,
                                                                             bool commit,
                                                                             const TParams& params)
 {
-    auto txControl = NTable::TTxControl::BeginTx().CommitTx(commit);
-
-    auto result = Session_.ExecuteDataQuery(query, txControl, params).GetValueSync();
-    UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
+    while (true) {
+        auto txControl = NTable::TTxControl::BeginTx().CommitTx(commit);
 
-    return {std::move(result).ExtractResultSets(), std::make_unique<NTable::TTransaction>(*result.GetTransaction())};
+        auto result = Session_.ExecuteDataQuery(query, txControl, params).GetValueSync();
+        if (result.GetStatus() != EStatus::SESSION_BUSY) {
+            UNIT_ASSERT_C(result.IsSuccess(), result.GetIssues().ToString());
+            return {std::move(result).ExtractResultSets(), std::make_unique<NTable::TTransaction>(*result.GetTransaction())};
+        }
+        std::this_thread::sleep_for(100ms);
+    }
 }
 
 std::unique_ptr<TTransactionBase> TFixture::TTableSession::BeginTx()