Fixed topic tests from ydb repo

Author: Gazizonoki

src/client/topic/impl/common.h

@@ -383,6 +383,7 @@ class TBaseSessionEventsQueue : public ISignalable {
         while (!HasEventsImpl()) {
             std::unique_lock<std::mutex> lk(Mutex, std::adopt_lock);
             CondVar.wait(lk);
+            lk.release();
         }
     }
 

src/client/topic/impl/read_session.cpp

@@ -230,14 +230,19 @@ void TReadSession::UpdateOffsets(const NTable::TTransaction& tx)
 
     Y_ABORT_UNLESS(!topics.empty());
 
-    auto result = Client->UpdateOffsetsInTransaction(tx,
-                                                     topics,
-                                                     Settings.ConsumerName_,
-                                                     {}).GetValueSync();
-    Y_ABORT_UNLESS(!result.IsTransportError());
-
-    if (!result.IsSuccess()) {
-        ythrow yexception() << "error on update offsets: " << result;
+    while (true) {
+        auto result = Client->UpdateOffsetsInTransaction(tx,
+                                                         topics,
+                                                         Settings.ConsumerName_,
+                                                         {}).GetValueSync();
+        Y_ABORT_UNLESS(!result.IsTransportError());
+        if (result.GetStatus() != EStatus::SESSION_BUSY) {
+            if (!result.IsSuccess()) {
+                ythrow yexception() << "error on update offsets: " << result;
+            }
+            break;
+        }
+        Sleep(TDuration::MilliSeconds(1));
     }
 
     OffsetRanges.erase(std::make_pair(sessionId, txId));
@@ -309,9 +314,10 @@ bool TReadSession::Close(TDuration timeout) {
         issues.AddIssue(TStringBuilder() << "Session was closed after waiting " << timeout);
         EventsQueue->Close(TSessionClosedEvent(EStatus::TIMEOUT, std::move(issues)), deferred);
     }
-
-    std::lock_guard guard(Lock);
-    Aborting = true; // Set abort flag for doing nothing on destructor.
+    {
+        std::lock_guard guard(Lock);
+        Aborting = true; // Set abort flag for doing nothing on destructor.
+    }
     return result;
 }
 

src/client/topic/impl/read_session_impl.ipp

