Fix proxy crash on duplicate topics in request (#15467)

Author: FloatingCrowbar

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);
+        }
     }
-}
+}