fix crash from ui consumer info request: forward PQReadSessionsInfoWorkerFactory LOGBROKER-7376

fix build

forward PQReadSessionsInfoWorkerFactory

ref:0e9712c8ea8a3201acc17dca62bdf8adf7269f13

6 files changed, 67 insertions, 14 deletions

diff --git a/ydb/core/driver_lib/run/kikimr_services_initializers.cpp b/ydb/core/driver_lib/run/kikimr_services_initializers.cpp
index 2779210077c..15a004a5e2c 100644
--- a/ydb/core/driver_lib/run/kikimr_services_initializers.cpp
+++ b/ydb/core/driver_lib/run/kikimr_services_initializers.cpp
@@ -1791,15 +1791,18 @@ void TTabletMonitorInitializer::InitializeServices(NActors::TActorSystemSetup* s
 
 // TViewerInitializer
 
-TViewerInitializer::TViewerInitializer(const TKikimrRunConfig& runConfig)
+TViewerInitializer::TViewerInitializer(const TKikimrRunConfig& runConfig, std::shared_ptr<TModuleFactories> factories)
     : IKikimrServicesInitializer(runConfig)
     , KikimrRunConfig(runConfig)
+    , Factories(factories)
 {}
 
-void TViewerInitializer::InitializeServices(NActors::TActorSystemSetup* setup,
-                                            const NKikimr::TAppData* appData) {
+void TViewerInitializer::InitializeServices(
+    NActors::TActorSystemSetup* setup,
+    const NKikimr::TAppData* appData
+) {
     using namespace NViewer;
-    IActor* viewer = CreateViewer(KikimrRunConfig);
+    IActor* viewer = CreateViewer(KikimrRunConfig, Factories->PQReadSessionsInfoWorkerFactory);
     SetupPQVirtualHandlers(dynamic_cast<IViewer*>(viewer));
     SetupDBVirtualHandlers(dynamic_cast<IViewer*>(viewer));
     setup->LocalServices.push_back(std::pair<TActorId, TActorSetupCmd>(MakeViewerID(NodeId),
diff --git a/ydb/core/driver_lib/run/kikimr_services_initializers.h b/ydb/core/driver_lib/run/kikimr_services_initializers.h
index 407ce1bb7ba..6265bcc03b7 100644
--- a/ydb/core/driver_lib/run/kikimr_services_initializers.h
+++ b/ydb/core/driver_lib/run/kikimr_services_initializers.h
@@ -310,9 +310,12 @@ class TViewerInitializer : public IKikimrServicesInitializer {
     const TKikimrRunConfig& KikimrRunConfig;
 
 public:
-    TViewerInitializer(const TKikimrRunConfig& runConfig);
+    TViewerInitializer(const TKikimrRunConfig& runConfig, std::shared_ptr<TModuleFactories> factories);
 
     void InitializeServices(NActors::TActorSystemSetup *setup, const NKikimr::TAppData *appData) override;
+
+private:
+    std::shared_ptr<TModuleFactories> Factories;
 };
 
 class TLoadInitializer : public IKikimrServicesInitializer {
diff --git a/ydb/core/driver_lib/run/run.cpp b/ydb/core/driver_lib/run/run.cpp
index 00e2fd2a64e..5aa6bba9a26 100644
--- a/ydb/core/driver_lib/run/run.cpp
+++ b/ydb/core/driver_lib/run/run.cpp
@@ -1262,7 +1262,7 @@ TIntrusivePtr<TServiceInitializersList> TKikimrRunner::CreateServiceInitializers
     }
 
     if (serviceMask.EnableViewerService) {
-        sil->AddServiceInitializer(new TViewerInitializer(runConfig));
+        sil->AddServiceInitializer(new TViewerInitializer(runConfig, ModuleFactories));
     }
     if (serviceMask.EnableLoadService) {
         sil->AddServiceInitializer(new TLoadInitializer(runConfig));
diff --git a/ydb/core/viewer/json_pqconsumerinfo.h b/ydb/core/viewer/json_pqconsumerinfo.h
index 722fbce8d42..5ccabdb3786 100644
--- a/ydb/core/viewer/json_pqconsumerinfo.h
+++ b/ydb/core/viewer/json_pqconsumerinfo.h
@@ -14,10 +14,12 @@ namespace NViewer {
 
 using namespace NActors;
 
+
 class TJsonPQConsumerInfo : public TActorBootstrapped<TJsonPQConsumerInfo> {
     using TBase = TActorBootstrapped<TJsonPQConsumerInfo>;
     IViewer* Viewer;
     NMon::TEvHttpInfo::TPtr Event;
+    std::shared_ptr<NMsgBusProxy::IPersQueueGetReadSessionsInfoWorkerFactory> PQReadSessionsInfoWorkerFactory;
     NKikimrClient::TResponse Result;
     TJsonSettings JsonSettings;
     TString Topic;
@@ -33,9 +35,14 @@ public:
         return NKikimrServices::TActivity::VIEWER_HANDLER;
     }
 
-    TJsonPQConsumerInfo(IViewer* viewer, NMon::TEvHttpInfo::TPtr &ev)
+    TJsonPQConsumerInfo(
+        IViewer* viewer,
+        NMon::TEvHttpInfo::TPtr& ev,
+        std::shared_ptr<NMsgBusProxy::IPersQueueGetReadSessionsInfoWorkerFactory> pqReadSessionsInfoWorkerFactory = nullptr
+    )
         : Viewer(viewer)
         , Event(ev)
+        , PQReadSessionsInfoWorkerFactory(pqReadSessionsInfoWorkerFactory)
     {}
 
     void Bootstrap(const TActorContext& ctx) {
@@ -61,14 +68,24 @@ public:
             NKikimrClient::TPersQueueRequest request;
             request.MutableMetaRequest()->MutableCmdGetPartitionStatus()->SetClientId(Client);
             request.MutableMetaRequest()->MutableCmdGetPartitionStatus()->AddTopicRequest()->SetTopic(Topic);
-            ctx.Register(NMsgBusProxy::CreateActorServerPersQueue(ctx.SelfID, request, NMsgBusProxy::CreatePersQueueMetaCacheV2Id(), nullptr));
+            ctx.Register(NMsgBusProxy::CreateActorServerPersQueue(
+                ctx.SelfID,
+                request,
+                NMsgBusProxy::CreatePersQueueMetaCacheV2Id(),
+                PQReadSessionsInfoWorkerFactory
+            ));
             ++Requests;
         }
         {
             NKikimrClient::TPersQueueRequest request;
             request.MutableMetaRequest()->MutableCmdGetReadSessionsInfo()->SetClientId(Client);
             request.MutableMetaRequest()->MutableCmdGetReadSessionsInfo()->AddTopic(Topic);
-            ctx.Register(NMsgBusProxy::CreateActorServerPersQueue(ctx.SelfID, request, NMsgBusProxy::CreatePersQueueMetaCacheV2Id(), nullptr));
+            ctx.Register(NMsgBusProxy::CreateActorServerPersQueue(
+                ctx.SelfID,
+                request,
+                NMsgBusProxy::CreatePersQueueMetaCacheV2Id(),
+                PQReadSessionsInfoWorkerFactory
+            ));
             ++Requests;
         }
         Become(&TThis::StateRequestedTopicInfo, ctx, TDuration::MilliSeconds(Timeout), new TEvents::TEvWakeup());
diff --git a/ydb/core/viewer/viewer.cpp b/ydb/core/viewer/viewer.cpp
index c1d16c97b84..f70f84276a8 100644
--- a/ydb/core/viewer/viewer.cpp
+++ b/ydb/core/viewer/viewer.cpp
@@ -92,6 +92,23 @@ public:
     }
 };
 
+template <typename ActorRequestType>
+class TPersQueueJsonHandler : public TJsonHandler<ActorRequestType> {
+public:
+    TPersQueueJsonHandler(
+        std::shared_ptr<NMsgBusProxy::IPersQueueGetReadSessionsInfoWorkerFactory> pqReadSessionsInfoWorkerFactory
+    )
+        : PQReadSessionsInfoWorkerFactory(pqReadSessionsInfoWorkerFactory)
+    {}
+    
+    IActor* CreateRequestActor(IViewer* viewer, NMon::TEvHttpInfo::TPtr& event) override {
+        return new ActorRequestType(viewer, event, PQReadSessionsInfoWorkerFactory);
+    }
+
+private:
+    std::shared_ptr<NMsgBusProxy::IPersQueueGetReadSessionsInfoWorkerFactory> PQReadSessionsInfoWorkerFactory;
+};
+
 void SetupPQVirtualHandlers(IViewer* viewer) {
     viewer->RegisterVirtualHandler(
         NKikimrViewer::EObjectType::Root,
@@ -129,8 +146,12 @@ public:
         return NKikimrServices::TActivity::TABLET_MONITORING_PROXY;
     }
 
-    TViewer(const TKikimrRunConfig &kikimrRunConfig)
+    TViewer(
+        const TKikimrRunConfig &kikimrRunConfig,
+        std::shared_ptr<NMsgBusProxy::IPersQueueGetReadSessionsInfoWorkerFactory> pqReadSessionsInfoWorkerFactory
+    )
         : KikimrRunConfig(kikimrRunConfig)
+        , PQReadSessionsInfoWorkerFactory(pqReadSessionsInfoWorkerFactory)
     {}
 
     void Bootstrap(const TActorContext &ctx) {
@@ -191,7 +212,9 @@ public:
             JsonHandlers["/json/config"] = new TJsonHandler<TJsonConfig>;
             JsonHandlers["/json/counters"] = new TJsonHandler<TJsonCounters>;
             JsonHandlers["/json/topicinfo"] = new TJsonHandler<TJsonTopicInfo>;
-            JsonHandlers["/json/pqconsumerinfo"] = new TJsonHandler<TJsonPQConsumerInfo>;
+            JsonHandlers["/json/pqconsumerinfo"] = new TPersQueueJsonHandler<TJsonPQConsumerInfo>(
+                PQReadSessionsInfoWorkerFactory
+            );
             JsonHandlers["/json/tabletcounters"] = new TJsonHandler<TJsonTabletCounters>;
             JsonHandlers["/json/storage"] = new TJsonHandler<TJsonStorage>;
             JsonHandlers["/json/metainfo"] = new TJsonHandler<TJsonMetaInfo>;
@@ -249,6 +272,7 @@ public:
 private:
     THashMap<TString, TAutoPtr<TJsonHandlerBase>> JsonHandlers;
     const TKikimrRunConfig KikimrRunConfig;
+    std::shared_ptr<NMsgBusProxy::IPersQueueGetReadSessionsInfoWorkerFactory> PQReadSessionsInfoWorkerFactory;
     std::unordered_multimap<NKikimrViewer::EObjectType, TVirtualHandler> VirtualHandlersByParentType;
     std::unordered_map<NKikimrViewer::EObjectType, TContentHandler> ContentHandlers;
     TString AllowOrigin;
@@ -506,9 +530,12 @@ TString IViewer::TContentRequestContext::Dump() const
 
 ui32 CurrentMonitoringPort = 8765;
 
-IActor* CreateViewer(const TKikimrRunConfig &kikimrRunConfig) {
+IActor* CreateViewer(
+    const TKikimrRunConfig &kikimrRunConfig,
+    std::shared_ptr<NMsgBusProxy::IPersQueueGetReadSessionsInfoWorkerFactory> pqReadSessionsInfoWorkerFactory
+) {
     CurrentMonitoringPort = kikimrRunConfig.AppConfig.GetMonitoringConfig().GetMonitoringPort();
-    return new TViewer(kikimrRunConfig);
+    return new TViewer(kikimrRunConfig, pqReadSessionsInfoWorkerFactory);
 }
 
 TString TViewer::GetHTTPOKJSON(const NMon::TEvHttpInfo* request) {
diff --git a/ydb/core/viewer/viewer.h b/ydb/core/viewer/viewer.h
index 5b14a9e4a5f..f5544c78f8a 100644
--- a/ydb/core/viewer/viewer.h
+++ b/ydb/core/viewer/viewer.h
@@ -18,7 +18,10 @@ inline TActorId MakeViewerID(ui32 node = 0) {
     return TActorId(node, TStringBuf(x, 12));
 }
 
-IActor* CreateViewer(const TKikimrRunConfig &kikimrRunConfig);
+IActor* CreateViewer(
+    const TKikimrRunConfig &kikimrRunConfig,    
+    std::shared_ptr<NMsgBusProxy::IPersQueueGetReadSessionsInfoWorkerFactory> pqReadSessionsInfoWorkerFactory
+);
 
 class IViewer {
 public: