YQ-3909 Shared reading: fix Ingress statistics (#11943)

Author: kardymonds

ydb/core/fq/libs/row_dispatcher/topic_session.cpp

@@ -487,9 +487,6 @@ void TTopicSession::Handle(NFq::TEvPrivate::TEvSendStatisticToReadActor::TPtr&)
         if (!info.NextMessageOffset) {
             continue;
         }
-        if (*info.NextMessageOffset <= info.LastSendedNextMessageOffset) {
-            continue;
-        }
         auto event = std::make_unique<TEvRowDispatcher::TEvStatistics>();
         event->Record.SetPartitionId(PartitionId);
         event->Record.SetNextMessageOffset(*info.NextMessageOffset);

ydb/core/fq/libs/row_dispatcher/ut/topic_session_ut.cpp

@@ -161,6 +161,16 @@ class TFixture : public NUnitTest::TBaseFixture {
         return eventHolder->Get()->Record.MessagesSize();
     }
 
+    void ExpectStatisticToReadActor(TSet<NActors::TActorId> readActorIds) {
+        size_t count = readActorIds.size();
+        for (size_t i = 0; i < count; ++i) {
+            auto eventHolder = Runtime.GrabEdgeEvent<TEvRowDispatcher::TEvStatistics>(RowDispatcherActorId, TDuration::Seconds(GrabTimeoutSec));
+            UNIT_ASSERT(eventHolder.Get() != nullptr);
+            UNIT_ASSERT(readActorIds.contains(eventHolder->Get()->ReadActorId));
+            readActorIds.erase(eventHolder->Get()->ReadActorId);
+        }
+    }
+
     IPureCalcProgramFactory::TPtr PureCalcProgramFactory;
     NActors::TTestActorRuntime Runtime;
     TActorSystemStub ActorSystemStub;
@@ -194,6 +204,7 @@ Y_UNIT_TEST_SUITE(TopicSessionTests) {
         ExpectNewDataArrived({ReadActorId1, ReadActorId2});
         ExpectMessageBatch(ReadActorId1, { Json1 });
         ExpectMessageBatch(ReadActorId2, { Json1 });
+        ExpectStatisticToReadActor({ReadActorId1, ReadActorId2});
 
         auto source2 = BuildSource(topicName, false, "OtherConsumer");
         StartSession(ReadActorId3, source2);