YQ-3583 Improvements after load tests / to stable (#10019)

ydb-platform · Oct 3, 2024 · 0da0d6a · 0da0d6a
1 parent afee6e2
commit 0da0d6a
Show file tree

Hide file tree

Showing 2 changed files with 28 additions and 2 deletions.
diff --git a/ydb/core/fq/libs/row_dispatcher/topic_session.cpp b/ydb/core/fq/libs/row_dispatcher/topic_session.cpp
@@ -92,6 +92,7 @@ struct TEvPrivate {
 
 ui64 PrintStatePeriodSec = 60;
 ui64 MaxBatchSizeBytes = 10000000;
+ui64 MaxHandledEvents = 1000;
 
 TVector<TString> GetVector(const google::protobuf::RepeatedPtrField<TString>& value) {
     return {value.begin(), value.end()};
@@ -227,6 +228,7 @@ class TTopicSession : public TActorBootstrapped<TTopicSession> {
         cFunc(NActors::TEvents::TEvPoisonPill::EventType, PassAway);
         IgnoreFunc(NFq::TEvPrivate::TEvPqEventsReady);
         IgnoreFunc(NFq::TEvPrivate::TEvCreateSession);
+        IgnoreFunc(NFq::TEvPrivate::TEvDataParsed);
         IgnoreFunc(NFq::TEvPrivate::TEvDataAfterFilteration);
         IgnoreFunc(NFq::TEvPrivate::TEvStatus);
         IgnoreFunc(NFq::TEvPrivate::TEvDataFiltered);
@@ -442,7 +444,7 @@ void TTopicSession::Handle(TEvRowDispatcher::TEvGetNextBatch::TPtr& ev) {
 }
 
 void TTopicSession::HandleNewEvents() {
-    while (true) {
+    for (ui64 i = 0; i < MaxHandledEvents; ++i) {
         if (!ReadSession) {
             return;
         }
@@ -475,7 +477,6 @@ void TTopicSession::TTopicEventProcessor::operator()(NYdb::NTopic::TReadSessionE
         LOG_ROW_DISPATCHER_TRACE("Data received: " << message.DebugString(true));
 
         TString item = message.GetData();
-        item.Detach();
         Self.SendToParsing(message.GetOffset(), item);
         Self.LastMessageOffset = message.GetOffset();
     }
@@ -540,6 +541,10 @@ void TTopicSession::SendToParsing(ui64 offset, const TString& message) {
         }
     }
 
+    if (ClientsWithoutPredicate.size() == Clients.size()) {
+        return;
+    }
+
     try {
         Parser->Push(offset, message);
     } catch (const std::exception& e) {

diff --git a/ydb/core/fq/libs/row_dispatcher/ut/topic_session_ut.cpp b/ydb/core/fq/libs/row_dispatcher/ut/topic_session_ut.cpp
@@ -171,6 +171,27 @@ Y_UNIT_TEST_SUITE(TopicSessionTests) {
         StopSession(ReadActorId2, source);
     }
 
+    Y_UNIT_TEST_F(TwoSessionWithoutPredicate, TFixture) {
+        const TString topicName = "twowithoutpredicate";
+        PQCreateStream(topicName);
+        Init(topicName);
+        auto source1 = BuildSource(topicName, true);
+        auto source2 = BuildSource(topicName, true);
+        StartSession(ReadActorId1, source1);
+        StartSession(ReadActorId2, source2);
+
+        const std::vector<TString> data = { Json1 };
+        PQWrite(data, topicName);
+        ExpectNewDataArrived({ReadActorId1, ReadActorId2});
+        Runtime.Send(new IEventHandle(TopicSession, ReadActorId1, new TEvRowDispatcher::TEvGetNextBatch()));
+        Runtime.Send(new IEventHandle(TopicSession, ReadActorId2, new TEvRowDispatcher::TEvGetNextBatch()));
+        ExpectMessageBatch(ReadActorId1, { Json1 });
+        ExpectMessageBatch(ReadActorId2, { Json1 });
+
+        StopSession(ReadActorId1, source1);
+        StopSession(ReadActorId2, source2);
+    }
+
     Y_UNIT_TEST_F(SessionWithPredicateAndSessionWithoutPredicate, TFixture) {
         const TString topicName = "topic2";
         PQCreateStream(topicName);