Remove deprecated field of TReleasePartition (#7689)

Author: nshestakov

ydb/core/kafka_proxy/actors/kafka_read_session_actor.cpp

@@ -580,81 +580,47 @@ void TKafkaReadSessionActor::HandleReleasePartition(TEvPersQueue::TEvReleasePart
     auto newPartitionsToLockCount = newPartitionsToLockIt == NewPartitionsToLockOnTime.end() ? 0 : newPartitionsToLockIt->second.size();
 
     auto topicPartitionsIt = TopicPartitions.find(pathIt->second->GetInternalName());
-    Y_ABORT_UNLESS(record.GetCount() <= (topicPartitionsIt.IsEnd() ? 0 : topicPartitionsIt->second.ToLock.size() + topicPartitionsIt->second.ReadingNow.size()) + newPartitionsToLockCount);
-
-    if (!group) {
-        for (ui32 c = 0; c < record.GetCount(); ++c) {
-            // if some partition not locked yet, then release it without rebalance
-            if (newPartitionsToLockCount > 0) {
-                newPartitionsToLockCount--;
-                InformBalancerAboutPartitionRelease(topicInfoIt->first, newPartitionsToLockIt->second.back().PartitionId, ctx);
-                newPartitionsToLockIt->second.pop_back();
-                continue;
-            }
-
-            if (!topicPartitionsIt->second.ToLock.empty()) {
-                auto partitionToReleaseIt = topicPartitionsIt->second.ToLock.begin();
-                topicPartitionsIt->second.ToLock.erase(partitionToReleaseIt);
-                InformBalancerAboutPartitionRelease(topicInfoIt->first, *partitionToReleaseIt, ctx);
-                continue;
-            }
-
-            NeedRebalance = true;
-            ui32 partitionToRelease = 0;
-            ui32 i = 0;
-
-            for (auto curPartition : topicPartitionsIt->second.ReadingNow) {
-                if (!topicPartitionsIt->second.ToRelease.contains(curPartition)) {
-                    ++i;
-                    if (rand() % i == 0) {
-                        partitionToRelease = curPartition;
-                    }
-                }
-            }
+    Y_ABORT_UNLESS(1 <= (topicPartitionsIt.IsEnd() ? 0 : topicPartitionsIt->second.ToLock.size() + topicPartitionsIt->second.ReadingNow.size()) + newPartitionsToLockCount);
 
-            topicPartitionsIt->second.ToRelease.emplace(partitionToRelease);
-        }
-    } else {
-        auto partitionToRelease = record.GetGroup() - 1;
+    auto partitionToRelease = record.GetGroup() - 1;
 
-        if (newPartitionsToLockIt != NewPartitionsToLockOnTime.end()) {
-            auto& newPartitions = newPartitionsToLockIt->second;
-            for (auto& newPartition : newPartitions) {
-                if (newPartition.PartitionId == partitionToRelease) {
-                    InformBalancerAboutPartitionRelease(topicInfoIt->first, partitionToRelease, ctx);
+    if (newPartitionsToLockIt != NewPartitionsToLockOnTime.end()) {
+        auto& newPartitions = newPartitionsToLockIt->second;
+        for (auto& newPartition : newPartitions) {
+            if (newPartition.PartitionId == partitionToRelease) {
+                InformBalancerAboutPartitionRelease(topicInfoIt->first, partitionToRelease, ctx);
 
-                    auto tmp = std::move(newPartitions);
-                    newPartitions.reserve(tmp.size() - 1);
+                auto tmp = std::move(newPartitions);
+                newPartitions.reserve(tmp.size() - 1);
 
-                    for (auto& t : tmp) {
-                        if (t.PartitionId != partitionToRelease) {
-                            newPartitions.push_back(t);
-                        }
+                for (auto& t : tmp) {
+                    if (t.PartitionId != partitionToRelease) {
+                        newPartitions.push_back(t);
                     }
-
-                    return;
                 }
-            }
-        }
 
-        if (topicPartitionsIt != TopicPartitions.end()) {
-            if (topicPartitionsIt->second.ToLock.contains(partitionToRelease)) {
-                InformBalancerAboutPartitionRelease(topicInfoIt->first, partitionToRelease, ctx);
-                topicPartitionsIt->second.ToLock.erase(partitionToRelease);
                 return;
             }
+        }
+    }
 
-            if (topicPartitionsIt->second.ReadingNow.contains(partitionToRelease) && !topicPartitionsIt->second.ToRelease.contains(partitionToRelease)) {
-                InformBalancerAboutPartitionRelease(topicInfoIt->first, partitionToRelease, ctx);
-                NeedRebalance = true;
-                topicPartitionsIt->second.ReadingNow.erase(partitionToRelease);
-                return;
-            }
+    if (topicPartitionsIt != TopicPartitions.end()) {
+        if (topicPartitionsIt->second.ToLock.contains(partitionToRelease)) {
+            InformBalancerAboutPartitionRelease(topicInfoIt->first, partitionToRelease, ctx);
+            topicPartitionsIt->second.ToLock.erase(partitionToRelease);
+            return;
         }
 
-        KAFKA_LOG_I("ignored ev release topic# " << record.GetTopic()
-                 << ", reason# partition " << partitionToRelease << " isn`t locked");
+        if (topicPartitionsIt->second.ReadingNow.contains(partitionToRelease) && !topicPartitionsIt->second.ToRelease.contains(partitionToRelease)) {
+            InformBalancerAboutPartitionRelease(topicInfoIt->first, partitionToRelease, ctx);
+            NeedRebalance = true;
+            topicPartitionsIt->second.ReadingNow.erase(partitionToRelease);
+            return;
+        }
     }
+
+    KAFKA_LOG_I("ignored ev release topic# " << record.GetTopic()
+             << ", reason# partition " << partitionToRelease << " isn`t locked");
 }
 
 void TKafkaReadSessionActor::InformBalancerAboutPartitionRelease(const TString& topic, ui64 partition, const TActorContext& ctx) {

ydb/core/persqueue/read_balancer__balancing.cpp

@@ -536,7 +536,6 @@ 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());
 

ydb/core/protos/pqconfig.proto

@@ -649,7 +649,7 @@ message TReleasePartition {
     optional uint64 Generation = 2;
     optional string Session = 3;
     optional string ClientId = 4;
-    optional uint32 Count = 5 [deprecated = true]; // Remove at 2025-1
+    reserved 5; // optional uint32 Count = 5;
     optional NActorsProto.TActorId PipeClient = 6;
     optional uint32 Group = 7;
     optional string Path = 8;

ydb/services/deprecated/persqueue_v0/grpc_pq_read_actor.cpp

@@ -1164,37 +1164,35 @@ void TReadSessionActor::Handle(TEvPersQueue::TEvReleasePartition::TPtr& ev, cons
         return;
     }
 
-    for (ui32 c = 0; c < record.GetCount(); ++c) {
-        Y_ABORT_UNLESS(!Partitions.empty());
-
-        TActorId actorId = TActorId{};
-        auto jt = Partitions.begin();
-        ui32 i = 0;
-        for (auto it = Partitions.begin(); it != Partitions.end(); ++it) {
-            if (it->first.first == clientName && !it->second.Releasing && (group == 0 || it->first.second + 1 == group)) {
-                ++i;
-                if (rand() % i == 0) { //will lead to 1/n probability for each of n partitions
-                    actorId = it->second.Actor;
-                    jt = it;
-                }
+    Y_ABORT_UNLESS(!Partitions.empty());
+
+    TActorId actorId = TActorId{};
+    auto jt = Partitions.begin();
+    ui32 i = 0;
+    for (auto it = Partitions.begin(); it != Partitions.end(); ++it) {
+        if (it->first.first == clientName && !it->second.Releasing && (group == 0 || it->first.second + 1 == group)) {
+            ++i;
+            if (rand() % i == 0) { //will lead to 1/n probability for each of n partitions
+                actorId = it->second.Actor;
+                jt = it;
             }
         }
-        Y_ABORT_UNLESS(actorId);
+    }
+    Y_ABORT_UNLESS(actorId);
 
-        {
-            auto it = TopicCounters.find(name);
-            Y_ABORT_UNLESS(it != TopicCounters.end());
-            it->second.PartitionsToBeReleased.Inc();
-        }
+    {
+        auto it = TopicCounters.find(name);
+        Y_ABORT_UNLESS(it != TopicCounters.end());
+        it->second.PartitionsToBeReleased.Inc();
+    }
 
-        LOG_INFO_S(ctx, NKikimrServices::PQ_READ_PROXY, PQ_LOG_PREFIX << " releasing " << jt->first.first << ":" << jt->first.second);
-        jt->second.Releasing = true;
+    LOG_INFO_S(ctx, NKikimrServices::PQ_READ_PROXY, PQ_LOG_PREFIX << " releasing " << jt->first.first << ":" << jt->first.second);
+    jt->second.Releasing = true;
 
-        ctx.Send(actorId, new TEvPQProxy::TEvReleasePartition());
-        if (ClientsideLocksAllowed && jt->second.LockSent && !jt->second.Reading) { //locked and no active reads
-            if (!ProcessReleasePartition(jt, BalanceRightNow, false, ctx)) { // returns false if actor died
-                return;
-            }
+    ctx.Send(actorId, new TEvPQProxy::TEvReleasePartition());
+    if (ClientsideLocksAllowed && jt->second.LockSent && !jt->second.Reading) { //locked and no active reads
+        if (!ProcessReleasePartition(jt, BalanceRightNow, false, ctx)) { // returns false if actor died
+            return;
         }
     }
     AnswerForCommitsIfCan(ctx); // in case of killing partition

ydb/services/persqueue_v1/actors/read_session_actor.cpp

@@ -1410,8 +1410,6 @@ void TReadSessionActor<UseMigrationProtocol>::Handle(TEvPersQueue::TEvReleasePar
     Y_ABORT_UNLESS(record.GetSession() == Session);
     Y_ABORT_UNLESS(record.GetClientId() == ClientId);
 
-    const ui32 group = record.HasGroup() ? record.GetGroup() : 0;
-
     auto pathIter = FullPathToConverter.find(NPersQueue::NormalizeFullPath(record.GetPath()));
     Y_ABORT_UNLESS(pathIter != FullPathToConverter.end());
 
@@ -1445,61 +1443,29 @@ void TReadSessionActor<UseMigrationProtocol>::Handle(TEvPersQueue::TEvReleasePar
         }
     };
 
-    if (!group) {
-        // Release partitions by count
-        for (ui32 c = 0; c < record.GetCount(); ++c) {
-            if (Partitions.empty()) {
-                return CloseSession(PersQueue::ErrorCode::ErrorCode::ERROR,
-                                    TStringBuilder() << "internal error: can`t release partition #01",
-                                    ctx);
-            }
-
-            auto jt = Partitions.end();
-            ui32 i = 0;
+    ui32 partitionId = record.GetGroup() - 1;
+    bool found = false;
 
-            for (auto it = Partitions.begin(); it != Partitions.end(); ++it) {
-                auto& partitionInfo = it->second;
-                if (!partitionInfo.Releasing && partitionInfo.Topic->GetInternalName() == converter->GetInternalName()) {
-                    ++i;
-                    if (rand() % i == 0) { // will lead to 1/n probability for each of n partitions
-                        jt = it;
-                    }
-                }
-            }
+    // Release partitions by partition id
+    LOG_DEBUG_S(ctx, NKikimrServices::PQ_READ_PROXY, PQ_LOG_PREFIX << " gone release"
+        << ": partition# " << partitionId);
 
-            if (jt == Partitions.end()) {
-                return CloseSession(PersQueue::ErrorCode::ErrorCode::ERROR,
-                                    TStringBuilder() << "internal error: can`t release partition #02",
-                                    ctx);
+    for (auto it = Partitions.begin(); it != Partitions.end(); ++it) {
+        auto& partitionInfo = it->second;
+        if (partitionInfo.Topic->GetInternalName() == converter->GetInternalName() && partitionId == partitionInfo.Partition.Partition) {
+            if (!partitionInfo.Releasing) {
+                doRelease(it);
             }
 
-            doRelease(jt);
-        }
-    } else {
-        ui32 partitionId = group - 1;
-        bool found = false;
-
-        // Release partitions by partition id
-        LOG_DEBUG_S(ctx, NKikimrServices::PQ_READ_PROXY, PQ_LOG_PREFIX << " gone release"
-            << ": partition# " << partitionId);
-
-        for (auto it = Partitions.begin(); it != Partitions.end(); ++it) {
-            auto& partitionInfo = it->second;
-            if (partitionInfo.Topic->GetInternalName() == converter->GetInternalName() && partitionId == partitionInfo.Partition.Partition) {
-                if (!partitionInfo.Releasing) {
-                    doRelease(it);
-                }
-
-                found = true;
-                break;
-            }
+            found = true;
+            break;
         }
+    }
 
-        if (!found) {
-            return CloseSession(PersQueue::ErrorCode::ErrorCode::ERROR,
-                                TStringBuilder() << "internal error: releasing unknown partition " << partitionId,
-                                ctx);
-        }
+    if (!found) {
+        return CloseSession(PersQueue::ErrorCode::ErrorCode::ERROR,
+                            TStringBuilder() << "internal error: releasing unknown partition " << partitionId,
+                            ctx);
     }
 }