Merge 9470dac into 9d83402

Author: nshestakov

ydb/core/persqueue/read_balancer__balancing.cpp

@@ -165,7 +165,7 @@ void TPartitionFamily::Release(const TActorContext& ctx, ETargetStatus targetSta
 
     LOG_INFO_S(ctx, NKikimrServices::PERSQUEUE_READ_BALANCER,
             GetPrefix() << " release partitions [" << JoinRange(", ", LockedPartitions.begin(), LockedPartitions.end())
-            << "]. Target status " << targetStatus);
+            << "] pipe " << Session->Pipe << ". Target status " << targetStatus);
 
     Status = EStatus::Releasing;
     TargetStatus = targetStatus;
@@ -533,6 +533,7 @@ std::unique_ptr<TEvPersQueue::TEvReleasePartition> TPartitionFamily::MakeEvRelea
     r.SetPath(TopicPath());
     r.SetGeneration(TabletGeneration());
     r.SetClientId(Session->ClientId);
+    r.SetCount(1);
     r.SetGroup(partitionId + 1);
     ActorIdToProto(Session->Pipe, r.MutablePipeClient());
 
@@ -1661,20 +1662,20 @@ void TBalancer::Handle(TEvTabletPipe::TEvServerDisconnected::TPtr& ev, const TAc
 
     LOG_INFO_S(ctx, NKikimrServices::PERSQUEUE_READ_BALANCER,
             GetPrefix() << "pipe " << ev->Get()->ClientId << " disconnected; active server actors: "
-            << (it != Sessions.end() ? it->second->ServerActors : -1));
+            << it->second->ServerActors);
 
     auto& session = it->second;
     if (--(session->ServerActors) > 0) {
         return;
     }
 
     if (!session->SessionName.empty()) {
-        LOG_NOTICE_S(ctx, NKikimrServices::PERSQUEUE_READ_BALANCER,
-                GetPrefix() << "pipe " << ev->Get()->ClientId << " client "
-                << session->ClientId << " disconnected session " << session->SessionName);
-
         auto* consumer = GetConsumer(session->ClientId);
         if (consumer) {
+            LOG_NOTICE_S(ctx, NKikimrServices::PERSQUEUE_READ_BALANCER,
+                    GetPrefix() << "pipe " << ev->Get()->ClientId << " client "
+                    << session->ClientId << " disconnected session " << session->SessionName);
+
             consumer->UnregisterReadingSession(session.get(), ctx);
 
             if (consumer->Sessions.empty()) {

ydb/services/persqueue_v1/actors/read_session_actor.cpp

@@ -1419,6 +1419,9 @@ void TReadSessionActor<UseMigrationProtocol>::Handle(TEvPersQueue::TEvReleasePar
     Y_ABORT_UNLESS(it != Topics.end());
 
     if (it->second.PipeClient != ActorIdFromProto(record.GetPipeClient())) {
+        LOG_INFO_S(ctx, NKikimrServices::PQ_READ_PROXY, PQ_LOG_PREFIX << " try releasing but pipe is bad"
+            << ": group# " << group << " count# " << record.GetCount() << " expected: " << it->second.PipeClient
+            << " actual: " << ActorIdFromProto(record.GetPipeClient()));
         return;
     }