Fix kafka with enabled proxy (#2017)

Author: niksaveliev

ydb/core/kafka_proxy/actors/actors.h

@@ -10,6 +10,11 @@
 
 namespace NKafka {
 
+static constexpr int ProxyNodeId = 1;
+static constexpr char UnderlayPrefix[] = "u-";
+
+static_assert(sizeof(UnderlayPrefix) == 3);
+
 enum EAuthSteps {
  WAIT_HANDSHAKE,
  WAIT_AUTH,

ydb/core/kafka_proxy/actors/kafka_find_coordinator_actor.cpp

@@ -25,7 +25,7 @@ void TKafkaFindCoordinatorActor::Bootstrap(const NActors::TActorContext& ctx) {
 
  bool withProxy = Context->Config.HasProxy() && !Context->Config.GetProxy().GetHostname().Empty();
  if (withProxy) {
- SendResponseOkAndDie(Context->Config.GetProxy().GetHostname(), Context->Config.GetProxy().GetPort(), -1, ctx);
+ SendResponseOkAndDie(Context->Config.GetProxy().GetHostname(), Context->Config.GetProxy().GetPort(), NKafka::ProxyNodeId, ctx);
  return;
  }
 
@@ -54,6 +54,8 @@ void TKafkaFindCoordinatorActor::SendResponseOkAndDie(const TString& host, i32 p
  response->Port = port;
  response->NodeId = nodeId;
 
+ KAFKA_LOG_D("FIND_COORDINATOR response. Host#: " << host << ", Port#: " << port << ", NodeId# " << nodeId);
+
  Send(Context->ConnectionId, new TEvKafka::TEvResponse(CorrelationId, response, static_cast<EKafkaErrors>(response->ErrorCode)));
  Die(ctx);
 }
@@ -71,15 +73,21 @@ void TKafkaFindCoordinatorActor::SendResponseFailAndDie(EKafkaErrors error, cons
 
  response->Coordinators.push_back(coordinator);
  }
- 
+
+ response->ErrorCode = error;
+ 
  Send(Context->ConnectionId, new TEvKafka::TEvResponse(CorrelationId, response, static_cast<EKafkaErrors>(response->ErrorCode)));
  Die(ctx);
 }
 
 void TKafkaFindCoordinatorActor::Handle(NKikimr::NIcNodeCache::TEvICNodesInfoCache::TEvGetAllNodesInfoResponse::TPtr& ev, const NActors::TActorContext& ctx) {
  auto iter = ev->Get()->NodeIdsMapping->find(ctx.SelfID.NodeId());
  Y_ABORT_UNLESS(!iter.IsEnd());
+ 
 auto host = (*ev->Get()->Nodes)[iter->second].Host;
+ if (host.StartsWith(UnderlayPrefix)) {
+ host = host.substr(sizeof(UnderlayPrefix) - 1);
+ }
  KAFKA_LOG_D("FIND_COORDINATOR incoming TEvGetAllNodesInfoResponse. Host#: " << host);
  SendResponseOkAndDie(host, Context->Config.GetListeningPort(), ctx.SelfID.NodeId(), ctx);
 }

ydb/core/kafka_proxy/actors/kafka_metadata_actor.cpp

@@ -6,11 +6,6 @@
 namespace NKafka {
 using namespace NKikimr::NGRpcProxy::V1;
 
-static constexpr int ProxyNodeId = 1;
-static constexpr char UnderlayPrefix[] = "u-";
-
-static_assert(sizeof(UnderlayPrefix) == 3);
-
 NActors::IActor* CreateKafkaMetadataActor(const TContext::TPtr context,
  const ui64 correlationId,
  const TMessagePtr<TMetadataRequestData>& message) {