YQ-4848 Rebalance partitions after new nodes connected

ydb/core/fq/libs/actors/nodes_manager.cpp

@@ -297,7 +297,7 @@ class TNodesManagerActor : public NActors::TActorBootstrapped<TNodesManagerActor
     }
 
     void Handle(NFq::TEvNodesManager::TEvGetNodesRequest::TPtr& ev) {
-        LOG_D("TNodesManagerActor::TEvGetNodesRequest");
+        LOG_T("Received TNodesManagerActor::TEvGetNodesRequest");
         auto response = MakeHolder<NFq::TEvNodesManager::TEvGetNodesResponse>();
         response->NodeIds.reserve(Peers.size());
         for (const auto& info : Peers) {
@@ -347,6 +347,8 @@ class TNodesManagerActor : public NActors::TActorBootstrapped<TNodesManagerActor
     }
 
     void HandleResponse(NFq::TEvInternalService::TEvHealthCheckResponse::TPtr& ev) {
+        LOG_T("Received TEvHealthCheckResponse");
+
         try {
             const auto& status = ev->Get()->Status.GetStatus();
             if (!ev->Get()->Status.IsSuccess()) {
@@ -356,6 +358,8 @@ class TNodesManagerActor : public NActors::TActorBootstrapped<TNodesManagerActor
 
             auto nodesInfo = MakeIntrusive<TIntrusiveVector<TEvInterconnect::TNodeInfo>>();
             nodesInfo->reserve(res.nodes().size());
+            LOG_T("Nodes count " << res.nodes().size());
+
 
             Peers.clear();
             std::set<ui32> nodeIds; // may be not unique

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

@@ -16,7 +16,8 @@ message TRowDispatcherCoordinatorConfig {
     // Topic partitions will be distributed uniformly up to TopicPartitionsLimitPerNode
     // if (number nodes) * TopicPartitionsLimitPerNode < (number topic partitions)
     // Request will hang up infinitely, disabled by default
-    uint64 TopicPartitionsLimitPerNode = 4;  
+    uint64 TopicPartitionsLimitPerNode = 4; // deprecated
+    uint64 RebalancingTimeoutSec = 5;       // Automatic rebalancing partition after new nodes connected / nodes disconnected.
 }
 
 message TJsonParserConfig {

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

@@ -29,6 +29,7 @@ TRowDispatcherSettings::TCoordinatorSettings::TCoordinatorSettings(const NConfig
     : LocalMode(config.GetLocalMode())
     , Database(config.GetDatabase())
     , CoordinationNodePath(config.GetCoordinationNodePath())
+    , RebalancingTimeout(TDuration::Seconds(config.GetRebalancingTimeoutSec()))
 {}
 
 TRowDispatcherSettings::TCoordinatorSettings::TCoordinatorSettings(const NKikimrConfig::TStreamingQueriesConfig::TExternalStorageConfig& config)

ydb/core/fq/libs/row_dispatcher/common/row_dispatcher_settings.h

@@ -55,6 +55,7 @@ class TRowDispatcherSettings {
         YDB_ACCESSOR(bool, LocalMode, false);
         YDB_ACCESSOR_MUTABLE(TExternalStorageSettings, Database, {});
         YDB_ACCESSOR_DEF(TString, CoordinationNodePath);
+        YDB_ACCESSOR(TDuration, RebalancingTimeout, TDuration::Seconds(120));
     };
 
     enum class EConsumerMode {