fixed memory travel when PartitionGraph was changed (#20558)

Author: nshestakov

ydb/core/kafka_proxy/actors/kafka_read_session_actor.cpp

@@ -476,7 +476,7 @@ void TKafkaReadSessionActor::HandleAuthOk(NGRpcProxy::V1::TEvPQProxy::TEvAuthRes
 
     for (const auto& [name, t] : ev->Get()->TopicAndTablets) {
         auto internalName = t.TopicNameConverter->GetInternalName();
-        TopicsInfo[internalName] = NGRpcProxy::TTopicHolder::FromTopicInfo(t);
+        TopicsInfo[internalName] = NKikimr::NGRpcProxy::TTopicHolder(t);
         FullPathToConverter[t.TopicNameConverter->GetPrimaryPath()] = t.TopicNameConverter;
         FullPathToConverter[t.TopicNameConverter->GetSecondaryPath()] = t.TopicNameConverter;
     }

ydb/core/persqueue/partition.cpp

@@ -3263,9 +3263,7 @@ void TPartition::EmulatePostProcessUserAct(const TEvPQ::TEvSetClientInfo& act,
         );
 
         userInfo.Offset = offset;
-        if (userInfo.Offset <= (i64)StartOffset) {
-            userInfo.AnyCommits = false;
-        }
+        userInfo.AnyCommits = userInfo.Offset > (i64)StartOffset;
 
         if (LastOffsetHasBeenCommited(userInfo)) {
             SendReadingFinished(user);

ydb/core/persqueue/partition_init.cpp

@@ -433,12 +433,12 @@ void TInitInfoRangeStep::Handle(TEvKeyValue::TEvResponse::TPtr &ev, const TActor
                 Y_ABORT_UNLESS(key);
                 RequestInfoRange(ctx, Partition()->Tablet, PartitionId(), *key);
             } else {
-                Done(ctx);
+                PostProcessing(ctx);
             }
             break;
         }
         case NKikimrProto::NODATA:
-            Done(ctx);
+            PostProcessing(ctx);
             break;
         case NKikimrProto::ERROR:
             PQ_LOG_ERROR("read topic error");
@@ -450,6 +450,16 @@ void TInitInfoRangeStep::Handle(TEvKeyValue::TEvResponse::TPtr &ev, const TActor
     };
 }
 
+void TInitInfoRangeStep::PostProcessing(const TActorContext& ctx) {
+    auto& usersInfoStorage = Partition()->UsersInfoStorage;
+    for (auto& [_, userInfo] : usersInfoStorage->GetAll()) {
+        userInfo.AnyCommits = userInfo.Offset > (i64)Partition()->StartOffset;
+    }
+
+    Done(ctx);
+}
+
+
 
 //
 // TInitDataRangeStep

ydb/core/persqueue/partition_init.h

@@ -140,6 +140,8 @@ class TInitInfoRangeStep: public TBaseKVStep {
 
     void Execute(const TActorContext& ctx) override;
     void Handle(TEvKeyValue::TEvResponse::TPtr& ev, const TActorContext& ctx) override;
+
+    void PostProcessing(const TActorContext& ctx);
 };
 
 class TInitDataRangeStep: public TBaseKVStep {

ydb/core/persqueue/ut/common/autoscaling_ut_common.cpp

@@ -113,6 +113,7 @@ TTopicSdkTestSetup CreateSetup() {
     setup.GetRuntime().GetAppData().PQConfig.SetTopicsAreFirstClassCitizen(true);
     setup.GetRuntime().GetAppData().PQConfig.SetUseSrcIdMetaMappingInFirstClass(true);
     setup.GetRuntime().GetAppData().PQConfig.SetBalancerWakeupIntervalSec(1);
+    setup.GetRuntime().GetAppData().PQConfig.SetACLRetryTimeoutSec(1);
 
     return setup;
 }

ydb/core/persqueue/ut/ut_with_sdk/commitoffset_ut.cpp

@@ -618,6 +618,49 @@ Y_UNIT_TEST_SUITE(CommitOffset) {
             return true;
         });
     }
