[+] wait before reading

Author: Alek5andr-Kotov

ydb/core/tx/datashard/datashard_ut_change_exchange.cpp

@@ -1206,6 +1206,7 @@ Y_UNIT_TEST_SUITE(Cdc) {
         {
             TTestYdsEnv env(tableDesc, streamDesc);
 
+            DBGTRACE_LOG("exec SQL");
             for (const auto& query : queries) {
                 ExecSQL(env.GetServer(), env.GetEdgeActor(), query);
             }
@@ -1246,6 +1247,8 @@ Y_UNIT_TEST_SUITE(Cdc) {
 
             // get records
             {
+                WaitForDataRecords(client, shardIt);
+
                 auto res = client.GetRecords(shardIt).ExtractValueSync();
                 UNIT_ASSERT_C(res.IsSuccess(), res.GetIssues().ToString());
                 UNIT_ASSERT_VALUES_EQUAL(res.GetResult().records().size(), records.size());
@@ -1267,6 +1270,19 @@ Y_UNIT_TEST_SUITE(Cdc) {
             }
         }
 
+        static void WaitForDataRecords(TDataStreamsClient& client, const TString& shardIt) {
+            int n = 0;
+            for (; n < 100; ++n) {
+                auto res = client.GetRecords(shardIt).ExtractValueSync();
+                UNIT_ASSERT_C(res.IsSuccess(), res.GetIssues().ToString());
+                if (res.GetResult().records().size()) {
+                    break;
+                }
+                Sleep(TDuration::MilliSeconds(100));
+            }
+            UNIT_ASSERT_VALUES_UNEQUAL(n, 100);
+        }
+
         static void Write(const TShardedTableOptions& tableDesc, const TCdcStream& streamDesc) {
             TTestYdsEnv env(tableDesc, streamDesc);