YQ-4848 Rebalance partitions after new nodes connected (#28397)

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

Author: kardymonds

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) {

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

@@ -28,6 +28,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

@@ -54,6 +54,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 {