@@ -2184,22 +2184,23 @@ TReadSessionEventsQueue<UseMigrationProtocol>::GetEvents(bool block, std::option
     std::vector<TReadSessionEventInfo<UseMigrationProtocol>> eventInfos;
     const size_t maxCount = maxEventsCount ? *maxEventsCount : std::numeric_limits<size_t>::max();
     TUserRetrievedEventsInfoAccumulator<UseMigrationProtocol> accumulator;
+    {
+        std::lock_guard<std::mutex> guard(TParent::Mutex);
+        eventInfos.reserve(Min(TParent::Events.size() + TParent::CloseEvent.has_value(), maxCount));
+        do {
+            if (block) {
+                TParent::WaitEventsImpl();
+            }
 
-    std::lock_guard<std::mutex> guard(TParent::Mutex);
-    eventInfos.reserve(Min(TParent::Events.size() + TParent::CloseEvent.has_value(), maxCount));
-    do {
-        if (block) {
-            TParent::WaitEventsImpl();
-        }
-
-        while (TParent::HasEventsImpl() && eventInfos.size() < maxCount && maxByteSize > 0) {
-            TReadSessionEventInfo<UseMigrationProtocol> event = GetEventImpl(maxByteSize, accumulator);
-            eventInfos.emplace_back(std::move(event));
-            if (eventInfos.back().IsSessionClosedEvent()) {
-                break;
+            while (TParent::HasEventsImpl() && eventInfos.size() < maxCount && maxByteSize > 0) {
+                TReadSessionEventInfo<UseMigrationProtocol> event = GetEventImpl(maxByteSize, accumulator);
+                eventInfos.emplace_back(std::move(event));
+                if (eventInfos.back().IsSessionClosedEvent()) {
+                    break;
+                }
             }
-        }
-    } while (block && eventInfos.empty());
+        } while (block && eventInfos.empty());
+    }
 
     accumulator.OnUserRetrievedEvent();
 
@@ -2222,18 +2223,19 @@ TReadSessionEventsQueue<UseMigrationProtocol>::GetEvent(bool block, size_t maxBy
 
     std::optional<TReadSessionEventInfo<UseMigrationProtocol>> eventInfo;
     TUserRetrievedEventsInfoAccumulator<UseMigrationProtocol> accumulator;
+    {
+        std::lock_guard<std::mutex> guard(TParent::Mutex);
+        do {
+            if (block) {
+                TParent::WaitEventsImpl();
+            }
 
-    std::lock_guard<std::mutex> guard(TParent::Mutex);
-    do {
-        if (block) {
-            TParent::WaitEventsImpl();
-        }
-
-        if (TParent::HasEventsImpl()) {
-            eventInfo = GetEventImpl(maxByteSize, accumulator);
-        }
+            if (TParent::HasEventsImpl()) {
+                eventInfo = GetEventImpl(maxByteSize, accumulator);
+            }
 
-    } while (block && !eventInfo);
+        } while (block && !eventInfo);
+    }
 
     accumulator.OnUserRetrievedEvent();
 

src/client/topic/ut/topic_to_table_ut.cpp

@@ -2124,32 +2124,32 @@ Y_UNIT_TEST_F(ReadRuleGeneration, TFixture)
     NotifySchemeShard({.EnablePQConfigTransactionsAtSchemeShard = false});
 
     // Users have created their own topic on it
-    CreateTopic(TEST_TOPIC);
+    CreateTopic(TString{TEST_TOPIC});
 
     // And they wrote their messages into it
-    WriteToTopic(TEST_TOPIC, TEST_MESSAGE_GROUP_ID, "message-1");
-    WriteToTopic(TEST_TOPIC, TEST_MESSAGE_GROUP_ID, "message-2");
-    WriteToTopic(TEST_TOPIC, TEST_MESSAGE_GROUP_ID, "message-3");
+    WriteToTopic(TString{TEST_TOPIC}, TEST_MESSAGE_GROUP_ID, "message-1");
+    WriteToTopic(TString{TEST_TOPIC}, TEST_MESSAGE_GROUP_ID, "message-2");
+    WriteToTopic(TString{TEST_TOPIC}, TEST_MESSAGE_GROUP_ID, "message-3");
 
     // And he had a consumer
-    AddConsumer(TEST_TOPIC, {"consumer-1"});
+    AddConsumer(TString{TEST_TOPIC}, {"consumer-1"});
 
     // We read messages from the topic and committed offsets
-    auto messages = ReadFromTopic(TEST_TOPIC, "consumer-1", TDuration::Seconds(2));
+    auto messages = ReadFromTopic(TString{TEST_TOPIC}, "consumer-1", TDuration::Seconds(2));
     UNIT_ASSERT_VALUES_EQUAL(messages.size(), 3);
-    CloseTopicReadSession(TEST_TOPIC, "consumer-1");
+    CloseTopicReadSession(TString{TEST_TOPIC}, "consumer-1");
 
     // And then the Logbroker team turned on the feature flag
     NotifySchemeShard({.EnablePQConfigTransactionsAtSchemeShard = true});
 
     // Users continued to write to the topic
-    WriteToTopic(TEST_TOPIC, TEST_MESSAGE_GROUP_ID, "message-4");
+    WriteToTopic(TString{TEST_TOPIC}, TEST_MESSAGE_GROUP_ID, "message-4");
 
     // Users have added new consumers
-    AddConsumer(TEST_TOPIC, {"consumer-2"});
+    AddConsumer(TString{TEST_TOPIC}, {"consumer-2"});
 
     // And they wanted to continue reading their messages
-    messages = ReadFromTopic(TEST_TOPIC, "consumer-1", TDuration::Seconds(2));
+    messages = ReadFromTopic(TString{TEST_TOPIC}, "consumer-1", TDuration::Seconds(2));
     UNIT_ASSERT_VALUES_EQUAL(messages.size(), 1);
 }