Fixed kafka not present in metadata (#47348)

13 files changed, 622 insertions, 95 deletions

diff --git a/ydb/core/kafka_proxy/actors/actors.h b/ydb/core/kafka_proxy/actors/actors.h
index 01cef480536..a6e9f9d3955 100644
--- a/ydb/core/kafka_proxy/actors/actors.h
+++ b/ydb/core/kafka_proxy/actors/actors.h
@@ -117,6 +117,8 @@ inline EKafkaErrors ConvertErrorCode(Ydb::StatusIds::StatusCode status) {
             return EKafkaErrors::UNKNOWN_TOPIC_OR_PARTITION;
         case Ydb::StatusIds::UNAUTHORIZED:
             return EKafkaErrors::TOPIC_AUTHORIZATION_FAILED;
+        case Ydb::StatusIds::TIMEOUT:
+            return EKafkaErrors::REQUEST_TIMED_OUT;
         default:
             return EKafkaErrors::UNKNOWN_SERVER_ERROR;
     }
diff --git a/ydb/core/kafka_proxy/actors/kafka_metadata_actor.cpp b/ydb/core/kafka_proxy/actors/kafka_metadata_actor.cpp
index 14f27c53907..1f1d9f629d3 100644
--- a/ydb/core/kafka_proxy/actors/kafka_metadata_actor.cpp
+++ b/ydb/core/kafka_proxy/actors/kafka_metadata_actor.cpp
@@ -1,5 +1,6 @@
 #include "kafka_metadata_actor.h"
 
+#include <ydb/core/actorlib_impl/long_timer.h>
 #include <ydb/core/base/statestorage.h>
 #include <ydb/core/grpc_services/grpc_endpoint.h>
 #include <ydb/core/kafka_proxy/actors/kafka_create_topics_actor.h>
@@ -50,6 +51,8 @@ void TKafkaMetadataActor::Bootstrap(const TActorContext& ctx) {
     }
 
     Become(&TKafkaMetadataActor::StateWork);
+    TimeoutTimerActorId = CreateLongTimer(ctx, RequestTimeout,
+        new IEventHandle(SelfId(), SelfId(), new TEvents::TEvWakeup()));
     RespondIfRequired(ctx);
 }
 
@@ -283,7 +286,12 @@ void TKafkaMetadataActor::HandleLocationResponse(TEvLocationResponse::TPtr ev, c
                 {"topicName", topic.Name},
                 {"code", locationResponse->Status},
                 {"issues", locationResponse->Issues.ToOneLineString()});
