Fixed topic test build

Author: stanislav-shchetinin

src/client/persqueue_public/ut/basic_usage_ut.cpp

@@ -71,7 +71,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
             std::visit(TOverloaded {
                 [&](TReadSessionEvent::TDataReceivedEvent& event) {
                     for (auto& message: event.GetMessages()) {
-                        TString sourceId = message.GetMessageGroupId();
+                        std::string sourceId = message.GetMessageGroupId();
                         ui32 seqNo = message.GetSeqNo();
                         UNIT_ASSERT_VALUES_EQUAL(readMessageCount + 1, seqNo);
                         ++readMessageCount;
@@ -190,16 +190,16 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
         auto readSession = client.CreateReadSession(readSettings);
 
         auto event = readSession->GetEvent(true);
-        UNIT_ASSERT(event.Defined());
+        UNIT_ASSERT(event.has_value());
 
         auto& createPartitionStream = std::get<TReadSessionEvent::TCreatePartitionStreamEvent>(*event);
         createPartitionStream.Confirm();
 
         UNIT_CHECK_GENERATED_EXCEPTION(readSession->GetEvent(true, 0), TContractViolation);
-        UNIT_CHECK_GENERATED_EXCEPTION(readSession->GetEvents(true, Nothing(), 0), TContractViolation);
+        UNIT_CHECK_GENERATED_EXCEPTION(readSession->GetEvents(true, std::nullopt, 0), TContractViolation);
 
         event = readSession->GetEvent(true, 1);
-        UNIT_ASSERT(event.Defined());
+        UNIT_ASSERT(event.has_value());
 
         auto& dataReceived = std::get<TReadSessionEvent::TDataReceivedEvent>(*event);
         dataReceived.Commit();
@@ -380,7 +380,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
             std::visit(TOverloaded {
                 [&](TReadSessionEvent::TDataReceivedEvent& event) {
                     for (auto& message: event.GetMessages()) {
-                        TString sourceId = message.GetMessageGroupId();
+                        std::string sourceId = message.GetMessageGroupId();
                         ui32 seqNo = message.GetSeqNo();
                         UNIT_ASSERT_VALUES_EQUAL(readMessageCount + 1, seqNo);
                         ++readMessageCount;
@@ -447,7 +447,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
     public:
         TBrokenCredentialsProvider() {}
         virtual ~TBrokenCredentialsProvider() {}
-        TStringType GetAuthInfo() const {
+        std::string GetAuthInfo() const {
             ythrow yexception() << "exception during creation";
             return "";
         }
@@ -463,7 +463,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
             return std::make_shared<TBrokenCredentialsProvider>();
         }
 
-        virtual TStringType GetClientIdentity() const {
+        virtual std::string GetClientIdentity() const {
             return "abacaba";
         }
     };

src/client/persqueue_public/ut/read_session_ut.cpp

@@ -316,7 +316,7 @@ Y_UNIT_TEST_SUITE(RetryPolicy) {
                         Cerr << "===Data event\n";
                         auto& clusterName = event.GetPartitionStream()->GetCluster();
                         for (auto& message: event.GetMessages()) {
-                            TString sourceId = message.GetMessageGroupId();
+                            auto sourceId = TString{message.GetMessageGroupId()};
                             ui32 seqNo = message.GetSeqNo();
                             if (sourceId == sourceId1) {
                                 UNIT_ASSERT_VALUES_EQUAL(seqNo, seqNoByClusterSrc1[clusterName] + 1);

src/client/persqueue_public/ut/retry_policy_ut.cpp

@@ -51,7 +51,7 @@ namespace NKikimr::NPersQueueTests {
         std::optional<ui32> partitionGroup,
         std::optional<TString> codec,
         std::optional<bool> reconnectOnFailure,
-        THashMap<TString, TString> sessionMeta
+        std::unordered_map<std::string, std::string> sessionMeta
     ) {
         auto settings = TWriteSessionSettings().Path(topic).MessageGroupId(sourceId);
         if (partitionGroup) settings.PartitionGroupId(*partitionGroup);
@@ -84,7 +84,7 @@ namespace NKikimr::NPersQueueTests {
             auto future = reader->WaitEvent();
             future.Wait(timeout);
 
-            TMaybe<NYdb::NPersQueue::TReadSessionEvent::TEvent> event = reader->GetEvent(false, 1);
+            std::optional<NYdb::NPersQueue::TReadSessionEvent::TEvent> event = reader->GetEvent(false, 1);
             if (!event)
                 return {};
             if (auto dataEvent = std::get_if<NYdb::NPersQueue::TReadSessionEvent::TDataReceivedEvent>(&*event)) {

src/client/persqueue_public/ut/ut_utils/data_plane_helpers.cpp

@@ -4,6 +4,8 @@
 #include <ydb-cpp-sdk/client/table/table.h>
 #include <src/client/persqueue_public/persqueue.h>
 
+#include <util/generic/hash.h>
+
 namespace NKikimr::NPersQueueTests {
 
     std::shared_ptr<NYdb::NPersQueue::IWriteSession> CreateWriter(
@@ -34,7 +36,7 @@ namespace NKikimr::NPersQueueTests {
         std::optional<ui32> partitionGroup = {},
         std::optional<TString> codec = {},
         std::optional<bool> reconnectOnFailure = {},
-        THashMap<TString, TString> sessionMeta = {}
+        std::unordered_map<std::string, std::string> sessionMeta = {}
     );
 
     std::shared_ptr<NYdb::NPersQueue::IReadSession> CreateReader(

src/client/persqueue_public/ut/ut_utils/data_plane_helpers.h

@@ -88,7 +88,7 @@ class SDKTestSetup {
         }
     }
 
-    static TString GetTestTopic() {
+    static std::string GetTestTopic() {
         return "test-topic";
     }
 

src/client/persqueue_public/ut/ut_utils/sdk_test_setup.h

@@ -45,7 +45,7 @@ class TPersQueueYdbSdkTestSetup : public ::NPersQueue::SDKTestSetup {
             NYdb::TDriverConfig cfg;
             cfg.SetEndpoint(TStringBuilder() << "localhost:" << Server.GrpcPort);
             cfg.SetDatabase("/Root");
-            cfg.SetLog(CreateLogBackend("cerr", ELogPriority::TLOG_DEBUG));
+            cfg.SetLog(std::unique_ptr<TLogBackend>(CreateLogBackend("cerr", ELogPriority::TLOG_DEBUG).Release()));
             Driver = MakeHolder<NYdb::TDriver>(cfg);
         }
         return *Driver;
@@ -168,7 +168,7 @@ struct TYDBClientEventLoop : public ::NPersQueue::IClientEventLoop {
                     }
                     Y_ABORT_UNLESS(continueToken);
 
-                    TMaybe<ui64> seqNo = Nothing();
+                    std::optional<ui64> seqNo = std::nullopt;
                     if (!AutoSeqNo) {
                         seqNo = acknowledgeableMessage.SequenceNumber;
                         Log << TLOG_INFO << "[" << sourceId << "] Write messages with sequence numbers "
@@ -377,7 +377,7 @@ class TYdbPqTestExecutor : public IAsyncExecutor {
         Stop = true;
         Thread.Join();
     }
-    void PostImpl(TVector<TFunction>&& fs) override {
+    void PostImpl(std::vector<TFunction>&& fs) override {
         for (auto& f : fs) {
             TasksQueue.Enqueue(std::move(f));
         }

src/client/persqueue_public/ut/ut_utils/ut_utils.h

@@ -381,7 +381,7 @@ class TBaseSessionEventsQueue : public ISignalable {
 
     void WaitEventsImpl() { // Assumes that we're under lock. Posteffect: HasEventsImpl() is true.
         while (!HasEventsImpl()) {
-            std::unique_lock<std::mutex> lk(Mutex, std::defer_lock);
+            std::unique_lock<std::mutex> lk(Mutex, std::adopt_lock);
             CondVar.wait(lk);
         }
     }

src/client/topic/impl/common.h

@@ -16,6 +16,8 @@
 #include <library/cpp/threading/future/future.h>
 #include <library/cpp/threading/future/async.h>
 
+#include <util/stream/zlib.h>
+
 #include <future>
 
 namespace NYdb::NTopic::NTests {
@@ -99,7 +101,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
         NYdb::TDriverConfig cfg;
         cfg.SetEndpoint(TStringBuilder() << "invalid:" << setup.GetServer().GrpcPort);
         cfg.SetDatabase("/Invalid");
-        cfg.SetLog(CreateLogBackend("cerr", ELogPriority::TLOG_DEBUG));
+        cfg.SetLog(std::unique_ptr<TLogBackend>(CreateLogBackend("cerr", ELogPriority::TLOG_DEBUG).Release()));
         auto driver = NYdb::TDriver(cfg);
 
         {
@@ -113,13 +115,13 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
             auto writeSession = client.CreateWriteSession(writeSettings);
 
             auto event = writeSession->GetEvent(true);
-            UNIT_ASSERT(event.Defined() && std::holds_alternative<TSessionClosedEvent>(event.GetRef()));
+            UNIT_ASSERT(event && std::holds_alternative<TSessionClosedEvent>(event.value()));
         }
 
         {
             auto settings = TTopicClientSettings()
                 .Database({"/Root"})
-                .DiscoveryEndpoint({TStringBuilder() << "localhost:" << setup.GetServer().GrpcPort});
+                .DiscoveryEndpoint("localhost:" + std::to_string(setup.GetServer().GrpcPort));
 
             TTopicClient client(driver, settings);
 
@@ -130,7 +132,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
             auto writeSession = client.CreateWriteSession(writeSettings);
 
             auto event = writeSession->GetEvent(true);
-            UNIT_ASSERT(event.Defined() && !std::holds_alternative<TSessionClosedEvent>(event.GetRef()));
+            UNIT_ASSERT(event && !std::holds_alternative<TSessionClosedEvent>(event.value()));
         }
     }
 
@@ -171,13 +173,13 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
             auto readSession = client.CreateReadSession(readSettings);
 
             auto event = readSession->GetEvent(true);
-            UNIT_ASSERT(event.Defined());
+            UNIT_ASSERT(event.has_value());
 
             auto& startPartitionSession = std::get<TReadSessionEvent::TStartPartitionSessionEvent>(*event);
             startPartitionSession.Confirm();
 
             event = readSession->GetEvent(true);
-            UNIT_ASSERT(event.Defined());
+            UNIT_ASSERT(event.has_value());
 
             auto& dataReceived = std::get<TReadSessionEvent::TDataReceivedEvent>(*event);
             dataReceived.Commit();
@@ -234,16 +236,16 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
         auto readSession = client.CreateReadSession(readSettings);
 
         auto event = readSession->GetEvent(true);
-        UNIT_ASSERT(event.Defined());
+        UNIT_ASSERT(event.has_value());
 
         auto& startPartitionSession = std::get<TReadSessionEvent::TStartPartitionSessionEvent>(*event);
         startPartitionSession.Confirm();
 
         UNIT_CHECK_GENERATED_EXCEPTION(readSession->GetEvent(true, 0), TContractViolation);
-        UNIT_CHECK_GENERATED_EXCEPTION(readSession->GetEvents(true, Nothing(), 0), TContractViolation);
+        UNIT_CHECK_GENERATED_EXCEPTION(readSession->GetEvents(true, std::nullopt, 0), TContractViolation);
 
         event = readSession->GetEvent(true, 1);
-        UNIT_ASSERT(event.Defined());
+        UNIT_ASSERT(event.has_value());
 
         auto& dataReceived = std::get<TReadSessionEvent::TDataReceivedEvent>(*event);
         dataReceived.Commit();
@@ -335,7 +337,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
         auto description = result.GetConsumerDescription();
         UNIT_ASSERT(description.GetPartitions().size() == 1);
         auto stats = description.GetPartitions().front().GetPartitionConsumerStats();
-        UNIT_ASSERT(stats.Defined());
+        UNIT_ASSERT(stats.has_value());
         UNIT_ASSERT(stats->GetCommittedOffset() == 50);
     }
 
@@ -695,7 +697,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
         auto description = result.GetTopicDescription();
         UNIT_ASSERT(description.GetPartitions().size() == 1);
         auto stats = description.GetPartitions().front().GetPartitionStats();
-        UNIT_ASSERT(stats.Defined());
+        UNIT_ASSERT(stats.has_value());
         UNIT_ASSERT_VALUES_EQUAL(stats->GetEndOffset(), count);
 
     }
@@ -779,7 +781,7 @@ Y_UNIT_TEST_SUITE(BasicUsage) {
             std::visit(TOverloaded {
                 [&](TReadSessionEvent::TDataReceivedEvent& event) {
                     for (auto& message: event.GetMessages()) {
-                        TString sourceId = message.GetMessageGroupId();
+                        std::string sourceId = message.GetMessageGroupId();
                         ui32 seqNo = message.GetSeqNo();
                         UNIT_ASSERT_VALUES_EQUAL(readMessageCount + 1, seqNo);
                         ++readMessageCount;
@@ -829,11 +831,11 @@ Y_UNIT_TEST_SUITE(TSettingsValidation) {
 
         auto client = setup.MakeClient();
         ui64 producerIndex = 0u;
-        auto runTest = [&](TString producer, TString msgGroup, const TMaybe<bool>& useDedup, bool useSeqNo, EExpectedTestResult result) ->bool
+        auto runTest = [&](TString producer, TString msgGroup, const std::optional<bool>& useDedup, bool useSeqNo, EExpectedTestResult result) ->bool
         {
             TWriteSessionSettings writeSettings;
             writeSettings.Path(setup.GetTopicPath()).Codec(NTopic::ECodec::RAW);
-            TString useDedupStr = useDedup.Defined() ? ToString(*useDedup) : "<unset>";
+            TString useDedupStr = useDedup.has_value() ? ToString(*useDedup) : "<unset>";
             if (producer) {
                 producer += ToString(producerIndex);
             }
@@ -848,7 +850,7 @@ Y_UNIT_TEST_SUITE(TSettingsValidation) {
                  << useDedupStr << ", manual SeqNo: " << useSeqNo << Endl;
 
             try {
-                if (useDedup.Defined()) {
+                if (useDedup.has_value()) {
                     writeSettings.DeduplicationEnabled(useDedup);
                 }
                 auto session = client.CreateWriteSession(writeSettings);
@@ -857,12 +859,12 @@ Y_UNIT_TEST_SUITE(TSettingsValidation) {
                 ui64 written = 0;
                 while (written < 10) {
                     auto event = session->GetEvent(true);
-                    if (std::holds_alternative<TSessionClosedEvent>(event.GetRef())) {
+                    if (std::holds_alternative<TSessionClosedEvent>(event.value())) {
                         auto closed = std::get<TSessionClosedEvent>(*event);
                         Cerr << "Session failed with error: " << closed.DebugString() << Endl;
                         UNIT_ASSERT(result == EExpectedTestResult::FAIL_ON_RPC);
                         return false;
-                    } else if (std::holds_alternative<TWriteSessionEvent::TReadyToAcceptEvent>(event.GetRef())) {
+                    } else if (std::holds_alternative<TWriteSessionEvent::TReadyToAcceptEvent>(event.value())) {
                         token = std::move(std::get<TWriteSessionEvent::TReadyToAcceptEvent>(*event).ContinuationToken);
                         if (useSeqNo) {
                             session->Write(std::move(*token), "data", seqNo++);
@@ -948,10 +950,10 @@ Y_UNIT_TEST_SUITE(TSettingsValidation) {
 
         auto readSession = client.CreateReadSession(readSettings);
         auto event = readSession->GetEvent(true);
-        UNIT_ASSERT(event.Defined());
+        UNIT_ASSERT(event.has_value());
 
         auto& closeEvent = std::get<NYdb::NTopic::TSessionClosedEvent>(*event);
-        UNIT_ASSERT(closeEvent.DebugString().Contains("Too small max memory usage"));
+        UNIT_ASSERT(closeEvent.DebugString().contains("Too small max memory usage"));
     }
 
 } // Y_UNIT_TEST_SUITE(TSettingsValidation)

src/client/topic/ut/basic_usage_ut.cpp

@@ -144,7 +144,7 @@ namespace NYdb::NTopic::NTests {
 
                         UNIT_ASSERT_GT(consumerStats->GetLastReadOffset(), 0);
                         UNIT_ASSERT_GT(consumerStats->GetCommittedOffset(), 0);
-                        UNIT_ASSERT_GE(consumerStats->GetReadSessionId(), 0);
+                        UNIT_ASSERT_GE(TString{consumerStats->GetReadSessionId()}, 0);
                         UNIT_ASSERT_VALUES_EQUAL(consumerStats->GetReaderName(), "");
                     } else {
                         UNIT_ASSERT_VALUES_EQUAL(stats->GetStartOffset(), 0);
@@ -296,29 +296,29 @@ namespace NYdb::NTopic::NTests {
 
                 // Event 1: start partition session
                 {
-                    TMaybe<TReadSessionEvent::TEvent> event = readSession->GetEvent(true);
+                    std::optional<TReadSessionEvent::TEvent> event = readSession->GetEvent(true);
                     UNIT_ASSERT(event);
-                    auto startPartitionSession = std::get_if<TReadSessionEvent::TStartPartitionSessionEvent>(event.Get());
+                    auto startPartitionSession = std::get_if<TReadSessionEvent::TStartPartitionSessionEvent>(&event.value());
                     UNIT_ASSERT_C(startPartitionSession, DebugString(*event));
 
                     startPartitionSession->Confirm();
                 }
 
                 // Event 2: data received
                 {
-                    TMaybe<TReadSessionEvent::TEvent> event = readSession->GetEvent(true);
+                    std::optional<TReadSessionEvent::TEvent> event = readSession->GetEvent(true);
                     UNIT_ASSERT(event);
-                    auto dataReceived = std::get_if<TReadSessionEvent::TDataReceivedEvent>(event.Get());
+                    auto dataReceived = std::get_if<TReadSessionEvent::TDataReceivedEvent>(&event.value());
                     UNIT_ASSERT_C(dataReceived, DebugString(*event));
 
                     dataReceived->Commit();
                 }
 
                 // Event 3: commit acknowledgement
                 {
-                    TMaybe<TReadSessionEvent::TEvent> event = readSession->GetEvent(true);
+                    std::optional<TReadSessionEvent::TEvent> event = readSession->GetEvent(true);
                     UNIT_ASSERT(event);
-                    auto commitOffsetAck = std::get_if<TReadSessionEvent::TCommitOffsetAcknowledgementEvent>(event.Get());
+                    auto commitOffsetAck = std::get_if<TReadSessionEvent::TCommitOffsetAcknowledgementEvent>(&event.value());
 
                     UNIT_ASSERT_C(commitOffsetAck, DebugString(*event));
 
@@ -363,7 +363,7 @@ namespace NYdb::NTopic::NTests {
             if (allowUpdateRow) {
                 acl.AddAccess(NACLib::EAccessType::Allow, NACLib::UpdateRow, authToken);
             }
-            setup.GetServer().AnnoyingClient->ModifyACL("/Root", TEST_TOPIC, acl.SerializeAsString());
+            setup.GetServer().AnnoyingClient->ModifyACL("/Root", TString{TEST_TOPIC}, acl.SerializeAsString());
 
             return client.DescribePartition(existingTopic ? TEST_TOPIC : "bad-topic", testPartitionId, settings).GetValueSync();
         }
@@ -409,7 +409,7 @@ namespace NYdb::NTopic::NTests {
                 if (resultStatus == EStatus::SUCCESS) {
                     auto& p = result.GetPartitionDescription().GetPartition();
                     UNIT_ASSERT(p.GetActive());
-                    UNIT_ASSERT(p.GetPartitionLocation().Defined());
+                    UNIT_ASSERT(p.GetPartitionLocation().has_value());
                 }
             }
         }