Fix LastOffset bug in PQ (#28667)

Author: FloatingCrowbar

ydb/core/persqueue/pqtablet/partition/partition_read.cpp

@@ -504,6 +504,10 @@ TMaybe<TReadAnswer> TReadInfo::AddBlobsFromBody(const TVector<NPQ::TRequestedBlo
                 if (res.IsLastPart()) {
                     PartNo = 0;
                     ++Offset;
+		    if (LastOffset && Offset >= LastOffset) {
+                        needStop = true;
+                        break;
+                    }
                 } else {
                     ++PartNo;
                 }
@@ -640,6 +644,9 @@ TReadAnswer TReadInfo::FormAnswer(
             if (updateUsage(writeBlob)) {
                 break;
             }
+	    if (LastOffset && Offset >= LastOffset) {
+                break;
+            }
         }
     }
 

ydb/core/persqueue/ut/counters_ut.cpp

@@ -815,11 +815,13 @@ Y_UNIT_TEST(PartitionKeyCompaction) {
             group->GetNamedCounter("name", "topic.partition.write.lag_milliseconds_max", false)->Set(600);
             group->GetNamedCounter("name", "topic.partition.uptime_milliseconds_min", false)->Set(30000);
             group->GetNamedCounter("name", "topic.partition.write.lag_milliseconds_max", false)->Set(600);
+            group->GetNamedCounter("name", "topic.partition.read.throttled_microseconds_max", false)->Set(2000);
             group = group->GetSubgroup("consumer", "__ydb_compaction_consumer");
             group->GetNamedCounter("name", "topic.partition.write.lag_milliseconds_max", false)->Set(200);
             group->GetNamedCounter("name", "topic.partition.end_to_end_lag_milliseconds_max", false)->Set(30000);
             group->GetNamedCounter("name", "topic.partition.read.throttled_microseconds_max", false)->Set(2000);
             group->GetNamedCounter("name", "topic.partition.read.idle_milliseconds_max", false)->Set(300);
+            group->GetNamedCounter("name", "topic.partition.read.lag_milliseconds_max", false)->Set(300);
 
             TStringStream countersStr;
             dbGroup->OutputHtml(countersStr);

ydb/core/persqueue/ut/resources/counters_topics_extended.html

@@ -28,7 +28,7 @@
                         name=topic.partition.producers_count_max: 1
                         name=topic.partition.read.inflight_throttled_microseconds_max: 0
                         name=topic.partition.read.speed_limit_bytes_per_second: 20000000000
-                        name=topic.partition.read.throttled_microseconds_max: 148
+                        name=topic.partition.read.throttled_microseconds_max: 2000
                         name=topic.partition.read_without_consumer.speed_limit_bytes_per_second: 0
                         name=topic.partition.read_without_consumer.throttled_microseconds_max: 0
                         name=topic.partition.storage_bytes_max: 7549747200
@@ -53,12 +53,12 @@
                             name=topic.partition.committed_read_lag_milliseconds_max: 0
                             name=topic.partition.end_to_end_lag_milliseconds_max: 30000
                             name=topic.partition.read.idle_milliseconds_max: 300
-                            name=topic.partition.read.lag_messages_max: 0
-                            name=topic.partition.read.lag_milliseconds_max: 0
+                            name=topic.partition.read.lag_messages_max: 1
+                            name=topic.partition.read.lag_milliseconds_max: 300
                             name=topic.partition.read.speed_limit_bytes_per_second: 4194304
                             name=topic.partition.read.throttled_microseconds_max: 2000
                             name=topic.partition.write.lag_milliseconds_max: 200
-                            name=topic.read.lag_messages: 0
+                            name=topic.read.lag_messages: 1
 
                         consumer=client:
                             name=topic.partition.alive_count: 0