YQ-3924 / YQ-4095 Fix actor name / memory leak / to stable (#14366)

Author: kardymonds

ydb/core/fq/libs/row_dispatcher/events/data_plane.h

@@ -163,12 +163,13 @@ struct TEvRowDispatcher {
         TTopicSessionStatistic Stat;
     };
 
-// Network events (without seqNo checks)
-
+    // two purposes: confirm seqNo and check the availability of the recipient actor (wait TEvUndelivered)
     struct TEvHeartbeat : public NActors::TEventPB<TEvHeartbeat, NFq::NRowDispatcherProto::TEvHeartbeat, EEv::EvHeartbeat> {
         TEvHeartbeat() = default;
     };
 
+// Network events (without seqNo checks)
+
     struct TEvNoSession : public NActors::TEventPB<TEvNoSession, NFq::NRowDispatcherProto::TEvNoSession, EEv::EvNoSession> {
         TEvNoSession() = default;
     };

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

@@ -17,8 +17,10 @@ namespace {
 class TTopicFormatHandler : public NActors::TActor<TTopicFormatHandler>, public ITopicFormatHandler, public TTypeParser {
     using TBase = NActors::TActor<TTopicFormatHandler>;
 
+public:
     static constexpr char ActorName[] = "FQ_ROW_DISPATCHER_FORMAT_HANDLER";
 
+private:
     struct TCounters {
         TCountersDesc Desc;
 

ydb/core/fq/libs/row_dispatcher/protos/events.proto

@@ -105,6 +105,7 @@ message TEvSessionError {
 
 message TEvHeartbeat {
     uint32 PartitionId = 1;
+    optional NYql.NDqProto.TMessageTransportMeta TransportMeta = 100;
 }
 
 message TEvNoSession {

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

@@ -884,9 +884,7 @@ void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvHeartbeat::TPtr& ev) {
     }
     LWPROBE(Heartbeat, ev->Sender.ToString(), ev->Get()->Record.GetPartitionId(), it->second->QueryId, ev->Get()->Record.ByteSizeLong());
     LOG_ROW_DISPATCHER_TRACE("Received TEvHeartbeat from " << ev->Sender << ", part id " << ev->Get()->Record.GetPartitionId() << " query id " << it->second->QueryId);
-    if (ev->Cookie != it->second->Generation) {
-        LOG_ROW_DISPATCHER_WARN("Wrong message generation (TEvHeartbeat), sender " << ev->Sender << " cookie " << ev->Cookie << ", session generation " << it->second->Generation << ", query id " << it->second->QueryId); 
-    }
+    CheckSession(it->second, ev);
 }
 
 void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvNoSession::TPtr& ev) {
@@ -983,7 +981,7 @@ void TRowDispatcher::Handle(const NYql::NDq::TEvRetryQueuePrivate::TEvEvHeartbea
     if (needSend) {
         LOG_ROW_DISPATCHER_TRACE("Send TEvHeartbeat to " << sessionInfo->ReadActorId << " query id " << sessionInfo->QueryId);
         auto event = std::make_unique<NFq::TEvRowDispatcher::TEvHeartbeat>();
-        Send(new IEventHandle(sessionInfo->ReadActorId, SelfId(), event.release(), 0, sessionInfo->Generation));
+        sessionInfo->EventsQueue.Send(new NFq::TEvRowDispatcher::TEvHeartbeat(), sessionInfo->Generation);
     }
 }
 

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

@@ -680,23 +680,14 @@ void TDqPqRdReadActor::Handle(const NYql::NDq::TEvRetryQueuePrivate::TEvEvHeartb
     bool needSend = sessionInfo.EventsQueue.Heartbeat();
     if (needSend) {
         SRC_LOG_T("Send TEvEvHeartbeat");
-        Send(sessionInfo.RowDispatcherActorId, new NFq::TEvRowDispatcher::TEvHeartbeat(), IEventHandle::FlagTrackDelivery, sessionInfo.Generation);
+        sessionInfo.EventsQueue.Send(new NFq::TEvRowDispatcher::TEvHeartbeat(), sessionInfo.Generation);
     }
 }
 
 void TDqPqRdReadActor::Handle(const NFq::TEvRowDispatcher::TEvHeartbeat::TPtr& ev) {
     SRC_LOG_T("Received TEvHeartbeat from " << ev->Sender << ", generation " << ev->Cookie);
     Counters.Heartbeat++;
-    auto sessionIt = Sessions.find(ev->Sender);
-    if (sessionIt == Sessions.end()) {
-        SRC_LOG_W("Ignore TEvHeartbeat from " << ev->Sender << ", generation " << ev->Cookie);
-        SendNoSession(ev->Sender, ev->Cookie);
-        return;
-    }
-
-    if (ev->Cookie != sessionIt->second.Generation) {
-        SendNoSession(ev->Sender, ev->Cookie);
-    }
+    FindSession(ev);
 }
 
 void TDqPqRdReadActor::Handle(NFq::TEvRowDispatcher::TEvCoordinatorChanged::TPtr& ev) {

ydb/tests/fq/yds/test_row_dispatcher.py

@@ -977,6 +977,7 @@ def test_sensors(self, kikimr, client):
         wait_actor_count(kikimr, "DQ_PQ_READ_ACTOR", 1)
         wait_actor_count(kikimr, "FQ_ROW_DISPATCHER_SESSION", 1)
         wait_actor_count(kikimr, "FQ_ROW_DISPATCHER_COMPILE_SERVICE", COMPUTE_NODE_COUNT)
+        wait_actor_count(kikimr, "FQ_ROW_DISPATCHER_FORMAT_HANDLER", 1)
         wait_row_dispatcher_sensor_value(kikimr, "ClientsCount", 1)
         wait_row_dispatcher_sensor_value(kikimr, "RowsSent", 1, exact_match=False)
         wait_row_dispatcher_sensor_value(kikimr, "IncomingRequests", 1, exact_match=False)