fix initialization

Author: nshestakov

ydb/core/persqueue/pq_impl.cpp

@@ -1781,10 +1781,12 @@ void TPersQueue::Handle(TEvPersQueue::TEvStatus::TPtr& ev, const TActorContext&
     }
 
     ui32 cnt = 0;
-    for (auto& p : Partitions) {
-         cnt += p.second.InitDone;
+    for (auto& [_, partitionInfo] : Partitions) {
+         cnt += partitionInfo.InitDone;
     }
 
+    Cerr << ">>>>> InitDone=" << cnt << Endl;
+
     TActorId ans = CreateStatusProxyActor(TabletID(), ev->Sender, cnt, ev->Cookie, ctx);
     for (auto& p : Partitions) {
         if (!p.second.InitDone)

ydb/core/persqueue/read_balancer.cpp

@@ -768,6 +768,8 @@ void TPersQueueReadBalancer::RequestTabletIfNeeded(const ui64 tabletId, const TA
                 AggregatedStats.Cookies[tabletId] = cookie;
             }
 
+            LOG_DEBUG(ctx, NKikimrServices::PERSQUEUE_READ_BALANCER, 
+                TStringBuilder() << "Send TEvPersQueue::TEvStatus TabletId: " << tabletId << " Cookie: " << cookie);
             NTabletPipe::SendData(ctx, pipeClient, new TEvPersQueue::TEvStatus("", true), cookie);
         }
         NTabletPipe::SendData(ctx, pipeClient, new TEvPQ::TEvSubDomainStatus(SubDomainOutOfSpace));
@@ -780,7 +782,6 @@ void TPersQueueReadBalancer::Handle(TEvPersQueue::TEvStatusResponse::TPtr& ev, c
     ui64 tabletId = record.GetTabletId();
     ui64 cookie = ev->Cookie;
 
-
     if ((0 != cookie && cookie != AggregatedStats.Cookies[tabletId]) || (0 == cookie && !AggregatedStats.Cookies.contains(tabletId))) {
         return;
     }

ydb/core/persqueue/read_balancer.h

@@ -257,7 +257,14 @@ class TPersQueueReadBalancer : public TActor<TPersQueueReadBalancer>, public TTa
         }
         RegisterEvents.clear();
 
-        ctx.Send(ctx.SelfID, new TEvents::TEvWakeup());
+        for (auto& [_, partitionInfo] : PartitionsInfo) {
+            RequestTabletIfNeeded(partitionInfo.TabletId, ctx, true);
+        }
+
+        auto wakeupInterval = AppData(ctx)->PQConfig.GetBalancerWakeupIntervalSec();
+        Y_ABORT_UNLESS(0 < wakeupInterval);
+        ctx.Schedule(TDuration::Seconds(wakeupInterval), new TEvents::TEvWakeup());
+
         ctx.Send(ctx.SelfID, new TEvPersQueue::TEvUpdateACL());
     }
 

ydb/core/tx/schemeshard/ut_stats/ut_stats.cpp

@@ -303,6 +303,7 @@ Y_UNIT_TEST_SUITE(TSchemeshardStatsBatchingTest) {
         TTestEnv env(runtime);
 
         runtime.SetLogPriority(NKikimrServices::FLAT_TX_SCHEMESHARD, NActors::NLog::PRI_TRACE);
+        runtime.SetLogPriority(NKikimrServices::PERSQUEUE_READ_BALANCER, NActors::NLog::PRI_TRACE);
 
         auto& appData = runtime.GetAppData();