class Kafka::Protocol::MetadataResponse

def self.decode(decoder)
  brokers = decoder.array do
    node_id = decoder.int32
    host = decoder.string
    port = decoder.int32
    rack = decoder.string
    BrokerInfo.new(
      node_id: node_id,
      host: host,
      port: port
    )
  end
  controller_id = decoder.int32
  topics = decoder.array do
    topic_error_code = decoder.int16
    topic_name = decoder.string
    is_internal = decoder.boolean
    partitions = decoder.array do
      PartitionMetadata.new(
        partition_error_code: decoder.int16,
        partition_id: decoder.int32,
        leader: decoder.int32,
        replicas: decoder.array { decoder.int32 },
        isr: decoder.array { decoder.int32 },
      )
    end
    TopicMetadata.new(
      topic_error_code: topic_error_code,
      topic_name: topic_name,
      partitions: partitions,
    )
  end
  new(brokers: brokers, controller_id: controller_id, topics: topics)
end