def each_message(topic:, start_from_beginning: true, max_wait_time: 5, min_bytes: 1, max_bytes: 1048576, &block)
      default_offset ||= start_from_beginning ? :earliest : :latest
      offsets = Hash.new { default_offset }

      loop do
        operation = FetchOperation.new(
          cluster: @cluster,
          logger: @logger,
          min_bytes: min_bytes,
          max_wait_time: max_wait_time,
        )

        @cluster.partitions_for(topic).map(&:partition_id).each do |partition|
          partition_offset = offsets[partition]
          operation.fetch_from_partition(topic, partition, offset: partition_offset, max_bytes: max_bytes)
        end

        batches = operation.execute

        batches.each do |batch|
          batch.messages.each(&block)
          offsets[batch.partition] = batch.last_offset + 1 unless batch.unknown_last_offset?
        end
      end
    end
def each_message(topic:, start_from_beginning: true, max_wait_time: 5, min_bytes: 1, max_bytes: 1048576, &block)
      default_offset ||= start_from_beginning ? :earliest : :latest
      offsets = Hash.new { default_offset }

      loop do
        operation = FetchOperation.new(
          cluster: @cluster,
          logger: @logger,
          min_bytes: min_bytes,
          max_wait_time: max_wait_time,
        )

        @cluster.partitions_for(topic).map(&:partition_id).each do |partition|
          partition_offset = offsets[partition]
          operation.fetch_from_partition(topic, partition, offset: partition_offset, max_bytes: max_bytes)
        end

        batches = operation.execute

        batches.each do |batch|
          batch.messages.each(&block)
          offsets[batch.partition] = batch.last_offset + 1 unless batch.unknown_last_offset?
        end
      end
    end
Enumerate all messages in a topic.

 @param topic [String] the topic to consume messages from.

 @param start_from_beginning [Boolean] whether to start from the beginning
   of the topic or just subscribe to new messages being produced.

 @param max_wait_time [Integer] the maximum amount of time to wait before
   the server responds, in seconds.

 @param min_bytes [Integer] the minimum number of bytes to wait for. If set to
   zero, the broker will respond immediately, but the response may be empty.
   The default is 1 byte, which means that the broker will respond as soon as
   a message is written to the partition.

 @param max_bytes [Integer] the maximum number of bytes to include in the
   response message set. Default is 1 MB. You need to set this higher if you
   expect messages to be larger than this.

 @return [nil]