-            AddTopicError(topic, ConvertErrorCode(locationResponse->Status));
+            // Transient location failures (pipe retries exhausted / locations backoff) → retriable timeout.
+            const EKafkaErrors kafkaError =
+                (status == Ydb::StatusIds::UNAVAILABLE || status == Ydb::StatusIds::INTERNAL_ERROR)
+                    ? EKafkaErrors::REQUEST_TIMED_OUT
+                    : ConvertErrorCode(status);
+            AddTopicError(topic, kafkaError);
         }
     }
     if (InflyCreateTopics == 0) {
@@ -300,7 +308,8 @@ void TKafkaMetadataActor::Handle(const TEvKafka::TEvResponse::TPtr& ev, const TA
     InflyCreateTopics--;
     PendingResponses--;
     EKafkaErrors errorCode = ev->Get()->ErrorCode;
-    if (errorCode == EKafkaErrors::NONE_ERROR) {
+    if (errorCode == EKafkaErrors::NONE_ERROR || errorCode == EKafkaErrors::TOPIC_ALREADY_EXISTS) {
+        // Topic is available (created by us or raced with another create) — describe location.
         TActorId child = SendTopicRequest(topicName);
         TopicIndexes[child].push_back(topicIndex);
     } else {
@@ -346,8 +355,31 @@ void TKafkaMetadataActor::AddBroker(ui64 nodeId, const TString& host, ui64 port)
     }
 }
 
+void TKafkaMetadataActor::ApplyPendingTopicResponses() {
+    while (!PendingTopicResponses.empty()) {
+        auto& [index, ev] = *PendingTopicResponses.begin();
+        auto& topic = Response->Topics[index];
+        if (!WithProxy) {
+            auto topicNodes = CheckTopicNodes(ev.Get());
+            if (topicNodes.empty()) {
+                // Already tried YDB discovery. Throw error
+                YDB_LOG_ERROR("Could not discovery kafka port for topic",
+                    {LogPrefix()},
+                    {"topicName", topic.Name});
+                AddTopicError(topic, EKafkaErrors::LISTENER_NOT_FOUND);
+            } else {
+                AddTopicResponse(topic, ev.Get(), topicNodes);
+            }
+        } else {
+            AddTopicResponse(topic, ev.Get(), {});
+        }
+        PendingTopicResponses.erase(PendingTopicResponses.begin());
+    }
+}
+
 void TKafkaMetadataActor::RespondIfRequired(const TActorContext& ctx) {
     auto Respond = [&] {
+        CancelRequestTimeout();
         Send(Context->ConnectionId, new TEvKafka::TEvResponse(CorrelationId, Response, ErrorCode));
         Die(ctx);
     };
@@ -364,25 +396,7 @@ void TKafkaMetadataActor::RespondIfRequired(const TActorContext& ctx) {
         return;
     }
 
-    while (!PendingTopicResponses.empty()) {
-        auto& [index, ev] = *PendingTopicResponses.begin();
-        auto& topic = Response->Topics[index];
-        if (!WithProxy) {
-            auto topicNodes = CheckTopicNodes(ev.Get());
-            if (topicNodes.empty()) {
-                    // Already tried YDB discovery. Throw error
-                    YDB_LOG_ERROR("Could not discovery kafka port for topic",
-                        {LogPrefix()},
-                        {"topicName", topic.Name});
-                    AddTopicError(topic, EKafkaErrors::LISTENER_NOT_FOUND);
-            } else {
-                AddTopicResponse(topic, ev.Get(), topicNodes);
-            }
-        } else {
-            AddTopicResponse(topic, ev.Get(), {});
-        }
-        PendingTopicResponses.erase(PendingTopicResponses.begin());
-    }
+    ApplyPendingTopicResponses();
 
     if (NeedAllNodes) {
         for (const auto& [id, nodeInfo] : Nodes)
@@ -392,6 +406,38 @@ void TKafkaMetadataActor::RespondIfRequired(const TActorContext& ctx) {
     Respond();
 }
 
+void TKafkaMetadataActor::HandleWakeup(TEvents::TEvWakeup::TPtr&, const TActorContext& ctx) {
+    TimeoutTimerActorId = {};
+    YDB_LOG_ERROR("Metadata request timed out",
+        {LogPrefix()},
+        {"correlationId", CorrelationId},
+        {"pendingResponses", PendingResponses});
+    RespondWithTimeout(ctx);
+}
+
+void TKafkaMetadataActor::RespondWithTimeout(const TActorContext& ctx) {
+    ApplyPendingTopicResponses();
+
+    ErrorCode = EKafkaErrors::REQUEST_TIMED_OUT;
+    for (auto& topic : Response->Topics) {
+        // Keep already completed topics (success or earlier error); fail only unfinished ones.
+        if (topic.ErrorCode == EKafkaErrors::NONE_ERROR && topic.Partitions.empty()) {
+            topic.ErrorCode = EKafkaErrors::REQUEST_TIMED_OUT;
+        }
+    }
+
+    CancelRequestTimeout();
+    Send(Context->ConnectionId, new TEvKafka::TEvResponse(CorrelationId, Response, ErrorCode));
+    Die(ctx);
+}
+
+void TKafkaMetadataActor::CancelRequestTimeout() {
+    if (TimeoutTimerActorId) {
+        Send(TimeoutTimerActorId, new TEvents::TEvPoison());
+        TimeoutTimerActorId = {};
+    }
+}
+
 NStructuredLog::TStructuredMessage TKafkaMetadataActor::LogPrefix() const {
     return YDB_LOG_CREATE_MESSAGE(
         {"actorClassName", "TKafkaMetadataActor"},
diff --git a/ydb/core/kafka_proxy/actors/kafka_metadata_actor.h b/ydb/core/kafka_proxy/actors/kafka_metadata_actor.h
index d7ac9c8f928..89d74574a0d 100644
--- a/ydb/core/kafka_proxy/actors/kafka_metadata_actor.h
+++ b/ydb/core/kafka_proxy/actors/kafka_metadata_actor.h
@@ -52,6 +52,7 @@ private:
     void AddTopicResponse(TMetadataResponseData::TMetadataResponseTopic& topic, TEvLocationResponse* response,
                           const TVector<TNodeInfo*>& nodes);
     void AddTopicError(TMetadataResponseData::TMetadataResponseTopic& topic, EKafkaErrors errorCode);
+    void ApplyPendingTopicResponses();
     void RespondIfRequired(const NActors::TActorContext& ctx);
     void AddProxyNodeToBrokers();
     void AddBroker(ui64 nodeId, const TString& host, ui64 port);
@@ -63,6 +64,10 @@ private:
 
     void AddTopic(const TString& topic, ui64 index);
 
+    void HandleWakeup(TEvents::TEvWakeup::TPtr& ev, const NActors::TActorContext& ctx);
+    void RespondWithTimeout(const NActors::TActorContext& ctx);
+    void CancelRequestTimeout();
+
     STATEFN(StateWork) {
         switch (ev->GetTypeRewrite()) {
             HFunc(TEvLocationResponse, HandleLocationResponse);
@@ -70,6 +75,7 @@ private:
             hFunc(NKikimr::TEvDiscovery::TEvError, HandleDiscoveryError);
             hFunc(NKikimr::TEvPQ::TEvListAllTopicsResponse, HandleListTopics);
             HFunc(TEvKafka::TEvResponse, Handle);
+            HFunc(TEvents::TEvWakeup, HandleWakeup);
         }
     }
 
@@ -77,6 +83,8 @@ private:
     NStructuredLog::TStructuredMessage LogPrefix() const;
 
 private:
+    static constexpr TDuration RequestTimeout = TDuration::Seconds(30);
+
     const TContext::TPtr Context;
     const ui64 CorrelationId;
     const TMessagePtr<TMetadataRequestData> Message;
@@ -91,6 +99,7 @@ private:
     EKafkaErrors ErrorCode = EKafkaErrors::NONE_ERROR;
 
     TActorId DiscoveryCacheActor;
+    TActorId TimeoutTimerActorId;
     bool NeedAllNodes = false;
     bool HaveError = false;
     TMap<ui64, TSimpleSharedPtr<TEvLocationResponse>> PendingTopicResponses;
diff --git a/ydb/core/persqueue/events/global.h b/ydb/core/persqueue/events/global.h
index 77f0f5c6fe2..9cfcd5fb4b8 100644
--- a/ydb/core/persqueue/events/global.h
+++ b/ydb/core/persqueue/events/global.h
@@ -131,10 +131,13 @@ namespace NKikimr::TEvPersQueue {
 
     struct TEvGetPartitionsLocation: public TEventPB<TEvGetPartitionsLocation,
             NKikimrPQ::TGetPartitionsLocation, EvGetPartitionsLocation> {
-            TEvGetPartitionsLocation(const TVector<ui64>& partitionIds = {}) {
+            TEvGetPartitionsLocation(const TVector<ui64>& partitionIds = {}, TDuration timeout = {}) {
                 for (const auto& p : partitionIds) {
                     Record.AddPartitions(p);
                 }
+                if (timeout) {
+                    Record.SetTimeoutMs(timeout.MilliSeconds());
+                }
             }
     };
 
diff --git a/ydb/core/persqueue/pqrb/read_balancer.cpp b/ydb/core/persqueue/pqrb/read_balancer.cpp
index 5eafeb10d21..a1bb23dbf10 100644
--- a/ydb/core/persqueue/pqrb/read_balancer.cpp
+++ b/ydb/core/persqueue/pqrb/read_balancer.cpp
@@ -97,6 +97,12 @@ void TPersQueueReadBalancer::Die(const TActorContext& ctx) {
         NTabletPipe::CloseClient(ctx, pipe.second.PipeActor);
     }
     TabletPipes.clear();
+    PipesRequested.clear();
+    ReadyPartitionTablets = 0;
+    while (!PartitionsLocationQueue.empty()) {
+        SendPartitionsLocationError(PartitionsLocationQueue.front().Sender, ctx);
+        PartitionsLocationQueue.pop_front();
+    }
     if (PartitionsScaleManager) {
         PartitionsScaleManager->Die(ctx);
     }
@@ -162,6 +168,8 @@ void TPersQueueReadBalancer::InitDone(const TActorContext &ctx) {
 
     auto wakeupInterval = std::max<ui64>(AppData(ctx)->PQConfig.GetBalancerWakeupIntervalSec(), 1);
     ctx.Schedule(TDuration::Seconds(wakeupInterval), new TEvents::TEvWakeup());
+
+    ProcessPartitionsLocationQueue(ctx);
 }
 
 void TPersQueueReadBalancer::HandleWakeup(TEvents::TEvWakeup::TPtr& ev, const TActorContext &ctx) {
@@ -175,7 +183,13 @@ void TPersQueueReadBalancer::HandleWakeup(TEvents::TEvWakeup::TPtr& ev, const TA
             }
             break;
         }
+        case PARTITIONS_LOCATION_WAKEUP_TAG: {
+            PartitionsLocationWakeupScheduled = false;
+            ProcessPartitionsLocationQueue(ctx);
+            break;
+        }
         default: {
+            ProcessPartitionsLocationQueue(ctx);
             GetStat(ctx); //TODO: do it only on signals from outerspace right now
             CleanupReceiveAttemptPartitions(ctx);
             auto wakeupInterval = std::max<ui64>(AppData(ctx)->PQConfig.GetBalancerWakeupIntervalSec(), 1);
@@ -416,22 +430,34 @@ void TPersQueueReadBalancer::Handle(TEvTabletPipe::TEvClientConnected::TPtr& ev,
         it->second.Generation = ev->Get()->Generation;
         it->second.NodeId = ev->Get()->ServerId.NodeId();
 
+        if (!it->second.Ready && TabletsInfo.contains(tabletId)) {
+            it->second.Ready = true;
+            ++ReadyPartitionTablets;
+        }
+
         YDB_LOG_DEBUG("TEvClientConnected TabletId NodeId Generation",
             {"logPrefix", LogPrefix()},
             {"tabletId", tabletId},
             {"nodeId", ev->Get()->ServerId.NodeId()},
             {"generation", ev->Get()->Generation});
-    }
-    else
+    } else {
         YDB_LOG_INFO("TEvClientConnected Pipe is not found, TabletId",
             {"logPrefix", LogPrefix()},
             {"tabletId", tabletId});
+    }
+
+    ProcessPartitionsLocationQueue(ctx);
 }
 
 void TPersQueueReadBalancer::ClosePipe(const ui64 tabletId, const TActorContext& ctx)
 {
     auto it = TabletPipes.find(tabletId);
     if (it != TabletPipes.end()) {
+        if (it->second.Ready) {
+            PQ_ENSURE(ReadyPartitionTablets > 0);
+            --ReadyPartitionTablets;
+            it->second.Ready = false;
+        }
         NTabletPipe::CloseClient(ctx, it->second.PipeActor);
         TabletPipes.erase(it);
         PipesRequested.erase(tabletId);
@@ -622,74 +648,6 @@ void TPersQueueReadBalancer::GetStat(const TActorContext& ctx) {
     }
 }
 
-void TPersQueueReadBalancer::HandleOnInit(TEvPersQueue::TEvGetPartitionsLocation::TPtr& ev, const TActorContext& ctx) {
-    auto* evResponse = new TEvPersQueue::TEvGetPartitionsLocationResponse();
-    evResponse->Record.SetStatus(false);
-    ctx.Send(ev->Sender, evResponse);
-}
-
-void TPersQueueReadBalancer::Handle(TEvPersQueue::TEvGetPartitionsLocation::TPtr& ev, const TActorContext& ctx) {
-    const auto& request = ev->Get()->Record;
-    auto evResponse = std::make_unique<TEvPersQueue::TEvGetPartitionsLocationResponse>();
-
-    auto addPartitionToResponse = [&](ui64 partitionId, ui64 tabletId) {
-        if (PipesRequested.contains(tabletId)) {
-            return false;
-        }
-        auto iter = TabletPipes.find(tabletId);
-        if (iter == TabletPipes.end()) {
-            GetPipeClient(tabletId, ctx);
-            return false;
-        }
-
-        auto* pResponse = evResponse->Record.AddLocations();
-        pResponse->SetPartitionId(partitionId);
-        pResponse->SetNodeId(iter->second.NodeId.GetRef());
-        pResponse->SetGeneration(iter->second.Generation.GetRef());
-
-        YDB_LOG_DEBUG("The partition location was added to response: TabletId PartitionId NodeId Generation",
-            {"logPrefix", LogPrefix()},
-            {"tabletId", tabletId},
-            {"partitionId", partitionId},
-            {"pResponseNodeId", pResponse->GetNodeId()},
-            {"pResponseGeneration", pResponse->GetGeneration()});
-
-        return true;
-    };
-
-    auto sendError = [&]() {
-        auto response = std::make_unique<TEvPersQueue::TEvGetPartitionsLocationResponse>();
-        response->Record.SetStatus(false);
-        ctx.Send(ev->Sender, response.release());
-    };
-
-    if (request.PartitionsSize() == 0) {
-        if (!PipesRequested.empty() || TabletPipes.size() < TabletsInfo.size()) {
-            // Do not have all pipes connected.
-            return sendError();
-        }
-        for (const auto& [partitionId, partitionInfo] : PartitionsInfo) {
-            if (!addPartitionToResponse(partitionId, partitionInfo.TabletId)) {
-                return sendError();
-            }
-        }
-    } else {
-        for (const auto& partitionInRequest : request.GetPartitions()) {
-            auto partitionInfoIter = PartitionsInfo.find(partitionInRequest);
-            if (partitionInfoIter == PartitionsInfo.end()) {
-                return sendError();
-            }
-            if (!addPartitionToResponse(partitionInRequest, partitionInfoIter->second.TabletId)) {
-                return sendError();
-            }
-        }
-    }
-
-    evResponse->Record.SetStatus(true);
-    ctx.Send(ev->Sender, evResponse.release());
-}
-
-
 
 
 //
@@ -1142,6 +1100,7 @@ STFUNC(TPersQueueReadBalancer::StateInit) {
         HFunc(NSchemeShard::TEvSchemeShard::TEvSubDomainPathIdFound, Handle);
         HFunc(TEvTxProxySchemeCache::TEvWatchNotifyUpdated, Handle);
         HFunc(TEvPersQueue::TEvGetPartitionsLocation, HandleOnInit);
+        HFunc(TEvents::TEvWakeup, HandleWakeup);
         // MLP
         hFunc(TEvPQ::TEvMLPGetPartitionRequest, Handle);
         HFunc(TEvPQ::TEvMLPConsumerStatus, Handle);
diff --git a/ydb/core/persqueue/pqrb/read_balancer.h b/ydb/core/persqueue/pqrb/read_balancer.h
index 819f48007cb..f6f911629ad 100644
--- a/ydb/core/persqueue/pqrb/read_balancer.h
+++ b/ydb/core/persqueue/pqrb/read_balancer.h
@@ -120,6 +120,12 @@ class TPersQueueReadBalancer : public TActor<TPersQueueReadBalancer>,
 
     void HandleOnInit(TEvPersQueue::TEvGetPartitionsLocation::TPtr& ev, const TActorContext& ctx);
     void Handle(TEvPersQueue::TEvGetPartitionsLocation::TPtr& ev, const TActorContext& ctx);
+    void EnqueuePartitionsLocationRequest(TEvPersQueue::TEvGetPartitionsLocation::TPtr& ev, const TActorContext& ctx);
+    void ProcessPartitionsLocationQueue(const TActorContext& ctx);
+    bool TryRespondPartitionsLocation(const TActorId& sender, const NKikimrPQ::TGetPartitionsLocation& request, const TActorContext& ctx);
+    bool AllPartitionPipesReady() const;
+    void SchedulePartitionsLocationWakeup(const TActorContext& ctx);
+    void SendPartitionsLocationError(const TActorId& sender, const TActorContext& ctx);
 
     void Handle(TEvPersQueue::TEvGetPartitionIdForWrite::TPtr&, const TActorContext&);
 
@@ -254,10 +260,12 @@ private:
         TActorId PipeActor;
         TMaybe<ui64> NodeId;
         TMaybe<ui32> Generation;
+        bool Ready = false;
     };
 
     std::unordered_map<ui64, TPipeLocation> TabletPipes;
     std::unordered_set<ui64> PipesRequested;
+    ui32 ReadyPartitionTablets = 0;
 
     TDatabaseInfo DatabaseInfo;
 
@@ -280,6 +288,16 @@ private:
     std::deque<TAutoPtr<TEvPersQueue::TEvRegisterReadSession>> RegisterEvents;
     std::deque<TAutoPtr<TEvPersQueue::TEvUpdateBalancerConfig>> UpdateEvents;
 
+    static constexpr ui64 PARTITIONS_LOCATION_WAKEUP_TAG = 11;
+
+    struct TPartitionsLocationRequest {
+        TActorId Sender;
+        NKikimrPQ::TGetPartitionsLocation Record;
+        TInstant Deadline;
+    };
+    std::deque<TPartitionsLocationRequest> PartitionsLocationQueue;
+    bool PartitionsLocationWakeupScheduled = false;
+
     using TMLPRequests = std::variant<
         TEvPQ::TEvMLPGetRuntimeAttributesRequest::TPtr
     >;
diff --git a/ydb/core/persqueue/pqrb/read_balancer_partition_location.cpp b/ydb/core/persqueue/pqrb/read_balancer_partition_location.cpp
new file mode 100644
index 00000000000..a91f4c725a1
--- /dev/null
+++ b/ydb/core/persqueue/pqrb/read_balancer_partition_location.cpp
@@ -0,0 +1,172 @@
+#include "read_balancer.h"
+#include "read_balancer_log.h"
+
+#define YDB_LOG_THIS_FILE_COMPONENT NKikimrServices::PERSQUEUE_READ_BALANCER
+
+namespace NKikimr::NPQ {
+
+void TPersQueueReadBalancer::HandleOnInit(
+    TEvPersQueue::TEvGetPartitionsLocation::TPtr& ev,
+    const TActorContext& ctx)
+{
+    EnqueuePartitionsLocationRequest(ev, ctx);
+}
+
+void TPersQueueReadBalancer::SendPartitionsLocationError(
+    const TActorId& sender,
+    const TActorContext& ctx)
+{
+    auto response = std::make_unique<TEvPersQueue::TEvGetPartitionsLocationResponse>();
+    response->Record.SetStatus(false);
+    ctx.Send(sender, response.release());
+}
+
+bool TPersQueueReadBalancer::AllPartitionPipesReady() const
+{
+    return !TabletsInfo.empty() && ReadyPartitionTablets == TabletsInfo.size();
+}
+
+void TPersQueueReadBalancer::SchedulePartitionsLocationWakeup(const TActorContext& ctx)
+{
+    if (PartitionsLocationWakeupScheduled || PartitionsLocationQueue.empty()) {
+        return;
+    }
+
+    const auto now = TAppData::TimeProvider->Now();
+    const auto& deadline = PartitionsLocationQueue.front().Deadline;
+    const auto delay = deadline > now
+        ? std::max(deadline - now, TDuration::MilliSeconds(50))
+        : TDuration::MilliSeconds(50);
+
+    PartitionsLocationWakeupScheduled = true;
+    ctx.Schedule(delay, new TEvents::TEvWakeup(PARTITIONS_LOCATION_WAKEUP_TAG));
+}
+
+void TPersQueueReadBalancer::EnqueuePartitionsLocationRequest(
+    TEvPersQueue::TEvGetPartitionsLocation::TPtr& ev,
+    const TActorContext& ctx)
+{
+    const auto timeout = TDuration::MilliSeconds(ev->Get()->Record.GetTimeoutMs());
+
+    PartitionsLocationQueue.push_back(TPartitionsLocationRequest{
+        .Sender = ev->Sender,
+        .Record = std::move(ev->Get()->Record),
+        .Deadline = TAppData::TimeProvider->Now() + timeout,
+    });
+
+    YDB_LOG_DEBUG("Enqueue GetPartitionsLocation request",
+        {"logPrefix", LogPrefix()},
+        {"queueSize", PartitionsLocationQueue.size()},
+        {"timeout", timeout},
+        {"deadline", PartitionsLocationQueue.back().Deadline});
+
+    SchedulePartitionsLocationWakeup(ctx);
+}
+
+void TPersQueueReadBalancer::ProcessPartitionsLocationQueue(const TActorContext& ctx)
+{
+    const auto now = TAppData::TimeProvider->Now();
+    std::deque<TPartitionsLocationRequest> deferred;
+
+    while (!PartitionsLocationQueue.empty()) {
+        auto request = std::move(PartitionsLocationQueue.front());
+        PartitionsLocationQueue.pop_front();
+
+        // Prefer a successful answer whenever possible, even past the deadline.
+        if (TryRespondPartitionsLocation(request.Sender, request.Record, ctx)) {
+            continue;
+        }
+
+        if (request.Deadline <= now) {
+            YDB_LOG_DEBUG("GetPartitionsLocation request expired",
+                {"logPrefix", LogPrefix()},
+                {"sender", request.Sender});
+            SendPartitionsLocationError(request.Sender, ctx);
+            continue;
+        }
+
+        deferred.push_back(std::move(request));
+    }
+
+    PartitionsLocationQueue = std::move(deferred);
+    SchedulePartitionsLocationWakeup(ctx);
+}
+
+bool TPersQueueReadBalancer::TryRespondPartitionsLocation(
+    const TActorId& sender,
+    const NKikimrPQ::TGetPartitionsLocation& request,
+    const TActorContext& ctx)
+{
+    auto evResponse = std::make_unique<TEvPersQueue::TEvGetPartitionsLocationResponse>();
+
+    auto addPartitionToResponse = [&](ui64 partitionId, ui64 tabletId) {
+        if (PipesRequested.contains(tabletId)) {
+            return false;
+        }
+
+        auto iter = TabletPipes.find(tabletId);
+        if (iter == TabletPipes.end()) {
+            GetPipeClient(tabletId, ctx);
+            return false;
+        }
+
+        if (!iter->second.Ready) {
+            return false;
+        }
+
+        auto* pResponse = evResponse->Record.AddLocations();
+        pResponse->SetPartitionId(partitionId);
+        pResponse->SetNodeId(iter->second.NodeId.GetRef());
+        pResponse->SetGeneration(iter->second.Generation.GetRef());
+
+        YDB_LOG_DEBUG("The partition location was added to response",
+            {"logPrefix", LogPrefix()},
+            {"tabletId", tabletId},
+            {"partitionId", partitionId},
+            {"nodeId", pResponse->GetNodeId()},
+            {"generation", pResponse->GetGeneration()});
+
+        return true;
+    };
+
+    if (request.PartitionsSize() == 0) {
+        if (!AllPartitionPipesReady()) {
+            return false;
+        }
+
+        for (const auto& [partitionId, partitionInfo] : PartitionsInfo) {
+            if (!addPartitionToResponse(partitionId, partitionInfo.TabletId)) {
+                return false;
+            }
+        }
+    } else {
+        for (const auto& partitionInRequest : request.GetPartitions()) {
+            auto partitionInfoIter = PartitionsInfo.find(partitionInRequest);
+            if (partitionInfoIter == PartitionsInfo.end()) {
+                SendPartitionsLocationError(sender, ctx);
+                return true; // answered with error, drop from queue
+            }
+
+            if (!addPartitionToResponse(partitionInRequest, partitionInfoIter->second.TabletId)) {
+                return false;
+            }
+        }
+    }
+
+    evResponse->Record.SetStatus(true);
+    ctx.Send(sender, evResponse.release());
+    return true;
+}
+
+void TPersQueueReadBalancer::Handle(
+    TEvPersQueue::TEvGetPartitionsLocation::TPtr& ev,
+    const TActorContext& ctx)
+{
+    if (TryRespondPartitionsLocation(ev->Sender, ev->Get()->Record, ctx)) {
+        return;
+    }
+
+    EnqueuePartitionsLocationRequest(ev, ctx);
+}
+
+} // namespace NKikimr::NPQ
diff --git a/ydb/core/persqueue/pqrb/ut/partitions_location_queue_ut.cpp b/ydb/core/persqueue/pqrb/ut/partitions_location_queue_ut.cpp
new file mode 100644
index 00000000000..eee5f939096
--- /dev/null
+++ b/ydb/core/persqueue/pqrb/ut/partitions_location_queue_ut.cpp
@@ -0,0 +1,227 @@
+#include <ydb/core/persqueue/events/global.h>
+#include <ydb/core/persqueue/ut/common/pq_ut_common.h>
+
+#include <library/cpp/testing/gtest/gtest.h>
+
+namespace NKikimr::NPQ {
+namespace {
+
+THolder<TEvPersQueue::TEvGetPartitionsLocationResponse> SendLocationRequest(
+    TTestContext& tc,
+    TEvPersQueue::TEvGetPartitionsLocation* request,
+    TDuration timeout = TDuration::Seconds(10)
+) {
+    tc.Runtime->SendToPipe(tc.BalancerTabletId, tc.Edge, request, 0, GetPipeConfigWithRetries());
+    return tc.Runtime->GrabEdgeEvent<TEvPersQueue::TEvGetPartitionsLocationResponse>(timeout);
+}
+
+void WaitBalancerReady(TTestContext& tc, ui32 retries = 20) {
+    for (ui32 i = 0; i < retries; ++i) {
+        auto response = SendLocationRequest(tc, new TEvPersQueue::TEvGetPartitionsLocation());
+        ASSERT_TRUE(response);
+        if (response->Record.GetStatus()) {
+            return;
+        }
+        tc.Runtime->AdvanceCurrentTime(TDuration::MilliSeconds(100));
+        tc.Runtime->DispatchEvents();
+    }
+    FAIL() << "Could not get positive response from balancer";
+}
+
+} // namespace
+
+TEST(TPartitionsLocationQueue, AnswerAfterPipesBecomeReady) {
+    TTestContext tc;
+    tc.Prepare();
+    tc.Runtime->SetScheduledLimit(10000);
+
+    TVector<THolder<IEventHandle>> delayedConnects;
+    tc.Runtime->SetObserverFunc([&](TAutoPtr<IEventHandle>& ev) {
+        if (auto* msg = ev->CastAsLocal<TEvTabletPipe::TEvClientConnected>()) {
+            if (msg->TabletId == tc.TabletId && msg->Status == NKikimrProto::OK) {
+                delayedConnects.emplace_back(ev.Release());
+                return TTestActorRuntimeBase::EEventAction::DROP;
+            }
+        }
+        return TTestActorRuntimeBase::EEventAction::PROCESS;
+    });
+
+    PQTabletPrepare({}, {}, tc);
+    PQBalancerPrepare("topic", {{0, {tc.TabletId, 1}}}, /*ssId=*/1, tc);
+
+    ASSERT_FALSE(delayedConnects.empty());
+
+    tc.Runtime->SendToPipe(
+        tc.BalancerTabletId,
+        tc.Edge,
+        new TEvPersQueue::TEvGetPartitionsLocation(),
+        0,
+        GetPipeConfigWithRetries()
+    );
+
+    // Request must stay queued while PQ tablet pipe is not connected.
+    auto earlyResponse = tc.Runtime->GrabEdgeEvent<TEvPersQueue::TEvGetPartitionsLocationResponse>(
+        TDuration::MilliSeconds(200)
+    );
+    ASSERT_FALSE(earlyResponse) << "Location response must not arrive before pipes are ready";
+
+    tc.Runtime->SetObserverFunc(TTestActorRuntime::DefaultObserverFunc);
+    for (auto& ev : delayedConnects) {
+        tc.Runtime->Send(ev.Release());
+    }
+
+    auto response = tc.Runtime->GrabEdgeEvent<TEvPersQueue::TEvGetPartitionsLocationResponse>(
+        TDuration::Seconds(10)
+    );
+    ASSERT_TRUE(response);
+    ASSERT_TRUE(response->Record.GetStatus());
+    ASSERT_EQ(response->Record.LocationsSize(), 1u);
+    ASSERT_EQ(response->Record.GetLocations(0).GetPartitionId(), 0u);
+    ASSERT_GT(response->Record.GetLocations(0).GetNodeId(), 0u);
+}
+
+TEST(TPartitionsLocationQueue, ExpireQueuedRequest) {
+    TTestContext tc;
+    tc.Prepare();
+    tc.Runtime->SetScheduledLimit(10000);
+
+    const ui64 deadTabletId = MakeTabletID(false, 999);
+    PQBalancerPrepare("topic", {{0, {deadTabletId, 1}}}, /*ssId=*/1, tc);
+
+    tc.Runtime->SendToPipe(
+        tc.BalancerTabletId,
+        tc.Edge,
+        new TEvPersQueue::TEvGetPartitionsLocation(),
+        0,
+        GetPipeConfigWithRetries()
+    );
+
+    // Request should be queued while pipe to a missing tablet never becomes ready.
+    auto earlyResponse = tc.Runtime->GrabEdgeEvent<TEvPersQueue::TEvGetPartitionsLocationResponse>(
+        TDuration::MilliSeconds(200)
+    );
+    ASSERT_FALSE(earlyResponse) << "Location response must wait in queue before timeout";
+
+    tc.Runtime->ResetScheduledCount();
+    tc.Runtime->AdvanceCurrentTime(TDuration::Seconds(5) + TDuration::MilliSeconds(1));
+
+    auto response = tc.Runtime->GrabEdgeEvent<TEvPersQueue::TEvGetPartitionsLocationResponse>(
+        TDuration::Seconds(10)
+    );
+    ASSERT_TRUE(response);
+    ASSERT_FALSE(response->Record.GetStatus());
+}
+
+TEST(TPartitionsLocationQueue, HappyPathAfterPipesReady) {
+    TTestContext tc;
+    tc.Prepare();
+    tc.Runtime->SetScheduledLimit(10000);
+
+    PQTabletPrepare({}, {}, tc);
+    PQBalancerPrepare("topic", {{0, {tc.TabletId, 1}}}, /*ssId=*/1, tc);
+
+    WaitBalancerReady(tc);
+
+    auto response = SendLocationRequest(tc, new TEvPersQueue::TEvGetPartitionsLocation());
+    ASSERT_TRUE(response);
+    ASSERT_TRUE(response->Record.GetStatus());
+    ASSERT_EQ(response->Record.LocationsSize(), 1u);
+
+    auto* specific = new TEvPersQueue::TEvGetPartitionsLocation();
+    specific->Record.AddPartitions(0);
+    response = SendLocationRequest(tc, specific);
+    ASSERT_TRUE(response);
+    ASSERT_TRUE(response->Record.GetStatus());
+    ASSERT_EQ(response->Record.LocationsSize(), 1u);
+    ASSERT_EQ(response->Record.GetLocations(0).GetPartitionId(), 0u);
+
+    auto* unknown = new TEvPersQueue::TEvGetPartitionsLocation();
+    unknown->Record.AddPartitions(50);
+    response = SendLocationRequest(tc, unknown);
+    ASSERT_TRUE(response);
+    ASSERT_FALSE(response->Record.GetStatus());
+}
+
+TEST(TPartitionsLocationQueue, SinglePartitionNotBlockedByAllPartitions) {
+    TTestContext tc;
+    tc.Prepare();
+    tc.Runtime->SetScheduledLimit(10000);
+
+    const ui64 deadTabletId = MakeTabletID(false, 999);
+
+    TVector<THolder<IEventHandle>> delayedConnects;
+    tc.Runtime->SetObserverFunc([&](TAutoPtr<IEventHandle>& ev) {
+        if (auto* msg = ev->CastAsLocal<TEvTabletPipe::TEvClientConnected>()) {
+            if (msg->TabletId == tc.TabletId && msg->Status == NKikimrProto::OK) {
+                delayedConnects.emplace_back(ev.Release());
+                return TTestActorRuntimeBase::EEventAction::DROP;
+            }
+        }
+        return TTestActorRuntimeBase::EEventAction::PROCESS;
+    });
+
+    PQTabletPrepare({}, {}, tc);
+    PQBalancerPrepare(
+        "topic",
+        {{0, {tc.TabletId, 1}}, {1, {deadTabletId, 2}}},
+        /*ssId=*/1,
+        tc
+    );
+
+    ASSERT_FALSE(delayedConnects.empty());
+
+    // Head of queue: all partitions (blocked on dead tablet).
+    tc.Runtime->SendToPipe(
+        tc.BalancerTabletId,
+        tc.Edge,
+        new TEvPersQueue::TEvGetPartitionsLocation(),
+        0,
+        GetPipeConfigWithRetries()
+    );
+
+    // Behind it: only partition 0 (same tablet as delayed pipe).
+    auto* specific = new TEvPersQueue::TEvGetPartitionsLocation();
+    specific->Record.AddPartitions(0);
+    tc.Runtime->SendToPipe(
+        tc.BalancerTabletId,
+        tc.Edge,
+        specific,
+        0,
+        GetPipeConfigWithRetries()
+    );
+
+    auto earlyResponse = tc.Runtime->GrabEdgeEvent<TEvPersQueue::TEvGetPartitionsLocationResponse>(
+        TDuration::MilliSeconds(200)
+    );
+    ASSERT_FALSE(earlyResponse) << "Neither request can be answered before partition-0 pipe is ready";
+
+    tc.Runtime->SetObserverFunc(TTestActorRuntime::DefaultObserverFunc);
+    for (auto& ev : delayedConnects) {
+        tc.Runtime->Send(ev.Release());
+    }
+
+    // Specific request must be answered without waiting for the all-partitions request.
+    auto response = tc.Runtime->GrabEdgeEvent<TEvPersQueue::TEvGetPartitionsLocationResponse>(
+        TDuration::Seconds(10)
+    );
+    ASSERT_TRUE(response);
+    ASSERT_TRUE(response->Record.GetStatus());
+    ASSERT_EQ(response->Record.LocationsSize(), 1u);
+    ASSERT_EQ(response->Record.GetLocations(0).GetPartitionId(), 0u);
+
+    auto stillWaiting = tc.Runtime->GrabEdgeEvent<TEvPersQueue::TEvGetPartitionsLocationResponse>(
+        TDuration::MilliSeconds(200)
+    );
+    ASSERT_FALSE(stillWaiting) << "All-partitions request must stay queued while dead tablet is down";
+
+    tc.Runtime->ResetScheduledCount();
+    tc.Runtime->AdvanceCurrentTime(TDuration::Seconds(5) + TDuration::MilliSeconds(1));
+
+    auto expired = tc.Runtime->GrabEdgeEvent<TEvPersQueue::TEvGetPartitionsLocationResponse>(
+        TDuration::Seconds(10)
+    );
+    ASSERT_TRUE(expired);
+    ASSERT_FALSE(expired->Record.GetStatus());
+}
+
+} // namespace NKikimr::NPQ
diff --git a/ydb/core/persqueue/pqrb/ut/ya.make b/ydb/core/persqueue/pqrb/ut/ya.make
new file mode 100644
index 00000000000..d0eb416aacb
--- /dev/null
+++ b/ydb/core/persqueue/pqrb/ut/ya.make
@@ -0,0 +1,16 @@
+GTEST()
+
+SIZE(MEDIUM)
+
+SRCS(
+    partitions_location_queue_ut.cpp
+)
+
+PEERDIR(
+    ydb/core/persqueue/ut/common
+    ydb/core/testlib/default
+)
+
+YQL_LAST_ABI_VERSION()
+
+END()
diff --git a/ydb/core/persqueue/pqrb/ya.make b/ydb/core/persqueue/pqrb/ya.make
index 715de95a66d..596525cb66b 100644
--- a/ydb/core/persqueue/pqrb/ya.make
+++ b/ydb/core/persqueue/pqrb/ya.make
@@ -12,6 +12,7 @@ SRCS(
     read_balancer__mlp_balancing.cpp
     read_balancer_app.cpp
     read_balancer.cpp
+    read_balancer_partition_location.cpp
 )
 
 GENERATE_ENUM_SERIALIZATION(read_balancer__balancing.h)
@@ -27,4 +28,5 @@ PEERDIR(
 END()
 
 RECURSE_FOR_TESTS(
+    ut
 )
diff --git a/ydb/core/protos/pqevents_global.proto b/ydb/core/protos/pqevents_global.proto
index 91aef554cc7..fb8775aad02 100644
--- a/ydb/core/protos/pqevents_global.proto
+++ b/ydb/core/protos/pqevents_global.proto
@@ -88,6 +88,8 @@ message TReadSessionsInfoResponse {
 
 message TGetPartitionsLocation {
     repeated uint64 Partitions = 1;
+    // How long PQRB may keep the request queued waiting for pipes.
+    optional uint64 TimeoutMs = 2 [default = 5000];
 }
 
 message TPartitionsLocationResponse {
diff --git a/ydb/services/persqueue_v1/actors/schema_actors.cpp b/ydb/services/persqueue_v1/actors/schema_actors.cpp
index d79e74d8256..ce12a454d27 100644
--- a/ydb/services/persqueue_v1/actors/schema_actors.cpp
+++ b/ydb/services/persqueue_v1/actors/schema_actors.cpp
@@ -2,6 +2,7 @@
 
 #include "persqueue_utils.h"
 
+#include <ydb/core/actorlib_impl/long_timer.h>
 #include <ydb/core/persqueue/public/utils.h>
 #include <ydb/core/ydb_convert/topic_description.h>
 #include <ydb/core/ydb_convert/ydb_convert.h>
@@ -277,17 +278,37 @@ bool TDescribeTopicActorImpl::StateWork(TAutoPtr<IEventHandle>& ev, const TActor
         HFuncCtx(NKikimr::TEvPersQueue::TEvReadSessionsInfoResponse, Handle, ctx);
         HFuncCtx(TEvPersQueue::TEvGetPartitionsLocationResponse, Handle, ctx);
         HFuncCtx(TEvPQProxy::TEvRequestTablet, Handle, ctx);
+        HFuncCtx(TEvents::TEvWakeup, Handle, ctx);
         default: return false;
     }
     return true;
 }
 
 void TDescribeTopicActorImpl::PassAway(const TActorContext& ctx) {
+    CancelRequestTimeout(ctx);
     for (auto& [_, tablet] : Tablets) {
         NTabletPipe::CloseClient(ctx, tablet.Pipe);
     }
 }
 
+void TDescribeTopicActorImpl::CancelRequestTimeout(const TActorContext& ctx) {
+    if (TimeoutTimerActorId) {
+        ctx.Send(TimeoutTimerActorId, new TEvents::TEvPoison());
+        TimeoutTimerActorId = {};
+    }
+}
+
+TDuration TDescribeTopicActorImpl::RemainingRequestTimeout() const {
+    if (!RequestStartTime) {
+        return RequestTimeout;
+    }
+    const auto now = TAppData::TimeProvider->Now();
+    if (now >= *RequestStartTime + RequestTimeout) {
+        return TDuration::Zero();
+    }
+    return *RequestStartTime + RequestTimeout - now;
+}
+
 void TDescribeTopicActor::StateWork(TAutoPtr<IEventHandle>& ev) {
     if (!TDescribeTopicActorImpl::StateWork(ev, this->ActorContext())) {
         TBase::StateWork(ev);
@@ -310,6 +331,9 @@ void TDescribeTopicActorImpl::Handle(TEvTabletPipe::TEvClientDestroyed::TPtr& ev
 }
 
 void TDescribeTopicActor::RaiseError(const TString& error, const Ydb::PersQueue::ErrorCode::ErrorCode errorCode, const Ydb::StatusIds::StatusCode status, const TActorContext& ctx) {
+    if (TBase::IsDead) {
+        return;
+    }
     this->Request_->RaiseIssue(FillIssue(error, errorCode));
     TBase::Reply(status, ctx);
 }
@@ -363,6 +387,18 @@ void TDescribeTopicActorImpl::Handle(TEvPQProxy::TEvRequestTablet::TPtr& ev, con
     RequestTablet(tabletInfo, ctx);
 }
 
+void TDescribeTopicActorImpl::Handle(TEvents::TEvWakeup::TPtr&, const TActorContext& ctx) {
+    TimeoutTimerActorId = {};
+    YDB_LOG_DEBUG_CTX(ctx, "DescribeTopicImpl Request timed out",
+        {"selfId", ctx.SelfID});
+    RaiseError(
+        "Describe topic request timed out",
+        Ydb::PersQueue::ErrorCode::ERROR,
+        Ydb::StatusIds::TIMEOUT,
+        ctx
+    );
+}
+
 void TDescribeTopicActorImpl::RequestTablet(ui64 tabletId, const TActorContext& ctx) {
     auto it = Tablets.find(tabletId);
     if (it != Tablets.end()) {
@@ -443,7 +479,7 @@ void TDescribeTopicActorImpl::RequestPartitionsLocation(const TActorContext& ctx
     }
     NTabletPipe::SendData(
         ctx, Tablets[BalancerTabletId].Pipe,
-        new TEvPersQueue::TEvGetPartitionsLocation(partsVector)
+        new TEvPersQueue::TEvGetPartitionsLocation(partsVector, RemainingRequestTimeout())
     );
     ++RequestsInfly;
 }
@@ -534,6 +570,7 @@ void TDescribeTopicActorImpl::Handle(TEvPersQueue::TEvGetPartitionsLocationRespo
         auto res = ApplyResponse(ev, ctx);
         if (res) {
             GotLocation = true;
+            LocationsBackoff.Reset();
             AFL_ENSURE(RequestsInfly > 0);
             --RequestsInfly;
 
@@ -546,11 +583,25 @@ void TDescribeTopicActorImpl::Handle(TEvPersQueue::TEvGetPartitionsLocationRespo
         }
     }
 
+    if (!LocationsBackoff.HasMore()) {
+        YDB_LOG_DEBUG_CTX(ctx, "DescribeTopicImpl PartitionsLocation retries exceeded",
+            {"selfId", ctx.SelfID},
+            {"response", record.DebugString()});
+        return RaiseError(
+            "Partition locations are not available",
+            Ydb::PersQueue::ErrorCode::TABLET_PIPE_DISCONNECTED,
+            Ydb::StatusIds::UNAVAILABLE,
+            ctx
+        );
+    }
+
+    const auto delay = LocationsBackoff.Next();
     YDB_LOG_DEBUG_CTX(ctx, "DescribeTopicImpl Something wrong on location, retry",
         {"selfId", ctx.SelfID},
+        {"iteration", LocationsBackoff.GetIteration()},
+        {"delay", delay.ToString()},
         {"response", record.DebugString()});
-    //Something gone wrong, retry
-    ctx.Schedule(TDuration::MilliSeconds(200), new TEvPQProxy::TEvRequestTablet(BalancerTabletId));
+    ctx.Schedule(delay, new TEvPQProxy::TEvRequestTablet(BalancerTabletId));
 }
 
 void TDescribeTopicActorImpl::CheckCloseBalancerPipe(const TActorContext& ctx) {
@@ -781,6 +832,8 @@ bool TDescribeTopicActorImpl::ProcessTablets(
         Tablets[pi.GetTabletId()].TabletId = pi.GetTabletId();
     }
 
+    RequestStartTime = TAppData::TimeProvider->Now();
+
     for (auto& pair : Tablets) {
         RequestTablet(pair.second, ctx);
     }
@@ -790,6 +843,8 @@ bool TDescribeTopicActorImpl::ProcessTablets(
         return false;
     }
 
+    TimeoutTimerActorId = CreateLongTimer(ctx, RequestTimeout,
+        new IEventHandle(ctx.SelfID, ctx.SelfID, new TEvents::TEvWakeup()));
     return true;
 }
 
@@ -884,6 +939,9 @@ void TPartitionsLocationActor::Finalize() {
 }
 
 void TPartitionsLocationActor::RaiseError(const TString& error, const Ydb::PersQueue::ErrorCode::ErrorCode errorCode, const Ydb::StatusIds::StatusCode status, const TActorContext&) {
+    if (TBase::IsDead) {
+        return;
+    }
     this->AddIssue(FillIssue(error, errorCode));
     this->RespondWithCode(status);
 }
diff --git a/ydb/services/persqueue_v1/actors/schema_actors.h b/ydb/services/persqueue_v1/actors/schema_actors.h
index d455b45e353..d8ac7a15f1c 100644
--- a/ydb/services/persqueue_v1/actors/schema_actors.h
+++ b/ydb/services/persqueue_v1/actors/schema_actors.h
@@ -2,9 +2,12 @@
 
 #include "events.h"
 #include <ydb/core/persqueue/events/global.h>
+#include <ydb/core/util/backoff.h>
 #include <ydb/services/lib/actors/pq_schema_actor.h>
 #include <ydb/core/client/server/ic_nodes_cache_service.h>
 
+#include <optional>
+
 namespace NKikimr::NGRpcProxy::V1 {
 
 using namespace NKikimr::NGRpcService;
@@ -74,6 +77,8 @@ struct TDescribeTopicActorSettings {
 class TDescribeTopicActorImpl
 {
 protected:
+    static constexpr TDuration RequestTimeout = TDuration::Seconds(30);
+
     struct TTabletInfo {
         ui64 TabletId = 0;
         std::vector<ui32> Partitions;
@@ -100,6 +105,7 @@ public:
     void Handle(TEvPersQueue::TEvGetPartitionsLocationResponse::TPtr& ev, const TActorContext& ctx);
 
     void Handle(TEvPQProxy::TEvRequestTablet::TPtr& ev, const TActorContext& ctx);
+    void Handle(TEvents::TEvWakeup::TPtr& ev, const TActorContext& ctx);
 
     bool ProcessTablets(const NKikimrSchemeOp::TPersQueueGroupDescription& description, const TActorContext& ctx);
 
@@ -129,11 +135,18 @@ public:
     void PassAway(const TActorContext& ctx);
 
 private:
+    void CancelRequestTimeout(const TActorContext& ctx);
+
     std::map<ui64, TTabletInfo> Tablets;
     ui32 RequestsInfly = 0;
 
     bool GotLocation = false;
     bool GotReadSessions = false;
+    TBackoff LocationsBackoff = TBackoff(25, TDuration::MilliSeconds(10), TDuration::MilliSeconds(100));
+    TActorId TimeoutTimerActorId;
+    std::optional<TInstant> RequestStartTime;
+
+    TDuration RemainingRequestTimeout() const;
 
 protected:
     ui64 BalancerTabletId = 0;