fix

Author: nshestakov

ydb/core/persqueue/partition_read.cpp

@@ -430,7 +430,7 @@ TReadAnswer TReadInfo::FormAnswer(
     Y_ABORT_UNLESS(blobs.size() == Blobs.size());
     response->Check();
     bool needStop = false;
-    for (ui32 pos = 0; pos < blobs.size() && !needStop && size < Size; ++pos) {
+    for (ui32 pos = 0; pos < blobs.size() && !needStop; ++pos) {
         Y_ABORT_UNLESS(Blobs[pos].Offset == blobs[pos].Offset, "Mismatch %" PRIu64 " vs %" PRIu64, Blobs[pos].Offset, blobs[pos].Offset);
         Y_ABORT_UNLESS(Blobs[pos].Count == blobs[pos].Count, "Mismatch %" PRIu32 " vs %" PRIu32, Blobs[pos].Count, blobs[pos].Count);
 

ydb/core/persqueue/ut/pq_ut.cpp

@@ -1606,14 +1606,14 @@ Y_UNIT_TEST(TestPQRead) {
         CmdRead(0, 9, 1, 100_MB, 1, false, tc);
         CmdRead(0, 23, 3, 100_MB, 3, false, tc);
 
-        CmdRead(0, 3, 1000, 511_KB, 4, false, tc);
-        CmdRead(0, 3, 1000, 511_KB, 4, false, tc);
-        CmdRead(0, 3, 1000, 1_KB, 4, false, tc); //at least one message will be readed always
+        CmdRead(0, 3, 1000, 511_KB, 12, false, tc);
+        CmdRead(0, 3, 1000, 511_KB, 12, false, tc);
+        CmdRead(0, 3, 1000, 1_KB, 12, false, tc); //at least one message will be readed always
         CmdRead(0, 25, 1000, 1_KB, 1, false, tc); //at least one message will be readed always, from head
 
         activeZone = true;
-        CmdRead(0, 9, 1000, 3_MB, 6, false, tc);
-        CmdRead(0, 9, 1000, 3_MB - 10_KB, 6, false, tc);
+        CmdRead(0, 9, 1000, 3_MB, 14, false, tc);
+        CmdRead(0, 9, 1000, 3_MB - 10_KB, 14, false, tc);
         CmdRead(0, 25, 1000, 512_KB, 1, false, tc); //from head
         CmdRead(0, 24, 1000, 512_KB, 1, false, tc); //from head
 
@@ -1709,9 +1709,9 @@ Y_UNIT_TEST(TestPQReadAhead) {
         CmdRead(0, 4, 10, 100_MB, 10, false, tc);
 
         CmdRead(0, 0, Max<i32>(), 100_KB, 12, false, tc);
-        CmdRead(0, 1, Max<i32>(), 100_KB, 11, false, tc);
-        CmdRead(0, 2, Max<i32>(), 100_KB, 10, false, tc);
-        CmdRead(0, 3, Max<i32>(), 100_KB, 9, false, tc);
+        CmdRead(0, 1, Max<i32>(), 100_KB, 19, false, tc);
+        CmdRead(0, 2, Max<i32>(), 100_KB, 18, false, tc);
+        CmdRead(0, 3, Max<i32>(), 100_KB, 17, false, tc);
     });
 }
 

ydb/services/persqueue_v1/persqueue_ut.cpp

@@ -880,12 +880,12 @@ Y_UNIT_TEST_SUITE(TPersQueueTest) {
             writer->Close();
         }
 
-        void DoRead(ui64 assignId, ui64& nextReadId, ui32& currTotalMessages, ui32 messageLimit) {
+        void DoRead(ui64 assignId, ui64& nextReadId, ui32& currTotalMessages, const ui32 messageLimit) {
             // Get DirectReadResponse messages, send DirectReadAck messages.
 
             auto endTime = TInstant::Now() + TDuration::Seconds(10);
             while (currTotalMessages < messageLimit && endTime > TInstant::Now()) {
-                Cerr << "Wait for direct read id: " << nextReadId << ", currently have " << currTotalMessages << " messages" << Endl;
+                Cerr << "Wait for direct read id: " << nextReadId << ", currently have " << currTotalMessages << " messages, limit is " << messageLimit << Endl;
 
                 Ydb::Topic::StreamDirectReadMessage::FromServer resp;
                 UNIT_ASSERT(DirectStream->Read(&resp));
@@ -1013,37 +1013,6 @@ TPersQueueV1TestServer server{{.CheckACL=true, .NodeCount=1}};
         UNIT_ASSERT_VALUES_EQUAL(cachedData->Data.begin()->second.Reads.size(), 0);
     }
 
-    Y_UNIT_TEST(DirectReadNotCached) {
-        TPersQueueV1TestServer server{{.CheckACL=true, .NodeCount=1}};
-        SET_LOCALS;
-        TDirectReadTestSetup setup{server};
-
-        setup.InitControlSession("acc/topic1");
-        auto [partitionId, assignId] = setup.GetNextAssign("acc/topic1");
-        UNIT_ASSERT_VALUES_EQUAL(partitionId, 0);
-        setup.InitDirectSession("acc/topic1");
-        setup.SendReadSessionAssign(assignId);
-
-        ui32 totalMsg = 0;
-        ui64 nextReadId = 1;
-        Sleep(TDuration::Seconds(3));
-        setup.DoWrite(pqClient->GetDriver(), "acc/topic1", 1_MB, 50);
-        setup.DoRead(assignId, nextReadId, totalMsg, 46);
-
-        Topic::StreamReadMessage::FromClient req;
-        req.mutable_read_request()->set_bytes_size(40_MB);
-        if (!setup.ControlStream->Write(req)) {
-            ythrow yexception() << "write fail";
-        }
-        setup.DoRead(assignId, nextReadId, totalMsg, 50);
-
-        Sleep(TDuration::Seconds(1));
-        auto cachedData = RequestCacheData(runtime, new TEvPQ::TEvGetFullDirectReadData());
-        UNIT_ASSERT_VALUES_EQUAL(cachedData->Data.size(), 1);
-        UNIT_ASSERT_VALUES_EQUAL(cachedData->Data.begin()->second.StagedReads.size(), 0);
-        UNIT_ASSERT_VALUES_EQUAL(cachedData->Data.begin()->second.Reads.size(), 0);
-    }
-
     Y_UNIT_TEST(DirectReadBadCases) {
         TPersQueueV1TestServer server{{.CheckACL=true, .NodeCount=1}};
         SET_LOCALS;