Improvements of #25843 (#26164)

nshestakov · nshestakov · commit 1fabba3c7a20 · 2025-10-02T12:08:54.000Z
diff --git a/ydb/core/persqueue/public/fetcher/fetch_request_actor.cpp b/ydb/core/persqueue/public/fetcher/fetch_request_actor.cpp
diff --git a/ydb/core/persqueue/public/fetcher/fetch_request_actor.h b/ydb/core/persqueue/public/fetcher/fetch_request_actor.h
@@ -29,8 +29,8 @@ struct TFetchRequestSettings {
     TString Database;
     TString Consumer;
     TVector<TPartitionFetchRequest> Partitions;
-    ui64 MaxWaitTimeMs;
-    ui64 TotalMaxBytes;
+    ui64 MaxWaitTimeMs = 0;
+    ui64 TotalMaxBytes = 0;
 
     bool RuPerRequest = false;
     ui64 RequestId = 0;
diff --git a/ydb/core/persqueue/public/fetcher/fetch_request_ut.cpp b/ydb/core/persqueue/public/fetcher/fetch_request_ut.cpp
@@ -4,12 +4,12 @@
 #include <ydb/core/persqueue/events/internal.h>
 #include <ydb/core/testlib/tenant_runtime.h>
 #include <ydb/core/tx/scheme_board/cache.h>
-#include <ydb/public/sdk/cpp/src/client/persqueue_public/ut/ut_utils/ut_utils.h>
+#include <ydb/public/sdk/cpp/src/client/topic/ut/ut_utils/topic_sdk_test_setup.h>
 
 namespace NKikimr::NPQ {
 using namespace NPersQueue;
 //using namespace NYdb::NTopic;
-using namespace NYdb::NPersQueue::NTests;
+using namespace NYdb::NTopic::NTests;
 
 
 Y_UNIT_TEST_SUITE(TFetchRequestTests) {
@@ -26,7 +26,7 @@ Y_UNIT_TEST_SUITE(TFetchRequestTests) {
     }
 
     Y_UNIT_TEST(HappyWay) {
-        auto setup = std::make_shared<TPersQueueYdbSdkTestSetup>(TEST_CASE_NAME);
+        auto setup = std::make_shared<TTopicSdkTestSetup>(TEST_CASE_NAME);
         setup->GetServer().EnableLogs(
                 { NKikimrServices::TX_PROXY_SCHEME_CACHE, NKikimrServices::PQ_FETCH_REQUEST },
                 NActors::NLog::PRI_DEBUG
@@ -36,28 +36,18 @@ Y_UNIT_TEST_SUITE(TFetchRequestTests) {
 
         ui32 totalPartitions = 5;
         setup->CreateTopic("topic1", "dc1", totalPartitions);
-        setup->CreateTopic("topic2", "dc1", totalPartitions);
-        auto pqClient = setup->GetPersQueueClient();
-        auto settings1 = TWriteSessionSettings().Path("topic1").MessageGroupId("src-id").PartitionGroupId(2);
-        auto settings2 = TWriteSessionSettings().Path("topic2").MessageGroupId("src-id").PartitionGroupId(4);
-        auto ws1 = pqClient.CreateSimpleBlockingWriteSession(settings1);
-        auto ws2 = pqClient.CreateSimpleBlockingWriteSession(settings2);
-        
-        ws1->Write("Data 1-1");
-        ws1->Write("Data 1-2");
-        ws1->Write("Data 1-3");
-        
-        ws2->Write("Data 2-1");
-        ws2->Write("Data 2-2");
-
-        ws1->Close();
-        ws2->Close();
+        setup->Write("/Root/topic1", "Data 1-1", 1);
+        setup->Write("/Root/topic1", "Data 1-2", 1);
+        setup->Write("/Root/topic1", "Data 1-3", 1);
 
+        setup->CreateTopic("topic2", "dc1", totalPartitions);
+        setup->Write("/Root/topic1", "Data 2-1", 3);
+        setup->Write("/Root/topic1", "Data 2-2", 3);
 
         auto edgeId = runtime.AllocateEdgeActor();
-        TPartitionFetchRequest p1{"Root/PQ/rt3.dc1--topic1", 1, 1, 10000};
-        TPartitionFetchRequest p2{"Root/PQ/rt3.dc1--topic2", 3, 0, 10000};
-        TPartitionFetchRequest pbad{"Root/PQ/rt3.dc1--topic2", 2, 1, 10000};
+        TPartitionFetchRequest p1{"/Root/topic1", 1, 1, 10000};
+        TPartitionFetchRequest p2{"/Root/topic2", 3, 0, 10000};
+        TPartitionFetchRequest pbad{"/Root/topic2", 2, 1, 10000};
 
         TFetchRequestSettings settings{{}, NKikimr::NPQ::CLIENTID_WITHOUT_CONSUMER, {p1, p2, pbad}, 1000, 1000};
         auto fetchId = runtime.Register(CreatePQFetchRequestActor(settings, MakeSchemeCacheID(), edgeId));
@@ -68,25 +58,143 @@ Y_UNIT_TEST_SUITE(TFetchRequestTests) {
         UNIT_ASSERT_C(ev->Status == Ydb::StatusIds::SUCCESS, ev->Message);
         Cerr << "Got event: " << ev->Response.DebugString() << Endl;
         UNIT_ASSERT_VALUES_EQUAL(ev->Response.PartResultSize(), 3);
-        for (const auto& part : ev->Response.GetPartResult()) {
-            if (part.GetTopic().Contains("topic1")) {
-                UNIT_ASSERT(part.GetReadResult().GetErrorCode() == NPersQueue::NErrorCode::EErrorCode::OK);
-                UNIT_ASSERT_VALUES_EQUAL(part.GetPartition(), 1);
-                UNIT_ASSERT_VALUES_EQUAL(part.GetReadResult().GetResult(0).GetOffset(), 1);
-            } else {
-                UNIT_ASSERT(part.GetTopic().Contains("topic2"));
-                if (part.GetPartition() == 2) {
-                    UNIT_ASSERT(part.GetReadResult().GetErrorCode() != NPersQueue::NErrorCode::EErrorCode::OK);
-                } else {
-                    UNIT_ASSERT_VALUES_EQUAL(part.GetPartition(), 3);
-                    UNIT_ASSERT_VALUES_EQUAL(part.GetReadResult().GetResult(0).GetOffset(), 0);
-                }
-            }
+
+        {
+            auto& result = ev->Response.GetPartResult(0);
+            UNIT_ASSERT_VALUES_EQUAL(result.GetTopic(), "/Root/topic1");
+            UNIT_ASSERT_VALUES_EQUAL(result.GetPartition(), 1);
+            UNIT_ASSERT_VALUES_EQUAL(NPersQueue::NErrorCode::EErrorCode_Name(result.GetReadResult().GetErrorCode()),
+                NPersQueue::NErrorCode::EErrorCode_Name(NPersQueue::NErrorCode::EErrorCode::OK));
+            UNIT_ASSERT_VALUES_EQUAL(result.GetReadResult().GetMaxOffset(), 3);
+        }
+
+        {
+            auto& result = ev->Response.GetPartResult(1);
+            UNIT_ASSERT_VALUES_EQUAL(result.GetTopic(), "/Root/topic2");
+            UNIT_ASSERT_VALUES_EQUAL(result.GetPartition(), 3);
+            UNIT_ASSERT_VALUES_EQUAL(NPersQueue::NErrorCode::EErrorCode_Name(result.GetReadResult().GetErrorCode()),
+                NPersQueue::NErrorCode::EErrorCode_Name(NPersQueue::NErrorCode::EErrorCode::OK));
+            UNIT_ASSERT_VALUES_EQUAL(result.GetReadResult().GetMaxOffset(), 0);
+        }
+
+        {
+            auto& result = ev->Response.GetPartResult(2);
+            UNIT_ASSERT_VALUES_EQUAL(result.GetTopic(), "/Root/topic2");
+            UNIT_ASSERT_VALUES_EQUAL(result.GetPartition(), 2);
+            UNIT_ASSERT_VALUES_EQUAL(NPersQueue::NErrorCode::EErrorCode_Name(result.GetReadResult().GetErrorCode()),
+                NPersQueue::NErrorCode::EErrorCode_Name(NPersQueue::NErrorCode::EErrorCode::READ_ERROR_TOO_BIG_OFFSET));
+            UNIT_ASSERT_VALUES_EQUAL(result.GetReadResult().GetMaxOffset(), 0);
+        }
+    }
+
+    Y_UNIT_TEST(SmallBytesRead) {
+        auto setup = std::make_shared<TTopicSdkTestSetup>(TEST_CASE_NAME);
+        setup->GetServer().EnableLogs(
+                { NKikimrServices::TX_PROXY_SCHEME_CACHE, NKikimrServices::PQ_FETCH_REQUEST },
+                NActors::NLog::PRI_DEBUG
+        );
+        auto& runtime = setup->GetRuntime();
+        StartSchemeCache(runtime);
+
+        setup->CreateTopic("topic1", "dc1", 2);
+        setup->Write("/Root/topic1", TString(2_KB, 'a'), 0);
+
+        TPartitionFetchRequest p1 {"/Root/topic1", 0, 0, 1_KB};
+        TPartitionFetchRequest p2 {"/Root/topic1", 1, 0, 1_KB};
+
+        TFetchRequestSettings settings{
+            .Database = {},
+            .Consumer = NKikimr::NPQ::CLIENTID_WITHOUT_CONSUMER,
+            .Partitions = {p1, p2},
+            .MaxWaitTimeMs = 1000,
+            .TotalMaxBytes = 100
+        };
+
+        auto edgeId = runtime.AllocateEdgeActor();
+        auto fetchActorId = runtime.Register(CreatePQFetchRequestActor(settings, MakeSchemeCacheID(), edgeId));
+        runtime.EnableScheduleForActor(fetchActorId);
+        runtime.DispatchEvents();
+        
+        auto ev = runtime.GrabEdgeEvent<TEvPQ::TEvFetchResponse>();
+        Cerr << ev->Response.DebugString() << Endl;
+        UNIT_ASSERT_C(ev->Status == Ydb::StatusIds::SUCCESS, ev->Message);
+
+        UNIT_ASSERT_VALUES_EQUAL(ev->Response.PartResultSize(), 2);
+
+        {
+            auto& result = ev->Response.GetPartResult(0);
+            UNIT_ASSERT_VALUES_EQUAL(result.GetPartition(), 0);
+            UNIT_ASSERT_VALUES_EQUAL(NPersQueue::NErrorCode::EErrorCode_Name(result.GetReadResult().GetErrorCode()),
+                NPersQueue::NErrorCode::EErrorCode_Name(NPersQueue::NErrorCode::EErrorCode::OK));
+            UNIT_ASSERT_VALUES_EQUAL(result.GetReadResult().GetMaxOffset(), 1);
+            UNIT_ASSERT_VALUES_EQUAL(result.GetReadResult().ResultSize(), 1);
+            UNIT_ASSERT_VALUES_EQUAL(result.GetReadResult().GetResult(0).GetUncompressedSize(), 2_KB);
+        }
+
+        {
+            auto& result = ev->Response.GetPartResult(1);
+            UNIT_ASSERT_VALUES_EQUAL(result.GetPartition(), 1);
+            UNIT_ASSERT_VALUES_EQUAL(NPersQueue::NErrorCode::EErrorCode_Name(result.GetReadResult().GetErrorCode()),
+                NPersQueue::NErrorCode::EErrorCode_Name(NPersQueue::NErrorCode::EErrorCode::READ_NOT_DONE));
+            UNIT_ASSERT_VALUES_EQUAL(result.GetReadResult().GetMaxOffset(), 0);
+            UNIT_ASSERT_VALUES_EQUAL(result.GetReadResult().ResultSize(), 0);
+        }
+    }
+
+    Y_UNIT_TEST(EmptyTopic) {
+        auto setup = std::make_shared<TTopicSdkTestSetup>(TEST_CASE_NAME);
+        setup->GetServer().EnableLogs(
+                { NKikimrServices::TX_PROXY_SCHEME_CACHE, NKikimrServices::PQ_FETCH_REQUEST },
+                NActors::NLog::PRI_DEBUG
+        );
+        auto& runtime = setup->GetRuntime();
+        StartSchemeCache(runtime);
+
+        setup->CreateTopic("topic1", "dc1", 2);
+
+        TPartitionFetchRequest p1 {"/Root/topic1", 0, 0, 1_KB};
+        TPartitionFetchRequest p2 {"/Root/topic1", 1, 0, 1_KB};
+
+        TFetchRequestSettings settings{
+            .Database = {},
+            .Consumer = NKikimr::NPQ::CLIENTID_WITHOUT_CONSUMER,
+            .Partitions = {p1, p2},
+            .MaxWaitTimeMs = 100,
+            .TotalMaxBytes = 100
+        };
+
+        auto edgeId = runtime.AllocateEdgeActor();
+        auto fetchActorId = runtime.Register(CreatePQFetchRequestActor(settings, MakeSchemeCacheID(), edgeId));
+        runtime.EnableScheduleForActor(fetchActorId);
+        runtime.DispatchEvents();
+
+        auto ev = runtime.GrabEdgeEvent<TEvPQ::TEvFetchResponse>();
+        Cerr << ev->Response.DebugString() << Endl;
+        UNIT_ASSERT_C(ev->Status == Ydb::StatusIds::SUCCESS, ev->Message);
+
+        UNIT_ASSERT_VALUES_EQUAL(ev->Response.PartResultSize(), 2);
+
+        {
+            auto& result = ev->Response.GetPartResult(0);
+            UNIT_ASSERT_VALUES_EQUAL(result.GetPartition(), 0);
+            UNIT_ASSERT_VALUES_EQUAL(NPersQueue::NErrorCode::EErrorCode_Name(result.GetReadResult().GetErrorCode()),
+                NPersQueue::NErrorCode::EErrorCode_Name(NPersQueue::NErrorCode::EErrorCode::READ_NOT_DONE));
+            UNIT_ASSERT_VALUES_EQUAL(result.GetReadResult().GetMaxOffset(), 0);
+            UNIT_ASSERT_VALUES_EQUAL(result.GetReadResult().ResultSize(), 0);
+        }
+
+        {
+            auto& result = ev->Response.GetPartResult(1);
+            UNIT_ASSERT_VALUES_EQUAL(result.GetPartition(), 1);
+            UNIT_ASSERT_VALUES_EQUAL(NPersQueue::NErrorCode::EErrorCode_Name(result.GetReadResult().GetErrorCode()),
+                NPersQueue::NErrorCode::EErrorCode_Name(NPersQueue::NErrorCode::EErrorCode::READ_NOT_DONE));
+            UNIT_ASSERT_VALUES_EQUAL(result.GetReadResult().GetMaxOffset(), 0);
+            UNIT_ASSERT_VALUES_EQUAL(result.GetReadResult().ResultSize(), 0);
         }
     }
 
     Y_UNIT_TEST(BadTopicName) {
-        auto setup = std::make_shared<TPersQueueYdbSdkTestSetup>(TEST_CASE_NAME);
+        auto setup = std::make_shared<TTopicSdkTestSetup>(TEST_CASE_NAME);
         auto& runtime = setup->GetRuntime();
         StartSchemeCache(runtime);
 
@@ -96,8 +204,8 @@ Y_UNIT_TEST_SUITE(TFetchRequestTests) {
         setup->CreateTopic("topic1", "dc1", totalPartitions);
 
         auto edgeId = runtime.AllocateEdgeActor();
-        TPartitionFetchRequest p1{"Root/PQ/rt3.dc1--topic1", 1, 1, 10000};
-        TPartitionFetchRequest p2{"Root/PQ/rt3.dc1--topic2", 3, 0, 10000};
+        TPartitionFetchRequest p1{"/Root/topic1", 1, 1, 10000};
+        TPartitionFetchRequest p2{"/Root/topic2", 3, 0, 10000};
 
         TFetchRequestSettings settings{{}, NKikimr::NPQ::CLIENTID_WITHOUT_CONSUMER, {p1, p2}, 1000, 1000};
         auto fetchId = runtime.Register(CreatePQFetchRequestActor(settings, MakeSchemeCacheID(), edgeId));
@@ -108,7 +216,7 @@ Y_UNIT_TEST_SUITE(TFetchRequestTests) {
     }
 
     Y_UNIT_TEST(CheckAccess) {
-        auto setup = std::make_shared<TPersQueueYdbSdkTestSetup>(TEST_CASE_NAME);
+        auto setup = std::make_shared<TTopicSdkTestSetup>(TEST_CASE_NAME);
         auto& runtime = setup->GetRuntime();
         runtime.SetLogPriority(NKikimrServices::PQ_FETCH_REQUEST, NActors::NLog::EPriority::PRI_DEBUG);
         StartSchemeCache(runtime);
@@ -117,12 +225,12 @@ Y_UNIT_TEST_SUITE(TFetchRequestTests) {
         setup->CreateTopic("topic1", "dc1", totalPartitions);
 
         auto edgeId = runtime.AllocateEdgeActor();
-        TPartitionFetchRequest p1{"Root/PQ/rt3.dc1--topic1", 1, 1, 10000};
+        TPartitionFetchRequest p1{"/Root/topic1", 1, 1, 10000};
 
         {
             NACLib::TDiffACL acl;
             acl.AddAccess(NACLib::EAccessType::Allow, NACLib::SelectRow, "user1@staff");
-            setup->GetServer().AnnoyingClient->ModifyACL("/Root/PQ", "rt3.dc1--topic1", acl.SerializeAsString());
+            setup->GetServer().AnnoyingClient->ModifyACL("/Root", "topic1", acl.SerializeAsString());
 
             auto goodToken = MakeIntrusiveConst<NACLib::TUserToken>("user1@staff", TVector<TString>{});
             TFetchRequestSettings settings{
diff --git a/ydb/core/persqueue/public/fetcher/ut/ya.make b/ydb/core/persqueue/public/fetcher/ut/ya.make
@@ -9,6 +9,7 @@ SRCS(
 PEERDIR(
     library/cpp/testing/unittest
     ydb/public/sdk/cpp/src/client/persqueue_public/ut/ut_utils
+    ydb/public/sdk/cpp/src/client/topic/ut/ut_utils
 
 )
 
diff --git a/ydb/public/sdk/cpp/src/client/persqueue_public/ut/ut_utils/ut_utils.h b/ydb/public/sdk/cpp/src/client/persqueue_public/ut/ut_utils/ut_utils.h
@@ -97,6 +97,19 @@ class TPersQueueYdbSdkTestSetup : public ::NPersQueue::SDKTestSetup {
                 .ClusterDiscoveryMode(EClusterDiscoveryMode::On);
         return settings;
     }
+
+    void Write(const TString& topic, ui32 partitionId, const TString& data) {
+        auto settings = TWriteSessionSettings()
+            .Path(topic)
+            .MessageGroupId("src-id")
+            .PartitionGroupId(partitionId)
+            .Codec(ECodec::RAW);
+        auto writeSession = GetPersQueueClient().CreateSimpleBlockingWriteSession(settings);
+        
+        writeSession->Write(data);
+
+        writeSession->Close();
+    }
 };
 
 struct TYDBClientEventLoop : public ::NPersQueue::IClientEventLoop {
diff --git a/ydb/public/sdk/cpp/src/client/topic/ut/ut_utils/topic_sdk_test_setup.cpp b/ydb/public/sdk/cpp/src/client/topic/ut/ut_utils/topic_sdk_test_setup.cpp
@@ -56,10 +56,16 @@ TConsumerDescription TTopicSdkTestSetup::DescribeConsumer(const std::string& nam
 void TTopicSdkTestSetup::Write(const std::string& message, std::uint32_t partitionId,
                                const std::optional<std::string> producer,
                                std::optional<std::uint64_t> seqNo) {
+    Write(GetTopicPath(), message, partitionId, producer, seqNo);
+}
+
+void TTopicSdkTestSetup::Write(const std::string& topic, const std::string& message, std::uint32_t partitionId,
+                               const std::optional<std::string> producer,
+                               std::optional<std::uint64_t> seqNo) {                            
     TTopicClient client(MakeDriver());
 
     TWriteSessionSettings settings;
-    settings.Path(GetTopicPath());
+    settings.Path(topic);
     settings.PartitionId(partitionId);
     settings.DeduplicationEnabled(producer.has_value());
     if (producer) {
diff --git a/ydb/public/sdk/cpp/src/client/topic/ut/ut_utils/topic_sdk_test_setup.h b/ydb/public/sdk/cpp/src/client/topic/ut/ut_utils/topic_sdk_test_setup.h
@@ -32,6 +32,10 @@ class TTopicSdkTestSetup : public ITopicTestSetup {
                const std::optional<std::string> producer = std::nullopt,
                std::optional<std::uint64_t> seqNo = std::nullopt);
 
+    void Write(const std::string& topic, const std::string& message, std::uint32_t partitionId = 0,
+               const std::optional<std::string> producer = std::nullopt,
+               std::optional<std::uint64_t> seqNo = std::nullopt);
+
     struct TReadResult {
         std::shared_ptr<IReadSession> Reader;
         bool Timeout;

Original file line number	Diff line number	Diff line change
`@@ -9,6 +9,7 @@ SRCS(`
`9`	`9`	`PEERDIR(`
`10`	`10`	`library/cpp/testing/unittest`
`11`	`11`	`ydb/public/sdk/cpp/src/client/persqueue_public/ut/ut_utils`
	`12`	`+ ydb/public/sdk/cpp/src/client/topic/ut/ut_utils`
`12`	`13`
`13`	`14`	`)`
`14`	`15`