Merge of important bugfixes for PQ read proxy & kafka proxy

ydb/core/kafka_proxy/actors/kafka_metadata_actor.cpp

@@ -230,8 +230,8 @@ void TKafkaMetadataActor::AddTopicResponse(
 void TKafkaMetadataActor::HandleLocationResponse(TEvLocationResponse::TPtr ev, const TActorContext& ctx) {
     --PendingResponses;
 
-    auto* r = ev->Get();
     auto actorIter = TopicIndexes.find(ev->Sender);
+    TSimpleSharedPtr<TEvLocationResponse> locationResponse{ev->Release()};
 
     Y_DEBUG_ABORT_UNLESS(!actorIter.IsEnd());
     Y_DEBUG_ABORT_UNLESS(!actorIter->second.empty());
@@ -248,12 +248,12 @@ void TKafkaMetadataActor::HandleLocationResponse(TEvLocationResponse::TPtr ev, c
 
     for (auto index : actorIter->second) {
         auto& topic = Response->Topics[index];
-        if (r->Status == Ydb::StatusIds::SUCCESS) {
+        if (locationResponse->Status == Ydb::StatusIds::SUCCESS) {
             KAFKA_LOG_D("Describe topic '" << topic.Name << "' location finishied successful");
-            PendingTopicResponses.insert(std::make_pair(index, ev->Release()));
+            PendingTopicResponses.emplace(index, locationResponse);
         } else {
-            KAFKA_LOG_ERROR("Describe topic '" << topic.Name << "' location finishied with error: Code=" << r->Status << ", Issues=" << r->Issues.ToOneLineString());
-            AddTopicError(topic, ConvertErrorCode(r->Status));
+            KAFKA_LOG_ERROR("Describe topic '" << topic.Name << "' location finishied with error: Code=" << locationResponse->Status << ", Issues=" << locationResponse->Issues.ToOneLineString());
+            AddTopicError(topic, ConvertErrorCode(locationResponse->Status));
         }
     }
     RespondIfRequired(ctx);
@@ -307,7 +307,7 @@ void TKafkaMetadataActor::RespondIfRequired(const TActorContext& ctx) {
 
     if (NeedAllNodes) {
         for (const auto& [id, nodeInfo] : Nodes)
-        AddBroker(id, nodeInfo.Host, nodeInfo.Port);
+            AddBroker(id, nodeInfo.Host, nodeInfo.Port);
     }
 
     Respond();

ydb/core/kafka_proxy/actors/kafka_metadata_actor.h

@@ -85,10 +85,11 @@ class TKafkaMetadataActor: public NActors::TActorBootstrapped<TKafkaMetadataActo
     bool NeedAllNodes = false;
     bool HaveError = false;
     bool FallbackToIcDiscovery = false;
-    TMap<ui64, TAutoPtr<TEvLocationResponse>> PendingTopicResponses;
+    TMap<ui64, TSimpleSharedPtr<TEvLocationResponse>> PendingTopicResponses;
 
     THashMap<ui64, TNodeInfo> Nodes;
     THashMap<TString, TActorId> PartitionActors;
+    THashSet<ui64> HaveBrokers;
 
 };
 

ydb/core/kafka_proxy/ut/port_discovery_ut.cpp

@@ -189,12 +189,14 @@ namespace NKafka::NTests {
         }
 
         void CreateMetarequestActor(
-                const TActorId& edge, const TString& topicPath, auto* runtime, const auto& kafkaConfig, const TActorId& fakeCacheId = {}
+                const TActorId& edge, const TVector<TString>& topics, auto* runtime, const auto& kafkaConfig, const TActorId& fakeCacheId = {}
         ) {
             TMetadataRequestData::TPtr metaRequest = std::make_shared<TMetadataRequestData>();
-            metaRequest->Topics.emplace_back();
-            auto& topic = metaRequest->Topics[0];
-            topic.Name = topicPath;
+            for (const auto& topicPath : topics) {
+                metaRequest->Topics.emplace_back();
+                auto& topic = metaRequest->Topics.back();
+                topic.Name = topicPath;
+            }
 
             auto context = std::make_shared<TContext>(kafkaConfig);
             context->ConnectionId = edge;
@@ -215,14 +217,16 @@ namespace NKafka::NTests {
             runtime->EnableScheduleForActor(actorId);
         }
 
-        void CheckKafkaMetaResponse(TTestActorRuntime* runtime, ui64 kafkaPort, bool error = false) {
+        void CheckKafkaMetaResponse(TTestActorRuntime* runtime, ui64 kafkaPort, bool error = false, ui64 expectedCount = 1) {
             TAutoPtr<IEventHandle> handle;
             auto* ev = runtime->GrabEdgeEvent<TEvKafka::TEvResponse>(handle);
             UNIT_ASSERT(ev);
             auto response = dynamic_cast<TMetadataResponseData*>(ev->Response.get());
-            UNIT_ASSERT_VALUES_EQUAL(response->Topics.size(), 1);
+            UNIT_ASSERT_VALUES_EQUAL(response->Topics.size(), expectedCount);
             if (!error) {
-                UNIT_ASSERT(response->Topics[0].ErrorCode == EKafkaErrors::NONE_ERROR);
+                for (const auto& topic : response->Topics) {
+                    UNIT_ASSERT(topic.ErrorCode == EKafkaErrors::NONE_ERROR);
+                }
             } else {
                 UNIT_ASSERT(response->Topics[0].ErrorCode == EKafkaErrors::LISTENER_NOT_FOUND);
                 UNIT_ASSERT(ev->ErrorCode == EKafkaErrors::LISTENER_NOT_FOUND);
@@ -239,7 +243,7 @@ namespace NKafka::NTests {
             auto* runtime = server.GetRuntime();
             auto edge = runtime->AllocateEdgeActor();
 
-            CreateMetarequestActor(edge, NKikimr::JoinPath({"/Root/PQ/", topicName}), runtime,
+            CreateMetarequestActor(edge, {NKikimr::JoinPath({"/Root/PQ/", topicName})}, runtime,
                                    config);
 
             CheckKafkaMetaResponse(runtime, kafkaPort);
@@ -262,7 +266,7 @@ namespace NKafka::NTests {
             ep->set_node_id(9998);
             auto fakeCache = runtime->Register(new TFakeDiscoveryCache(leResult, false));
             runtime->EnableScheduleForActor(fakeCache);
-            CreateMetarequestActor(edge, NKikimr::JoinPath({"/Root/PQ/", topicName}), runtime,
+            CreateMetarequestActor(edge, {NKikimr::JoinPath({"/Root/PQ/", topicName})}, runtime,
                                    config, fakeCache);
 
             CheckKafkaMetaResponse(runtime, kafkaPort);
@@ -277,7 +281,7 @@ namespace NKafka::NTests {
             Ydb::Discovery::ListEndpointsResult leResult;
             auto fakeCache = runtime->Register(new TFakeDiscoveryCache(leResult, true));
             runtime->EnableScheduleForActor(fakeCache);
-            CreateMetarequestActor(edge, NKikimr::JoinPath({"/Root/PQ/", topicName}), runtime,
+            CreateMetarequestActor(edge, {NKikimr::JoinPath({"/Root/PQ/", topicName})}, runtime,
                                    config, fakeCache);
 
             CheckKafkaMetaResponse(runtime, kafkaPort, true);
@@ -296,10 +300,23 @@ namespace NKafka::NTests {
             ep->set_node_id(runtime->GetNodeId(0));
             auto fakeCache = runtime->Register(new TFakeDiscoveryCache(leResult, false));
             runtime->EnableScheduleForActor(fakeCache);
-            CreateMetarequestActor(edge, NKikimr::JoinPath({"/Root/PQ/", topicName}), runtime,
+            CreateMetarequestActor(edge, {NKikimr::JoinPath({"/Root/PQ/", topicName})}, runtime,
                                    config, fakeCache);
 
             CheckKafkaMetaResponse(runtime, 12345);
         }
+
+
+        Y_UNIT_TEST(MetadataActorDoubleTopic) {
+            auto [server, kafkaPort, config, topicName] = SetupServer("topic1");
+
+            auto* runtime = server.GetRuntime();
+            auto edge = runtime->AllocateEdgeActor();
+
+            auto path = NKikimr::JoinPath({"/Root/PQ/", topicName});
+            CreateMetarequestActor(edge, {path, path}, runtime, config);
+
+            CheckKafkaMetaResponse(runtime, kafkaPort, false, 2);
+        }
     }
-}
+}

ydb/core/persqueue/partition.cpp

@@ -2963,6 +2963,7 @@ TPartition::EProcessResult TPartition::PreProcessUserAct(
 void TPartition::CommitUserAct(TEvPQ::TEvSetClientInfo& act) {
     const bool strictCommitOffset = (act.Type == TEvPQ::TEvSetClientInfo::ESCI_OFFSET && act.Strict);
     const TString& user = act.ClientId;
+    RemoveUserAct(user);
     const auto& ctx = ActorContext();
     if (!PendingUsersInfo.contains(user) && AffectedUsers.contains(user)) {
         switch (act.Type) {
@@ -3082,7 +3083,6 @@ void TPartition::CommitUserAct(TEvPQ::TEvSetClientInfo& act) {
         return;
     }
 
-    RemoveUserAct(act.ClientId);
     return EmulatePostProcessUserAct(act, userInfo, ActorContext());
 }
 

ydb/core/persqueue/partition.h

@@ -29,6 +29,7 @@ namespace NKikimr::NPQ {
 
 static const ui32 MAX_BLOB_PART_SIZE = 500_KB;
 static const ui32 DEFAULT_BUCKET_COUNTER_MULTIPLIER = 20;
+static const ui32 MAX_USER_ACTS = 1000;
 
 using TPartitionLabeledCounters = TProtobufTabletLabeledCounters<EPartitionLabeledCounters_descriptor>;
 

ydb/core/persqueue/partition_read.cpp

@@ -28,8 +28,6 @@
 
 namespace NKikimr::NPQ {
 
-static const ui32 MAX_USER_ACTS = 1000;
-
 TMaybe<TInstant> GetReadFrom(ui32 maxTimeLagMs, ui64 readTimestampMs, TInstant consumerReadFromTimestamp, const TActorContext& ctx) {
     if (!(maxTimeLagMs > 0 || readTimestampMs > 0 || consumerReadFromTimestamp > TInstant::MilliSeconds(1))) {
         return {};

ydb/core/persqueue/ut/partition_ut.cpp

@@ -225,7 +225,8 @@ class TPartitionFixture : public NUnitTest::TBaseFixture {
     void SendSetOffset(ui64 cookie,
                        const TString& clientId,
                        ui64 offset,
-                       const TString& sessionId);
+                       const TString& sessionId,
+                       bool strict = false);
     void SendGetOffset(ui64 cookie,
                        const TString& clientId);
     void WaitCmdWrite(const TCmdWriteMatcher& matcher = {});
@@ -488,7 +489,8 @@ void TPartitionFixture::SendCreateSession(ui64 cookie,
 void TPartitionFixture::SendSetOffset(ui64 cookie,
                                       const TString& clientId,
                                       ui64 offset,
-                                      const TString& sessionId)
+                                      const TString& sessionId,
+                                      bool strict)
 {
     auto event = MakeHolder<TEvPQ::TEvSetClientInfo>(cookie,
                                                      clientId,
@@ -498,6 +500,7 @@ void TPartitionFixture::SendSetOffset(ui64 cookie,
                                                      0,
                                                      0,
                                                      TActorId{});
+    event->Strict = strict;
     Ctx->Runtime->SingleSys()->Send(new IEventHandle(ActorId, Ctx->Edge, event.Release()));
 }
 
@@ -712,7 +715,6 @@ void TPartitionFixture::WaitErrorResponse(const TErrorMatcher& matcher)
     }
 
     if (matcher.Error) {
-
         UNIT_ASSERT_VALUES_EQUAL(*matcher.Error, event->Error);
     }
 }
@@ -1796,6 +1798,28 @@ void TPartitionTxTestHelper::WaitTxPredicateReplyImpl(ui64 userActId, bool statu
 #endif
 }
 
+Y_UNIT_TEST_F(UserActCount, TPartitionFixture)
+{
+    // In the test, we check that the reference count for `UserInfo` decreases in case of errors. To do this,
+    // we send a large number of requests to which the server will respond with an error.
+
+    CreatePartition();
+
+    SendCreateSession(1, "client", "session-id", 2, 3);
+    WaitCmdWrite({.Count=2, .UserInfos={{0, {.Session="session-id", .Offset=0, .Generation=2, .Step=3}}}});
+    SendCmdWriteResponse(NMsgBusProxy::MSTATUS_OK);
+    WaitProxyResponse({.Cookie=1});
+
+    for (ui64 k = 0; k <= MAX_USER_ACTS; ++k) {
+        const ui64 cookie = 2 + k;
+        // 1 > EndOffset
+        SendSetOffset(cookie, "client", 1, "session-id", true); // strict = true
+        WaitCmdWrite({.Count=2, .UserInfos={{0, {.Session="session-id", .Offset=0, .Generation=2, .Step=3}}}});
+        SendCmdWriteResponse(NMsgBusProxy::MSTATUS_OK);
+        WaitErrorResponse({.Cookie=cookie, .ErrorCode=NPersQueue::NErrorCode::SET_OFFSET_ERROR_COMMIT_TO_FUTURE});
+    }
+}
+
 Y_UNIT_TEST_F(Batching, TPartitionFixture)
 {
     CreatePartition();

ydb/services/persqueue_v1/actors/partition_actor.cpp

@@ -1305,12 +1305,12 @@ void TPartitionActor::Handle(TEvPQProxy::TEvRead::TPtr& ev, const TActorContext&
     const auto req = ev->Get();
 
     auto request = MakeReadRequest(ReadOffset, 0, req->MaxCount, req->MaxSize, req->MaxTimeLagMs, req->ReadTimestampMs, DirectReadId);
-
+    RequestInfly = true;
+    CurrentRequest = request;
+
     if (!PipeClient) //Pipe will be recreated soon
         return;
 
-    RequestInfly = true;
-    CurrentRequest = request;
     TAutoPtr<TEvPersQueue::TEvRequest> event(new TEvPersQueue::TEvRequest);
     event->Record.Swap(&request);