[kafka] YT-24188: Support ListOffsets method

* Changelog entry
Type: feature
Component: kafka-proxy

Partially support of `ListOffsets` method
commit_hash:15a3207d5cb3e4cad7205af4d51fda4a937db912

Author: savnadya

yt/yt/client/kafka/error.h

@@ -12,6 +12,7 @@ DEFINE_ENUM_WITH_UNDERLYING_TYPE(EErrorCode, i16,
     ((TopicAuthorizationFailed)     (29))
     ((GroupAuthorizationFailed)     (30))
     ((SaslAuthenticationFailed)     (31))
+    ((InvalidTimestamp)             (32))
     ((UnsupportedSaslMechanism)     (33))
 );
 

yt/yt/client/kafka/requests-inl.h

@@ -39,7 +39,11 @@ void Deserialize(std::vector<T>& data, IKafkaProtocolReader* reader, bool isComp
         }
         data.resize(size);
     } else {
-        data.resize(reader->ReadInt32());
+        auto size = reader->ReadInt32();
+        if (size < 0) {
+            return;
+        }
+        data.resize(size);
     }
     for (auto& item : data) {
         item.Deserialize(reader, args...);

yt/yt/client/kafka/requests.cpp

@@ -757,4 +757,57 @@ void TRspProduce::Serialize(IKafkaProtocolWriter* writer, int apiVersion) const
 
 ////////////////////////////////////////////////////////////////////////////////
 
+void TReqListOffsetsTopicPartition::Deserialize(IKafkaProtocolReader* reader, int /*apiVersion*/)
+{
+    PartitionIndex = reader->ReadInt32();
+    Timestamp = reader->ReadInt64(); // TODO: use timestamp?
+    MaxNumOffsets = reader->ReadInt32();
+}
+
+void TReqListOffsetsTopic::Deserialize(IKafkaProtocolReader* reader, int apiVersion)
+{
+    Name = reader->ReadString();
+    Partitions.resize(reader->ReadInt32());
+    for (auto& partition : Partitions) {
+        partition.Deserialize(reader, apiVersion);
+    }
+}
+
+void TReqListOffsets::Deserialize(IKafkaProtocolReader* reader, int apiVersion)
+{
+    ReplicaId = reader->ReadInt32();
+    Topics.resize(reader->ReadInt32());
+    for (auto& topic : Topics) {
+        topic.Deserialize(reader, apiVersion);
+    }
+}
+
+void TRspListOffsetsTopicPartition::Serialize(IKafkaProtocolWriter* writer, int /*apiVersion*/) const
+{
+    writer->WriteInt32(PartitionIndex);
+    writer->WriteErrorCode(ErrorCode);
+    writer->WriteInt64(Offset);
+}
+
+void TRspListOffsetsTopic::Serialize(IKafkaProtocolWriter* writer, int apiVersion) const
+{
+    writer->WriteString(Name);
+    writer->WriteInt32(Partitions.size());
+    for (const auto& partition : Partitions) {
+        partition.Serialize(writer, apiVersion);
+    }
+}
+
+void TRspListOffsets::Serialize(IKafkaProtocolWriter* writer, int apiVersion) const
+{
+    writer->WriteInt32(Topics.size());
+    for (const auto& topic : Topics) {
+        topic.Serialize(writer, apiVersion);
+    }
+}
+
+////////////////////////////////////////////////////////////////////////////////
+
+
+
 } // namespace NYT::NKafka

yt/yt/client/kafka/requests.h

@@ -624,6 +624,71 @@ struct TRspProduce
 
 ////////////////////////////////////////////////////////////////////////////////
 
+struct TReqListOffsetsTopicPartition
+{
+    i32 PartitionIndex = 0;
+    i64 Timestamp = 0;
+    i32 MaxNumOffsets = 0;
+
+    std::vector<TTaggedField> TagBuffer;
+
+    void Deserialize(IKafkaProtocolReader* reader, int apiVersion);
+};
+
+struct TReqListOffsetsTopic
+{
+    TString Name;
+    std::vector<TReqListOffsetsTopicPartition> Partitions;
+
+    std::vector<TTaggedField> TagBuffer;
+
+    void Deserialize(IKafkaProtocolReader* reader, int apiVersion);
+};
+
+struct TReqListOffsets
+{
+    static constexpr ERequestType RequestType = ERequestType::ListOffsets;
+
+    i32 ReplicaId = 0;
+    std::vector<TReqListOffsetsTopic> Topics;
+
+    std::vector<TTaggedField> TagBuffer;
+
+    void Deserialize(IKafkaProtocolReader* reader, int apiVersion);
+};
+
+struct TRspListOffsetsTopicPartition
+{
+    i32 PartitionIndex = 0;
+    NKafka::EErrorCode ErrorCode = EErrorCode::None;
+    i64 Offset = 0;
+
+    std::vector<TTaggedField> TagBuffer;
+
+    void Serialize(IKafkaProtocolWriter* writer, int apiVersion) const;
+};
+
+struct TRspListOffsetsTopic
+{
+    TString Name;
+    std::vector<TRspListOffsetsTopicPartition> Partitions;
+
+    std::vector<TTaggedField> TagBuffer;
+
+    void Serialize(IKafkaProtocolWriter* writer, int apiVersion) const;
+};
+
+struct TRspListOffsets
+{
+    std::vector<TRspListOffsetsTopic> Topics;
+
+    std::vector<TTaggedField> TagBuffer;
+
+    void Serialize(IKafkaProtocolWriter* writer, int apiVersion) const;
+};
+
+////////////////////////////////////////////////////////////////////////////////
+
 } // namespace NYT::NKafka
 
 #define REQUESTS_INL_H_