+
+    Y_UNIT_TEST(DistributedTxCommit_LongReadSession) {
+        TTopicSdkTestSetup setup = CreateSetup();
+        PrepareAutopartitionedTopic(setup);
+
+        std::vector<NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent> messages;
+
+        auto result = setup.Read(TEST_TOPIC, TEST_CONSUMER, [&](auto& x) {
+            messages.push_back(x);
+            return true;
+        });
+
+        {
+            Cerr << ">>>>> Alter topic" << Endl << Flush;
+            TTopicClient client(setup.MakeDriver());
+            TAlterTopicSettings settings;
+            settings.SetRetentionPeriod(TDuration::Hours(1));
+            client.AlterTopic(TEST_TOPIC, settings).GetValueSync();
+        }
+
+        // Wait recheck acl happened
+        messages[0].GetMessages()[0].Commit(); // trigger recheck
+        Sleep(TDuration::Seconds(5));
+
+        Cerr << ">>>>> Commit message" << Endl << Flush;
+
+        bool first = true;
+        for (auto& e : messages) {
+            for (auto& m :e.GetMessages()) {
+                if (first) {
+                    first = false;
+                    continue;
+                }
+                m.Commit();
+            }
+        }
+
+        // Wait commit happened
+        Sleep(TDuration::Seconds(5));
+
+        result.Reader->Close();
+    }
+
 }
 
 } // namespace NKikimr

ydb/services/deprecated/persqueue_v0/grpc_pq_actor.h

