YQ-3779 Remove FlagSubscribeOnSession (use TEvSubscribe) / to stable (#11088)

Author: kardymonds

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

@@ -81,6 +81,7 @@ class TActorCoordinator : public TActorBootstrapped<TActorCoordinator> {
     THashMap<TPartitionKey, TActorId, TPartitionKeyHash> PartitionLocations;
     TCoordinatorMetrics Metrics;
     ui64 LocationRandomCounter = 0;
+    THashSet<TActorId> InterconnectSessions;
 
 public:
     TActorCoordinator(
@@ -116,6 +117,7 @@ class TActorCoordinator : public TActorBootstrapped<TActorCoordinator> {
     void AddRowDispatcher(NActors::TActorId actorId, bool isLocal);
     void PrintInternalState();
     NActors::TActorId GetAndUpdateLocation(const TPartitionKey& key);
+    void UpdateInterconnectSessions(const NActors::TActorId& interconnectSession);
 };
 
 TActorCoordinator::TActorCoordinator(
@@ -165,11 +167,24 @@ void TActorCoordinator::AddRowDispatcher(NActors::TActorId actorId, bool isLocal
     RowDispatchers.emplace(actorId, RowDispatcherInfo{true, isLocal});
 }
 
+void TActorCoordinator::UpdateInterconnectSessions(const NActors::TActorId& interconnectSession) {
+    if (!interconnectSession) {
+        return;
+    }
+    auto sessionsIt = InterconnectSessions.find(interconnectSession);
+    if (sessionsIt != InterconnectSessions.end()) {
+        return;
+    }
+    Send(interconnectSession, new NActors::TEvents::TEvSubscribe, IEventHandle::FlagTrackDelivery);
+    InterconnectSessions.insert(interconnectSession);
+}
+
 void TActorCoordinator::Handle(NActors::TEvents::TEvPing::TPtr& ev) {
     LOG_ROW_DISPATCHER_TRACE("TEvPing received, " << ev->Sender);
+    UpdateInterconnectSessions(ev->InterconnectSession);
     AddRowDispatcher(ev->Sender, false);
     LOG_ROW_DISPATCHER_TRACE("Send TEvPong to " << ev->Sender);
-    Send(ev->Sender, new NActors::TEvents::TEvPong(), IEventHandle::FlagTrackDelivery | IEventHandle::FlagSubscribeOnSession);
+    Send(ev->Sender, new NActors::TEvents::TEvPong(), IEventHandle::FlagTrackDelivery);
 }
 
 void TActorCoordinator::PrintInternalState() {
@@ -247,6 +262,7 @@ NActors::TActorId TActorCoordinator::GetAndUpdateLocation(const TPartitionKey& k
 
 void TActorCoordinator::Handle(NFq::TEvRowDispatcher::TEvCoordinatorRequest::TPtr& ev) {
     const auto source =  ev->Get()->Record.GetSource();
+    UpdateInterconnectSessions(ev->InterconnectSession);
 
     TStringStream str;
     str << "TEvCoordinatorRequest from " << ev->Sender.ToString() << ", " << source.GetTopicPath() << ", partIds: ";
@@ -281,7 +297,7 @@ void TActorCoordinator::Handle(NFq::TEvRowDispatcher::TEvCoordinatorRequest::TPt
     }
 
     LOG_ROW_DISPATCHER_DEBUG("Send TEvCoordinatorResult to " << ev->Sender);
-    Send(ev->Sender, response.release(), IEventHandle::FlagTrackDelivery | IEventHandle::FlagSubscribeOnSession, ev->Cookie);
+    Send(ev->Sender, response.release(), IEventHandle::FlagTrackDelivery, ev->Cookie);
     PrintInternalState();
 }
 

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

@@ -127,6 +127,7 @@ class TRowDispatcher : public TActorBootstrapped<TRowDispatcher> {
     const ::NMonitoring::TDynamicCounterPtr Counters;
     TRowDispatcherMetrics Metrics;
     NYql::IPqGateway::TPtr PqGateway;
+    THashSet<TActorId> InterconnectSessions;
 
     struct ConsumerCounters {
         ui64 NewDataArrived = 0;
@@ -216,6 +217,7 @@ class TRowDispatcher : public TActorBootstrapped<TRowDispatcher> {
     void Handle(NFq::TEvPrivate::TEvUpdateMetrics::TPtr&);
 
     void DeleteConsumer(const ConsumerSessionKey& key);
+    void UpdateInterconnectSessions(const NActors::TActorId& interconnectSession);
     void UpdateMetrics();
 
     STRICT_STFUNC(
@@ -279,12 +281,12 @@ void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvCoordinatorChanged::TPtr&
     LOG_ROW_DISPATCHER_DEBUG("Coordinator changed, old leader " << CoordinatorActorId << ", new " << ev->Get()->CoordinatorActorId);
 
     CoordinatorActorId = ev->Get()->CoordinatorActorId;
-    Send(*CoordinatorActorId, new NActors::TEvents::TEvPing(), IEventHandle::FlagTrackDelivery | IEventHandle::FlagSubscribeOnSession);
+    Send(*CoordinatorActorId, new NActors::TEvents::TEvPing(), IEventHandle::FlagTrackDelivery);
     for (auto actorId : CoordinatorChangedSubscribers) {
         Send(
             actorId,
             new NFq::TEvRowDispatcher::TEvCoordinatorChanged(ev->Get()->CoordinatorActorId),
-            IEventHandle::FlagTrackDelivery | IEventHandle::FlagSubscribeOnSession);
+            IEventHandle::FlagTrackDelivery);
     }
 }
 
@@ -324,11 +326,12 @@ void TRowDispatcher::Handle(NActors::TEvents::TEvPong::TPtr&) {
 
 void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvCoordinatorChangesSubscribe::TPtr& ev) {
     LOG_ROW_DISPATCHER_DEBUG("TEvCoordinatorChangesSubscribe from " << ev->Sender);
+    UpdateInterconnectSessions(ev->InterconnectSession);
     CoordinatorChangedSubscribers.insert(ev->Sender);
     if (!CoordinatorActorId) {
         return;
     }
-    Send(ev->Sender, new NFq::TEvRowDispatcher::TEvCoordinatorChanged(*CoordinatorActorId), IEventHandle::FlagTrackDelivery | IEventHandle::FlagSubscribeOnSession);
+    Send(ev->Sender, new NFq::TEvRowDispatcher::TEvCoordinatorChanged(*CoordinatorActorId), IEventHandle::FlagTrackDelivery);
 }
 
 void TRowDispatcher::UpdateMetrics() {
@@ -372,7 +375,7 @@ void TRowDispatcher::UpdateMetrics() {
 void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvStartSession::TPtr& ev) {
     LOG_ROW_DISPATCHER_DEBUG("TEvStartSession from " << ev->Sender << ", topicPath " << ev->Get()->Record.GetSource().GetTopicPath() <<
         " partitionId " << ev->Get()->Record.GetPartitionId());
-
+    UpdateInterconnectSessions(ev->InterconnectSession);
     TMaybe<ui64> readOffset;
     if (ev->Get()->Record.HasOffset()) {
         readOffset = ev->Get()->Record.GetOffset();
@@ -634,6 +637,18 @@ void TRowDispatcher::Handle(NFq::TEvRowDispatcher::TEvSessionStatistic::TPtr& ev
     }
 }
 
+void TRowDispatcher::UpdateInterconnectSessions(const NActors::TActorId& interconnectSession) {
+    if (!interconnectSession) {
+        return;
+    }
+    auto sessionsIt = InterconnectSessions.find(interconnectSession);
+    if (sessionsIt != InterconnectSessions.end()) {
+        return;
+    }
+    Send(interconnectSession, new NActors::TEvents::TEvSubscribe, IEventHandle::FlagTrackDelivery);
+    InterconnectSessions.insert(interconnectSession);
+}
+
 } // namespace
 
 ////////////////////////////////////////////////////////////////////////////////