YQ-2862 Add more logs to pq_dq_read_actor (#2335)

Author: kardymonds

ydb/library/yql/providers/pq/async_io/dq_pq_read_actor.cpp

@@ -382,7 +382,7 @@ class TDqPqReadActor : public NActors::TActor<TDqPqReadActor>, public IDqCompute
         NYdb::NPersQueue::TTopicReadSettings topicReadSettings;
         topicReadSettings.Path(SourceParams.GetTopicPath());
         auto partitionsToRead = GetPartitionsToRead();
-        SRC_LOG_D("PartitionsToRead: " << JoinSeq(", ", partitionsToRead));
+        SRC_LOG_D("RangesMode: " << RangesMode << ", PartitionsToRead: " << JoinSeq(", ", partitionsToRead));
         for (const auto partitionId : partitionsToRead) {
             topicReadSettings.AppendPartitionGroupIds(partitionId);
         }
@@ -495,6 +495,8 @@ class TDqPqReadActor : public NActors::TActor<TDqPqReadActor>, public IDqCompute
                 curBatch.Data.emplace_back(std::move(item));
                 curBatch.UsedSpace += size;
 
+                CheckAndUpdateOffset(message.GetPartitionStream(), message.GetOffset());
+
                 auto& offsets = curBatch.OffsetRanges[message.GetPartitionStream()];
                 if (!offsets.empty() && offsets.back().second == message.GetOffset()) {
                     offsets.back().second = message.GetOffset() + 1;
@@ -577,6 +579,18 @@ class TDqPqReadActor : public NActors::TActor<TDqPqReadActor>, public IDqCompute
             return std::make_pair(item, usedSpace);
         }
 
+        void CheckAndUpdateOffset(const NYdb::NPersQueue::TPartitionStream::TPtr& partitionStreamPtr, ui64 offset) {
+            auto offsetIt = Self.LastOffsetByPartitionStream.find(partitionStreamPtr);
+            if (offsetIt == Self.LastOffsetByPartitionStream.end()) {
+                Self.LastOffsetByPartitionStream[partitionStreamPtr] = offset;
+                return;
+            }
+            ui64 lastOffset = offsetIt->second;
+            if (offset <=  lastOffset) {
+                ythrow yexception() << "Invalid message offset " << offset << ", last offset " << lastOffset;
+            }
+        }
+
         TDqPqReadActor& Self;
         ui32 BatchCapacity;
         const TString& LogPrefix;
@@ -609,6 +623,7 @@ class TDqPqReadActor : public NActors::TActor<TDqPqReadActor>, public IDqCompute
     std::queue<TReadyBatch> ReadyBuffer;
     TMaybe<TDqSourceWatermarkTracker<TPartitionKey>> WatermarkTracker;
     TMaybe<TInstant> NextIdlenesCheckAt;
+    THashMap<NYdb::NPersQueue::TPartitionStream::TPtr, ui64> LastOffsetByPartitionStream;
 };
 
 std::pair<IDqComputeActorAsyncInput*, NActors::IActor*> CreateDqPqReadActor(

ydb/public/sdk/cpp/client/ydb_persqueue_core/impl/read_session.ipp

@@ -77,7 +77,7 @@ void TPartitionStreamImpl<UseMigrationProtocol>::Commit(ui64 startOffset, ui64 e
             Commits.EraseInterval(0, endOffset); // Drop only committed ranges;
         }
         for (auto range: toCommit) {
-            sessionShared->Commit(this, range.first, range.second);
+            sessionShared->Commit(this, range.first, Min(range.second, endOffset));
         }
     }
 }
@@ -1121,7 +1121,7 @@ inline void TSingleClusterReadSessionImpl<true>::OnReadDoneImpl(
     TDeferredActions<true>& deferred) {
     Y_ABORT_UNLESS(Lock.IsLocked());
 
-    LOG_LAZY(Log, TLOG_DEBUG, GetLogPrefix() << "Committed response: " << msg);
+    LOG_LAZY(Log, TLOG_INFO, GetLogPrefix() << "Committed response: " << msg);
 
     TMap<ui64, TIntrusivePtr<TPartitionStreamImpl<true>>> partitionStreams;
     for (const Ydb::PersQueue::V1::CommitCookie& cookieProto : msg.cookies()) {
@@ -1377,7 +1377,7 @@ inline void TSingleClusterReadSessionImpl<false>::OnReadDoneImpl(
     TDeferredActions<false>& deferred) {
     Y_ABORT_UNLESS(Lock.IsLocked());
 
-    LOG_LAZY(Log, TLOG_DEBUG, GetLogPrefix() << "Committed response: " << msg);
+    LOG_LAZY(Log, TLOG_INFO, GetLogPrefix() << "Committed response: " << msg);
 
     for (const auto& rangeProto : msg.partitions_committed_offsets()) {
         auto partitionStreamIt = PartitionStreams.find(rangeProto.partition_session_id());