You are an expert in writing code in many different languages. Your goal is to perform code completion for the following code, keeping in mind the rest of the code and the file meta data.
    Metadata details:
    filename: fetch_response.rb
    path of file: ./repos/ruby-kafka/lib/kafka/protocol
    
    the code of the file until where you have to start completion: # frozen_string_literal: true

require "kafka/protocol/message_set"
require "kafka/protocol/record_batch"

module Kafka
  module Protocol

    # A response to a fetch request.
    #
    # ## API Specification
    #
    #     FetchResponse => ThrottleTimeMS [TopicName [Partition ErrorCode HighwaterMarkOffset LastStableOffset [AbortedTransaction] Records]]
    #       ThrottleTimeMS => int32
    #       TopicName => string
    #       Partition => int32
    #       ErrorCode => int16
    #       HighwaterMarkOffset => int64
    #       LastStableOffset => int64
    #       MessageSetSize => int32
    #       AbortedTransaction => [
    #             ProducerId => int64
    #             FirstOffset => int64
    #       ]
    #
    class FetchResponse
      MAGIC_BYTE_OFFSET = 16
      MAGIC_BYTE_LENGTH = 1

      class FetchedPartition
        attr_reader :partition, :error_code
        attr_reader :highwater_mark_offset, :last_stable_offset, :aborted_transactions, :messages

        def initialize(partition:, error_code:, highwater_mark_offset:, last_stable_offset:, aborted_transactions:, messages:)
          @partition = partition
          @error_code = error_code
          @highwater_mark_offset = highwater_mark_offset
          @messages = messages
          @last_stable_offset = last_stable_offset
     
    
      def self.decode(decoder)
        throttle_time_ms = decoder.int32

        topics = decoder.array do
          topic_name = decoder.string

          partitions = decoder.array do
            partition = decoder.int32
            error_code = decoder.int16
            highwater_mark_offset = decoder.int64
            last_stable_offset = decoder.int64

            aborted_transactions = decoder.array do
              producer_id = decoder.int64
              first_offset = decoder.int64
              AbortedTransaction.new(
                producer_id: producer_id,
                first_offset: first_offset
              )
            end

            messages_raw = decoder.bytes
            messages = []

            if !messages_raw.nil? && !messages_raw.empty?
              messages_decoder = Decoder.from_string(messages_raw)

              magic_byte = messages_decoder.peek(MAGIC_BYTE_OFFSET, MAGIC_BYTE_LENGTH)[0].to_i
              if magic_byte == RecordBatch::MAGIC_BYTE
                until messages_decoder.eof?
                  begin
                    record_batch = RecordBatch.decode(messages_decoder)
                    messages << record_batch
                  rescue InsufficientDataMessage
                    if messages.length > 0
                      break
                    else
                      raise
                    end
                  end
                end