[kafka] YT-24188: Fix Metadata response format

* Changelog entry
Type: fix
Component: kafka-proxy

Fix Metadata response format
commit_hash:ca3f6ffea4e1cad12d638e8eb30bd1a4072cbbe6

2 files changed, 6 insertions, 3 deletions

diff --git a/yt/yt/client/kafka/requests.cpp b/yt/yt/client/kafka/requests.cpp
index f81e97040d..f1b65d4469 100644
--- a/yt/yt/client/kafka/requests.cpp
+++ b/yt/yt/client/kafka/requests.cpp
@@ -289,7 +289,7 @@ void TRspMetadataBroker::Serialize(IKafkaProtocolWriter* writer, int apiVersion)
     writer->WriteString(Host);
     writer->WriteInt32(Port);
     if (apiVersion >= 1) {
-        writer->WriteString(Rack);
+        writer->WriteNullableString(Rack);
     }
     if (apiVersion >= 9) {
         NKafka::Serialize(TagBuffer, writer, /*isCompact*/ true);
@@ -332,6 +332,9 @@ void TRspMetadataTopic::Serialize(IKafkaProtocolWriter* writer, int apiVersion)
 void TRspMetadata::Serialize(IKafkaProtocolWriter* writer, int apiVersion) const
 {
     NKafka::Serialize(Brokers, writer, apiVersion >= 9, apiVersion);
+    if (apiVersion >= 2) {
+        writer->WriteNullableString(ClusterId);
+    }
     if (apiVersion >= 1) {
         writer->WriteInt32(ControllerId);
     }
diff --git a/yt/yt/client/kafka/requests.h b/yt/yt/client/kafka/requests.h
index 04435ffcaf..0abefada5f 100644
--- a/yt/yt/client/kafka/requests.h
+++ b/yt/yt/client/kafka/requests.h
@@ -187,7 +187,7 @@ struct TRspMetadataBroker
     i32 NodeId = 0;
     TString Host;
     i32 Port = 0;
-    TString Rack;
+    std::optional<TString> Rack;
     std::vector<TTaggedField> TagBuffer;
 
     void Serialize(IKafkaProtocolWriter* writer, int apiVersion) const;
@@ -225,7 +225,7 @@ struct TRspMetadata
 {
     i32 ThrottleTimeMs = 0;
     std::vector<TRspMetadataBroker> Brokers;
-    i32 ClusterId = 0;
+    std::optional<TString> ClusterId;
     i32 ControllerId = 0;
     std::vector<TRspMetadataTopic> Topics;
     std::vector<TTaggedField> TagBuffer;