Moved commit "Fixed the test WriteToTopic_Demo_42" from ydb repo

Author: Alek5andr-Kotov

src/client/topic/ut/topic_to_table_ut.cpp

@@ -181,6 +181,10 @@ class TFixture : public NUnitTest::TBaseFixture {
 
     NTable::TDataQueryResult ExecuteDataQuery(NTable::TSession session, const TString& query, const NTable::TTxControl& control);
 
+    void Read_Exactly_N_Messages_From_Topic(const TString& topicPath,
+                                        const TString& consumerName,
+                                        size_t count);
+
 private:
     template<class E>
     E ReadEvent(TTopicReadSessionPtr reader, NTable::TTransaction& tx);
@@ -531,10 +535,13 @@ Y_UNIT_TEST_F(TwoSessionOneConsumer, TFixture)
 Y_UNIT_TEST_F(Offsets_Cannot_Be_Promoted_When_Reading_In_A_Transaction, TFixture)
 {
     WriteMessage("message");
+
     auto session = CreateTableSession();
     auto tx = BeginTx(session);
+
     auto reader = CreateReader();
     StartPartitionSession(reader, tx, 0);
+
     UNIT_ASSERT_EXCEPTION(ReadMessage(reader, {.Tx = tx, .CommitOffsets = true}), yexception);
 }
 
@@ -1002,6 +1009,18 @@ void TFixture::RestartLongTxService()
     }
 }
 
+void TFixture::Read_Exactly_N_Messages_From_Topic(const TString& topicPath,
+                                                  const TString& consumerName,
+                                                  size_t limit)
+{
+    size_t count = 0;
+    while (count < limit) {
+        auto messages = ReadFromTopic(topicPath, consumerName, TDuration::Seconds(2));
+        count += messages.size();
+    }
+    UNIT_ASSERT_VALUES_EQUAL(count, limit);
+}
+
 Y_UNIT_TEST_F(WriteToTopic_Demo_1, TFixture)
 {
     CreateTopic("topic_A");
@@ -2166,8 +2185,7 @@ Y_UNIT_TEST_F(WriteToTopic_Demo_40, TFixture)
 
     CommitTx(tx, EStatus::SUCCESS);
 
-    auto messages = ReadFromTopic("topic_A", TEST_CONSUMER, TDuration::Seconds(60));
-    UNIT_ASSERT_VALUES_EQUAL(messages.size(), 100);
+    Read_Exactly_N_Messages_From_Topic("topic_A", TEST_CONSUMER, 100);
 }
 
 Y_UNIT_TEST_F(WriteToTopic_Demo_41, TFixture)
@@ -2202,8 +2220,7 @@ Y_UNIT_TEST_F(WriteToTopic_Demo_42, TFixture)
 
     CommitTx(tx, EStatus::SUCCESS);
 
-    auto messages = ReadFromTopic("topic_A", TEST_CONSUMER, TDuration::Seconds(2));
-    UNIT_ASSERT_VALUES_EQUAL(messages.size(), 100);
+    Read_Exactly_N_Messages_From_Topic("topic_A", TEST_CONSUMER, 100);
 }
 
 Y_UNIT_TEST_F(WriteToTopic_Demo_43, TFixture)
@@ -2221,8 +2238,7 @@ Y_UNIT_TEST_F(WriteToTopic_Demo_43, TFixture)
 
     ExecuteDataQuery(tableSession, "SELECT 1", NTable::TTxControl::Tx(tx).CommitTx(true));
 
-    auto messages = ReadFromTopic("topic_A", TEST_CONSUMER, TDuration::Seconds(60));
-    UNIT_ASSERT_VALUES_EQUAL(messages.size(), 100);
+    Read_Exactly_N_Messages_From_Topic("topic_A", TEST_CONSUMER, 100);
 }
 
 Y_UNIT_TEST_F(WriteToTopic_Demo_44, TFixture)
@@ -2246,8 +2262,7 @@ Y_UNIT_TEST_F(WriteToTopic_Demo_44, TFixture)
 
     ExecuteDataQuery(tableSession, "SELECT 2", NTable::TTxControl::Tx(tx).CommitTx(true));
 
-    messages = ReadFromTopic("topic_A", TEST_CONSUMER, TDuration::Seconds(60));
-    UNIT_ASSERT_VALUES_EQUAL(messages.size(), 100);
+    Read_Exactly_N_Messages_From_Topic("topic_A", TEST_CONSUMER, 100);
 }
 
 }