@@ -804,7 +804,7 @@ class TReadSessionActor : public TActorBootstrapped<TReadSessionActor> {
     THashMap<std::pair<TString, ui32>, TPartitionActorInfo> Partitions; //topic[ClientSideName!]:partition -> info
 
     THashMap<TString, NPersQueue::TTopicConverterPtr> FullPathToConverter; // PrimaryFullPath -> Converter, for balancer replies matching
-    THashMap<TString, TTopicHolder> Topics; // PrimaryName ->topic info
+    THashMap<TString, TTopicHolder::TPtr> Topics; // PrimaryName ->topic info
 
     TVector<ui32> Groups;
     bool ReadOnlyLocal;

ydb/services/deprecated/persqueue_v0/grpc_pq_read_actor.cpp

@@ -146,7 +146,7 @@ class TPartitionActor : public NActors::TActorBootstrapped<TPartitionActor> {
 public:
      TPartitionActor(const TActorId& parentId, const TString& clientId, const ui64 cookie, const TString& session, const ui32 generation,
                         const ui32 step, const NPersQueue::TTopicConverterPtr& topic, const TString& database, const ui32 partition, const ui64 tabletID,
-                        const TReadSessionActor::TTopicCounters& counters, const TString& clientDC, std::set<NPQ::TPartitionGraph::Node*> parents);
+                        const TReadSessionActor::TTopicCounters& counters, const TString& clientDC, const TTopicHolder::TPtr& topicHolder);
     ~TPartitionActor();
 
     void Bootstrap(const NActors::TActorContext& ctx);
@@ -212,6 +212,7 @@ class TPartitionActor : public NActors::TActorBootstrapped<TPartitionActor> {
     void CommitDone(ui64 cookie, const TActorContext& ctx);
     void SendPartitionReady(const TActorContext& ctx);
 
+    const std::set<NPQ::TPartitionGraph::Node*>& GetParents() const;
 private:
     const TActorId ParentId;
     const TString InternalClientId;
@@ -267,7 +268,7 @@ class TPartitionActor : public NActors::TActorBootstrapped<TPartitionActor> {
     bool ReadingFinishedSent;
 
     std::unordered_map<ui64, std::shared_ptr<NKikimr::NGRpcProxy::V1::TDistributedCommitHelper>> Kqps;
-    std::set<NPQ::TPartitionGraph::Node*> Parents;
+    const TTopicHolder::TPtr TopicHolder;
 };
 
 
@@ -347,8 +348,8 @@ void TReadSessionActor::Die(const TActorContext& ctx) {
     }
 
     for (auto& t : Topics) {
-        if (t.second.PipeClient)
-            NTabletPipe::CloseClient(ctx, t.second.PipeClient);
+        if (t.second->PipeClient)
+            NTabletPipe::CloseClient(ctx, t.second->PipeClient);
     }
     LOG_INFO_S(ctx, NKikimrServices::PQ_READ_PROXY, PQ_LOG_PREFIX << " is DEAD");
 
@@ -844,7 +845,7 @@ void TReadSessionActor::RegisterSessions(const TActorContext& ctx) {
 
     for (auto& t : Topics) {
         auto& topic = t.first;
-        RegisterSession(t.second.PipeClient, topic, ctx);
+        RegisterSession(t.second->PipeClient, topic, ctx);
         NumPartitionsFromTopic[topic] = 0;
     }
 }
@@ -978,14 +979,14 @@ void TReadSessionActor::Handle(V1::TEvPQProxy::TEvAuthResultOk::TPtr& ev, const
 
         for (auto& [name, t] : ev->Get()->TopicAndTablets) {
             auto& topicHolder = Topics[t.TopicNameConverter->GetInternalName()];
-            topicHolder.TabletID = t.TabletID;
-            topicHolder.CloudId = t.CloudId;
-            topicHolder.DbId = t.DbId;
-            topicHolder.DbPath = t.DbPath;
-            topicHolder.IsServerless = t.IsServerless;
-            topicHolder.FolderId = t.FolderId;
-            topicHolder.FullConverter = t.TopicNameConverter;
-            topicHolder.PartitionGraph = t.PartitionGraph;
+            topicHolder->TabletID = t.TabletID;
+            topicHolder->CloudId = t.CloudId;
+            topicHolder->DbId = t.DbId;
+            topicHolder->DbPath = t.DbPath;
+            topicHolder->IsServerless = t.IsServerless;
+            topicHolder->FolderId = t.FolderId;
+            topicHolder->FullConverter = t.TopicNameConverter;
+            topicHolder->PartitionGraph = t.PartitionGraph;
             FullPathToConverter[t.TopicNameConverter->GetPrimaryPath()] = t.TopicNameConverter;
             const auto& second = t.TopicNameConverter->GetSecondaryPath();
             if (!second.empty()) {
@@ -999,7 +1000,7 @@ void TReadSessionActor::Handle(V1::TEvPQProxy::TEvAuthResultOk::TPtr& ev, const
             clientConfig.CheckAliveness = false;
 
             clientConfig.RetryPolicy = RetryPolicyForPipes;
-            t.second.PipeClient = ctx.RegisterWithSameMailbox(NTabletPipe::CreateClient(ctx.SelfID, t.second.TabletID, clientConfig));
+            t.second->PipeClient = ctx.RegisterWithSameMailbox(NTabletPipe::CreateClient(ctx.SelfID, t.second->TabletID, clientConfig));
         }
 
         RegisterSessions(ctx);
@@ -1014,7 +1015,7 @@ void TReadSessionActor::Handle(V1::TEvPQProxy::TEvAuthResultOk::TPtr& ev, const
                              NPersQueue::NErrorCode::BAD_REQUEST, ctx);
                 return;
             }
-            it->second.PartitionGraph = t.PartitionGraph;
+            it->second->PartitionGraph = t.PartitionGraph;
         }
     }
 
@@ -1054,7 +1055,7 @@ void TReadSessionActor::Handle(TEvPersQueue::TEvLockPartition::TPtr& ev, const T
     Y_ABORT_UNLESS(!intName.empty());
     auto jt = Topics.find(intName);
 
-    if (jt == Topics.end() || pipe != jt->second.PipeClient) { //this is message from old version of pipe
+    if (jt == Topics.end() || pipe != jt->second->PipeClient) { //this is message from old version of pipe
         LOG_DEBUG_S(
                 ctx, NKikimrServices::PQ_READ_PROXY,
                 PQ_LOG_PREFIX << " ignored ev lock for topic = " << converterIter->second->GetPrintableString()
@@ -1063,7 +1064,7 @@ void TReadSessionActor::Handle(TEvPersQueue::TEvLockPartition::TPtr& ev, const T
         return;
     }
 
-    auto* partitionNode = jt->second.PartitionGraph->GetPartition(record.GetPartition());
+    auto* partitionNode = jt->second->PartitionGraph->GetPartition(record.GetPartition());
     if (!partitionNode) {
         LOG_DEBUG_S(
             ctx, NKikimrServices::PQ_READ_PROXY,
@@ -1079,7 +1080,7 @@ void TReadSessionActor::Handle(TEvPersQueue::TEvLockPartition::TPtr& ev, const T
     // ToDo[counters]
     if (NumPartitionsFromTopic[intName]++ == 0) {
         if (AppData(ctx)->PQConfig.GetTopicsAreFirstClassCitizen()) {
-            SetupTopicCounters(converterIter->second, jt->second.CloudId, jt->second.DbId, jt->second.DbPath, jt->second.IsServerless, jt->second.FolderId);
+            SetupTopicCounters(converterIter->second, jt->second->CloudId, jt->second->DbId, jt->second->DbPath, jt->second->IsServerless, jt->second->FolderId);
         } else {
             SetupTopicCounters(converterIter->second);
         }
@@ -1088,11 +1089,11 @@ void TReadSessionActor::Handle(TEvPersQueue::TEvLockPartition::TPtr& ev, const T
     auto it = TopicCounters.find(intName);
     Y_ABORT_UNLESS(it != TopicCounters.end());
 
-    auto database = jt->second.DbPath;
+    auto database = jt->second->DbPath;
     IActor* partitionActor = new TPartitionActor(
             ctx.SelfID, InternalClientId, Cookie, Session, record.GetGeneration(),
-            record.GetStep(), jt->second.FullConverter, database, record.GetPartition(), record.GetTabletId(), it->second,
-            ClientDC, jt->second.PartitionGraph->GetPartition(record.GetPartition())->AllParents
+            record.GetStep(), jt->second->FullConverter, database, record.GetPartition(), record.GetTabletId(), it->second,
+            ClientDC, jt->second
     );
 
     TActorId actorId = ctx.Register(partitionActor);
@@ -1102,8 +1103,8 @@ void TReadSessionActor::Handle(TEvPersQueue::TEvLockPartition::TPtr& ev, const T
     Y_ABORT_UNLESS(record.GetGeneration() > 0);
     //Partitions use clientside name !
     auto pp = Partitions.insert({
-        std::make_pair(jt->second.FullConverter->GetClientsideName(), record.GetPartition()),
-        TPartitionActorInfo{actorId, (((ui64)record.GetGeneration()) << 32) + record.GetStep(), jt->second.FullConverter}
+        std::make_pair(jt->second->FullConverter->GetClientsideName(), record.GetPartition()),
+        TPartitionActorInfo{actorId, (((ui64)record.GetGeneration()) << 32) + record.GetStep(), jt->second->FullConverter}
     });
     Y_ABORT_UNLESS(pp.second);
     if (SessionsActive) {
@@ -1142,7 +1143,7 @@ void TReadSessionActor::Handle(TEvPQProxy::TEvPartitionStatus::TPtr& ev, const T
         Y_ABORT_UNLESS(ClientsideLocksAllowed);
         TReadResponse result;
         auto lock = result.MutableLock();
-        lock->SetTopic(topicIter->second.FullConverter->GetClientsideName());
+        lock->SetTopic(topicIter->second->FullConverter->GetClientsideName());
         lock->SetPartition(ev->Get()->Partition);
         lock->SetReadOffset(ev->Get()->Offset);
         lock->SetEndOffset(ev->Get()->EndOffset);
@@ -1211,7 +1212,7 @@ void TReadSessionActor::Handle(TEvPersQueue::TEvReleasePartition::TPtr& ev, cons
 
     TActorId pipe = ActorIdFromProto(record.GetPipeClient());
 
-    if (pipe != it->second.PipeClient) { //this is message from old version of pipe
+    if (pipe != it->second->PipeClient) { //this is message from old version of pipe
         return;
     }
 
@@ -1286,14 +1287,14 @@ void TReadSessionActor::InformBalancerAboutRelease(const THashMap<std::pair<TStr
     Y_ABORT_UNLESS(jt != Topics.end());
 
     req.SetSession(Session);
-    ActorIdToProto(jt->second.PipeClient, req.MutablePipeClient());
+    ActorIdToProto(jt->second->PipeClient, req.MutablePipeClient());
     req.SetClientId(InternalClientId);
     req.SetTopic(it->first.first);
     req.SetPartition(it->first.second);
 
     LOG_INFO_S(ctx, NKikimrServices::PQ_READ_PROXY, PQ_LOG_PREFIX << " released: " << it->first.first << ":" << it->first.second);
 
-    NTabletPipe::SendData(ctx, jt->second.PipeClient, request.Release());
+    NTabletPipe::SendData(ctx, jt->second->PipeClient, request.Release());
 }
 
 
@@ -1428,7 +1429,7 @@ bool TReadSessionActor::ProcessReleasePartition(const THashMap<std::pair<TString
 
 bool TReadSessionActor::ProcessBalancerDead(const ui64 tablet, const TActorContext& ctx) {
     for (auto& t : Topics) {
-        if (t.second.TabletID == tablet) {
+        if (t.second->TabletID == tablet) {
             LOG_INFO_S(ctx, NKikimrServices::PQ_READ_PROXY, PQ_LOG_PREFIX << " balancer for topic " << t.first << " is dead, restarting all from this topic");
 
             //Drop all partitions from this topic
@@ -1451,14 +1452,14 @@ bool TReadSessionActor::ProcessBalancerDead(const ui64 tablet, const TActorConte
             NTabletPipe::TClientConfig clientConfig;
             clientConfig.CheckAliveness = false;
             clientConfig.RetryPolicy = RetryPolicyForPipes;
-            t.second.PipeClient = ctx.RegisterWithSameMailbox(NTabletPipe::CreateClient(ctx.SelfID, t.second.TabletID, clientConfig));
+            t.second->PipeClient = ctx.RegisterWithSameMailbox(NTabletPipe::CreateClient(ctx.SelfID, t.second->TabletID, clientConfig));
             if (InitDone) {
                 if (PipeReconnects) {
                     ++(*PipeReconnects);
                     ++(*Errors);
                 }
 
-                RegisterSession(t.second.PipeClient, t.first, ctx);
+                RegisterSession(t.second->PipeClient, t.first, ctx);
             }
         }
     }
@@ -1833,7 +1834,7 @@ void TReadSessionActor::Handle(TEvPQProxy::TEvReadingStarted::TPtr& ev, const TA
     }
 
     auto& topic = it->second;
-    NTabletPipe::SendData(ctx, topic.PipeClient, new TEvPersQueue::TEvReadingPartitionStartedRequest(InternalClientId, msg->PartitionId));
+    NTabletPipe::SendData(ctx, topic->PipeClient, new TEvPersQueue::TEvReadingPartitionStartedRequest(InternalClientId, msg->PartitionId));
 }
 
 void TReadSessionActor::Handle(TEvPQProxy::TEvReadingFinished::TPtr& ev, const TActorContext& ctx) {
@@ -1845,7 +1846,7 @@ void TReadSessionActor::Handle(TEvPQProxy::TEvReadingFinished::TPtr& ev, const T
     }
 
     auto& topic = it->second;
-    NTabletPipe::SendData(ctx, topic.PipeClient, new TEvPersQueue::TEvReadingPartitionFinishedRequest(InternalClientId, msg->PartitionId, false, msg->FirstMessage));
+    NTabletPipe::SendData(ctx, topic->PipeClient, new TEvPersQueue::TEvReadingPartitionFinishedRequest(InternalClientId, msg->PartitionId, false, msg->FirstMessage));
 }
 
 
@@ -1854,7 +1855,7 @@ void TReadSessionActor::Handle(TEvPQProxy::TEvReadingFinished::TPtr& ev, const T
 TPartitionActor::TPartitionActor(
         const TActorId& parentId, const TString& internalClientId, const ui64 cookie, const TString& session,
         const ui32 generation, const ui32 step, const NPersQueue::TTopicConverterPtr& topic, const TString& database, const ui32 partition,
-        const ui64 tabletID, const TReadSessionActor::TTopicCounters& counters, const TString& clientDC, std::set<NPQ::TPartitionGraph::Node*> parents
+        const ui64 tabletID, const TReadSessionActor::TTopicCounters& counters, const TString& clientDC, const TTopicHolder::TPtr& topicHolder
 )
     : ParentId(parentId)
     , InternalClientId(internalClientId)
@@ -1891,7 +1892,7 @@ TPartitionActor::TPartitionActor(
     , Counters(counters)
     , FirstRead(true)
     , ReadingFinishedSent(false)
-    , Parents(parents)
+    , TopicHolder(topicHolder)
 {
 }
 
@@ -1922,11 +1923,21 @@ void TPartitionActor::CheckRelease(const TActorContext& ctx) {
     }
 }
 
+const std::set<NPQ::TPartitionGraph::Node*>& TPartitionActor::GetParents() const {
+    const auto* partition = TopicHolder->PartitionGraph->GetPartition(Partition);
+    if (partition) {
+        return partition->AllParents;
+    }
+
+    static std::set<NPQ::TPartitionGraph::Node*> empty;
+    return empty;
+}
 
 void TPartitionActor::SendCommit(const ui64 readId, const ui64 offset, const TActorContext& ctx) {
-    if (!ClientHasAnyCommits && Parents.size() != 0) {
+    const auto& parents = GetParents();
+    if (!ClientHasAnyCommits && parents.size() != 0) {
         std::vector<NKikimr::NGRpcProxy::V1::TDistributedCommitHelper::TCommitInfo> commits;
-        for (auto& parent: Parents) {
+        for (auto& parent: parents) {
             NKikimr::NGRpcProxy::V1::TDistributedCommitHelper::TCommitInfo commit {.PartitionId = parent->Id, .Offset = Max<i64>(), .KillReadSession = false, .OnlyCheckCommitedToFinish = true, .ReadSessionId = Session};
             commits.push_back(commit);
         }