Improvements after review

ydb-platform · Jan 13, 2025 · fd80112 · fd80112
1 parent 9b57613
commit fd80112
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 28 deletions.
diff --git a/ydb/core/driver_lib/run/kikimr_services_initializers.cpp b/ydb/core/driver_lib/run/kikimr_services_initializers.cpp
@@ -1723,6 +1723,8 @@ void TGRpcServicesInitializer::InitializeServices(NActors::TActorSystemSetup* se
             desc->Ssl = kakfaConfig.HasSslCertificate();
 
             desc->EndpointId = NGRpcService::KafkaEndpointId;
+            endpoints.push_back(std::move(desc));
+
         }
 
         for (auto &sx : config.GetExtEndpoints()) {
@@ -2773,13 +2775,11 @@ void TKafkaProxyServiceInitializer::InitializeServices(NActors::TActorSystemSetu
         settings.CertificateFile = Config.GetKafkaProxyConfig().GetCert();
         settings.PrivateKeyFile = Config.GetKafkaProxyConfig().GetKey();
 
-        if (Config.GetKafkaProxyConfig().GetEnableEndpointDiscovery()) {
-            setup->LocalServices.emplace_back(
-                NKafka::MakeKafkaDiscoveryCacheID(),
-                TActorSetupCmd(CreateDiscoveryCache(NGRpcService::KafkaEndpointId),
-                    TMailboxType::HTSwap, appData->UserPoolId)
-            );
-        }
+        setup->LocalServices.emplace_back(
+            NKafka::MakeKafkaDiscoveryCacheID(),
+            TActorSetupCmd(CreateDiscoveryCache(NGRpcService::KafkaEndpointId),
+                TMailboxType::HTSwap, appData->UserPoolId)
+        );
         setup->LocalServices.emplace_back(
             TActorId(),
             TActorSetupCmd(NKafka::CreateKafkaListener(MakePollerActorId(), settings, Config.GetKafkaProxyConfig(),

diff --git a/ydb/core/kafka_proxy/actors/kafka_metadata_actor.cpp b/ydb/core/kafka_proxy/actors/kafka_metadata_actor.cpp
@@ -11,8 +11,9 @@ using namespace NKikimr::NGRpcProxy::V1;
 
 NActors::IActor* CreateKafkaMetadataActor(const TContext::TPtr context,
                                           const ui64 correlationId,
-                                          const TMessagePtr<TMetadataRequestData>& message) {
-    return new TKafkaMetadataActor(context, correlationId, message);
+                                          const TMessagePtr<TMetadataRequestData>& message,
+                                          const TActorId& discoveryCacheActor) {
+    return new TKafkaMetadataActor(context, correlationId, message, discoveryCacheActor);
 }
 
 void TKafkaMetadataActor::Bootstrap(const TActorContext& ctx) {
@@ -23,10 +24,7 @@ void TKafkaMetadataActor::Bootstrap(const TActorContext& ctx) {
     if (WithProxy) {
         AddProxyNodeToBrokers();
     } else {
-        if (Context->Config.GetEnableEndpointDiscovery())
-            SendDiscoveryRequest();
-        else
-            RequestICNodeCache();
+        SendDiscoveryRequest();
 
         if (Message->Topics.size() == 0) {
             NeedCurrentNode = true;
@@ -42,10 +40,7 @@ void TKafkaMetadataActor::Bootstrap(const TActorContext& ctx) {
 }
 
 void TKafkaMetadataActor::SendDiscoveryRequest() {
-    if (!DiscoveryCacheActor) {
-        OwnDiscoveryCache = true;
-        DiscoveryCacheActor = Register(CreateDiscoveryCache(NGRpcService::KafkaEndpointId));
-    }
+    Y_VERIFY_DEBUG(DiscoveryCacheActor);
     PendingResponses++;
     Register(CreateDiscoverer(&MakeEndpointsBoardPath, Context->DatabasePath, SelfId(), DiscoveryCacheActor));
 }
@@ -270,6 +265,7 @@ void TKafkaMetadataActor::RespondIfRequired(const TActorContext& ctx) {
             return;
         }
         AddBroker(nodeIter->first, nodeIter->second.Host, nodeIter->second.Port);
+        NeedCurrentNode = false;
     }
     while (!PendingTopicResponses.empty()) {
         auto& [index, ev] = *PendingTopicResponses.begin();
@@ -293,14 +289,6 @@ void TKafkaMetadataActor::RespondIfRequired(const TActorContext& ctx) {
     Respond();
 }
 
-void TKafkaMetadataActor::Die(const TActorContext& ctx) {
-    if (OwnDiscoveryCache) {
-        Send(DiscoveryCacheActor, new TEvents::TEvPoison());
-        OwnDiscoveryCache = false;
-    }
-    TActor::Die(ctx);
-}
-
 TString TKafkaMetadataActor::LogPrefix() const {
     return TStringBuilder() << "TKafkaMetadataActor " << SelfId() << " ";
 }

diff --git a/ydb/core/kafka_proxy/actors/kafka_metadata_actor.h b/ydb/core/kafka_proxy/actors/kafka_metadata_actor.h
@@ -12,7 +12,7 @@ namespace NKafka {
 class TKafkaMetadataActor: public NActors::TActorBootstrapped<TKafkaMetadataActor> {
 public:
     TKafkaMetadataActor(const TContext::TPtr context, const ui64 correlationId, const TMessagePtr<TMetadataRequestData>& message,
-                        const TActorId& discoveryCacheActor = {})
+                        const TActorId& discoveryCacheActor)
         : Context(context)
         , CorrelationId(correlationId)
         , Message(message)
@@ -60,7 +60,6 @@ class TKafkaMetadataActor: public NActors::TActorBootstrapped<TKafkaMetadataActo
     }
 
     TString LogPrefix() const;
-    void Die(const TActorContext& ctx) override;
 
 private:
     const TContext::TPtr Context;
@@ -76,7 +75,6 @@ class TKafkaMetadataActor: public NActors::TActorBootstrapped<TKafkaMetadataActo
     EKafkaErrors ErrorCode = EKafkaErrors::NONE_ERROR;
 
     TActorId DiscoveryCacheActor;
-    bool OwnDiscoveryCache = false;
     bool NeedCurrentNode = false;
     bool HaveError = false;
     bool FallbackToIcDiscovery = false;