YQ-4121 Check generation by TEvNoSession / to stable (#14798)

Author: kardymonds

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

@@ -899,7 +899,15 @@ void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvHeartbeat::TPtr& ev) {
 }
 
 void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvNoSession::TPtr& ev) {
-    LOG_ROW_DISPATCHER_DEBUG("Received TEvNoSession from " << ev->Sender << ", cookie " << ev->Cookie);
+    LOG_ROW_DISPATCHER_DEBUG("Received TEvNoSession from " << ev->Sender << ", generation " << ev->Cookie);
+    auto consumerIt = Consumers.find(ev->Sender);
+    if (consumerIt == Consumers.end()) {
+        return;
+    }
+    const auto& consumer = consumerIt->second;
+    if (consumer->Generation != ev->Cookie) {
+        return;
+    }
     DeleteConsumer(ev->Sender);
 }
 

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

@@ -73,6 +73,7 @@ class TFixture : public NUnitTest::TBaseFixture {
         Runtime.SetLogPriority(NKikimrServices::FQ_ROW_DISPATCHER, NLog::PRI_TRACE);
         NConfig::TRowDispatcherConfig config;
         config.SetEnabled(true);
+        config.SetSendStatusPeriodSec(1);
         NConfig::TRowDispatcherCoordinatorConfig& coordinatorConfig = *config.MutableCoordinator();
         coordinatorConfig.SetCoordinationNodePath("RowDispatcher");
         auto& database = *coordinatorConfig.MutableDatabase();
@@ -156,9 +157,9 @@ class TFixture : public NUnitTest::TBaseFixture {
         Runtime.Send(new IEventHandle(RowDispatcher, readActorId, event.release(), 0, 1));
     }
 
-    void MockNoSession(TActorId readActorId) {
+    void MockNoSession(TActorId readActorId, ui64 generation) {
         auto event = std::make_unique<NFq::TEvRowDispatcher::TEvNoSession>();
-        Runtime.Send(new IEventHandle(RowDispatcher, readActorId, event.release(), 0, 1));
+        Runtime.Send(new IEventHandle(RowDispatcher, readActorId, event.release(), 0, generation));
     }
 
     void MockNewDataArrived(ui64 partitionId, TActorId topicSessionId, TActorId readActorId) {
@@ -181,10 +182,10 @@ class TFixture : public NUnitTest::TBaseFixture {
         Runtime.Send(new IEventHandle(RowDispatcher, topicSessionId, event.release()));
     }
 
-    void MockGetNextBatch(ui64 partitionId, TActorId readActorId, ui64 generation) {
+    void MockGetNextBatch(ui64 partitionId, TActorId readActorId, ui64 generation, ui64 seqNo = 2) {
         auto event = std::make_unique<NFq::TEvRowDispatcher::TEvGetNextBatch>();
         event->Record.SetPartitionId(partitionId);
-        event->Record.MutableTransportMeta()->SetSeqNo(2);
+        event->Record.MutableTransportMeta()->SetSeqNo(seqNo);
         Runtime.Send(new IEventHandle(RowDispatcher, readActorId, event.release(), 0, generation));
     }
 
@@ -236,11 +237,11 @@ class TFixture : public NUnitTest::TBaseFixture {
         return actorId;
     }
 
-    void ProcessData(NActors::TActorId readActorId, ui64 partId, NActors::TActorId topicSessionActorId, ui64 generation = 1) {
+    void ProcessData(NActors::TActorId readActorId, ui64 partId, NActors::TActorId topicSessionActorId, ui64 generation = 1, ui64 seqNo = 1) {
         MockNewDataArrived(partId, topicSessionActorId, readActorId);
         ExpectNewDataArrived(readActorId, partId);
 
-        MockGetNextBatch(partId, readActorId, generation);
+        MockGetNextBatch(partId, readActorId, generation, seqNo);
         ExpectGetNextBatch(topicSessionActorId, partId);
 
         MockMessageBatch(partId, topicSessionActorId, readActorId, generation);
@@ -440,13 +441,17 @@ Y_UNIT_TEST_SUITE(RowDispatcherTests) {
     }
 
     Y_UNIT_TEST_F(ProcessNoSession, TFixture) {
-        MockAddSession(Source1, {PartitionId0}, ReadActorId3);
+        ui64 generation = 42;
+        MockAddSession(Source1, {PartitionId0}, ReadActorId3, generation);
         auto topicSessionId = ExpectRegisterTopicSession();
-        ExpectStartSessionAck(ReadActorId3);
+        ExpectStartSessionAck(ReadActorId3, generation);
         ExpectStartSession(topicSessionId);
-        ProcessData(ReadActorId3, PartitionId0, topicSessionId);
+        ProcessData(ReadActorId3, PartitionId0, topicSessionId, generation, 2);
+
+        MockNoSession(ReadActorId3, generation - 1); // Ignore NoSession with wrong generation.
+        ProcessData(ReadActorId3, PartitionId0, topicSessionId, generation, 3);
 
-        MockNoSession(ReadActorId3);
+        MockNoSession(ReadActorId3, generation);
         ExpectStopSession(topicSessionId);
     }
 

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

@@ -944,7 +944,7 @@ TDqPqRdReadActor::TSession* TDqPqRdReadActor::FindAndUpdateSession(const TEventP
     auto& session = sessionIt->second;
 
     if (ev->Cookie != session.Generation) {
-        SRC_LOG_W("Wrong message generation (" << typeid(TEventPtr).name()  << "), sender " << ev->Sender << " cookie " << ev->Cookie << ", session generation " << session.Generation << ", send TEvStopSession");
+        SRC_LOG_W("Wrong message generation (" << typeid(TEventPtr).name()  << "), sender " << ev->Sender << " cookie " << ev->Cookie << ", session generation " << session.Generation << ", send TEvNoSession");
         SendNoSession(ev->Sender, ev->Cookie);
         return nullptr;
     }
@@ -995,13 +995,13 @@ void TDqPqRdReadActor::UpdateSessions() {
             continue;
         }
 
-        SRC_LOG_I("Create session to " << rowDispatcherActorId);
         auto queueId = ++NextEventQueueId;
         Sessions.emplace(
             std::piecewise_construct,
             std::forward_as_tuple(rowDispatcherActorId),
             std::forward_as_tuple(TxId, SelfId(), rowDispatcherActorId, queueId, ++NextGeneration));
         auto& session = Sessions.at(rowDispatcherActorId);
+        SRC_LOG_I("Create session to " << rowDispatcherActorId << ", generation " << session.Generation);
         for (auto partitionId : partitions) {
             session.Partitions[partitionId];
         }