def each_message(min_bytes: 1, max_bytes: 10485760, max_wait_time: 1, automatically_mark_as_processed: true)
      @fetcher.configure(
        min_bytes: min_bytes,
        max_bytes: max_bytes,
        max_wait_time: max_wait_time,
      )

      consumer_loop do
        batches = fetch_batches

        batches.each do |batch|
          batch.messages.each do |message|
            notification = {
              topic: message.topic,
              partition: message.partition,
              offset: message.offset,
              offset_lag: batch.highwater_mark_offset - message.offset - 1,
              create_time: message.create_time,
              key: message.key,
              value: message.value,
              headers: message.headers
            }

            # Instrument an event immediately so that subscribers don't have to wait until
            # the block is completed.
            @instrumenter.instrument("start_process_message.consumer", notification)

            @instrumenter.instrument("process_message.consumer", notification) do
              begin
                yield message unless message.is_control_record
                @current_offsets[message.topic][message.partition] = message.offset
              rescue => e
                location = "#{message.topic}/#{message.partition} at offset #{message.offset}"
                backtrace = e.backtrace.join("\n")
                @logger.error "Exception raised when processing #{location} -- #{e.class}: #{e}\n#{backtrace}"

                raise ProcessingError.new(message.topic, message.partition, message.offset)
              end
            end

            mark_message_as_processed(message) if automatically_mark_as_processed
            @offset_manager.commit_offsets_if_necessary

            trigger_heartbeat

            return if shutting_down?
          end

          # We've successfully processed a batch from the partition, so we can clear
          # the pause.
          pause_for(batch.topic, batch.partition).reset!
        end

        # We may not have received any messages, but it's still a good idea to
        # commit offsets if we've processed messages in the last set of batches.
        # This also ensures the offsets are retained if we haven't read any messages
        # since the offset retention period has elapsed.
        @offset_manager.commit_offsets_if_necessary
      end
    end
def each_message(min_bytes: 1, max_bytes: 10485760, max_wait_time: 1, automatically_mark_as_processed: true)
      @fetcher.configure(
        min_bytes: min_bytes,
        max_bytes: max_bytes,
        max_wait_time: max_wait_time,
      )

      consumer_loop do
        batches = fetch_batches

        batches.each do |batch|
          batch.messages.each do |message|
            notification = {
              topic: message.topic,
              partition: message.partition,
              offset: message.offset,
              offset_lag: batch.highwater_mark_offset - message.offset - 1,
              create_time: message.create_time,
              key: message.key,
              value: message.value,
              headers: message.headers
            }

            # Instrument an event immediately so that subscribers don't have to wait until
            # the block is completed.
            @instrumenter.instrument("start_process_message.consumer", notification)

            @instrumenter.instrument("process_message.consumer", notification) do
              begin
                yield message unless message.is_control_record
                @current_offsets[message.topic][message.partition] = message.offset
              rescue => e
                location = "#{message.topic}/#{message.partition} at offset #{message.offset}"
                backtrace = e.backtrace.join("\n")
                @logger.error "Exception raised when processing #{location} -- #{e.class}: #{e}\n#{backtrace}"

                raise ProcessingError.new(message.topic, message.partition, message.offset)
              end
            end

            mark_message_as_processed(message) if automatically_mark_as_processed
            @offset_manager.commit_offsets_if_necessary

            trigger_heartbeat

            return if shutting_down?
          end

          # We've successfully processed a batch from the partition, so we can clear
          # the pause.
          pause_for(batch.topic, batch.partition).reset!
        end

        # We may not have received any messages, but it's still a good idea to
        # commit offsets if we've processed messages in the last set of batches.
        # This also ensures the offsets are retained if we haven't read any messages
        # since the offset retention period has elapsed.
        @offset_manager.commit_offsets_if_necessary
      end
    end
Fetches and enumerates the messages in the topics that the consumer group
 subscribes to.

 Each message is yielded to the provided block. If the block returns
 without raising an exception, the message will be considered successfully
 processed. At regular intervals the offset of the most recent successfully
 processed message in each partition will be committed to the Kafka
 offset store. If the consumer crashes or leaves the group, the group member
 that is tasked with taking over processing of these partitions will resume
 at the last committed offsets.

 @param min_bytes [Integer] the minimum number of bytes to read before
   returning messages from each broker; if `max_wait_time` is reached, this
   is ignored.
 @param max_bytes [Integer] the maximum number of bytes to read before
   returning messages from each broker.
 @param max_wait_time [Integer, Float] the maximum duration of time to wait before
   returning messages from each broker, in seconds.
 @param automatically_mark_as_processed [Boolean] whether to automatically
   mark a message as successfully processed when the block returns
   without an exception. Once marked successful, the offsets of processed
   messages can be committed to Kafka.
 @yieldparam message [Kafka::FetchedMessage] a message fetched from Kafka.
 @raise [Kafka::ProcessingError] if there was an error processing a message.
   The original exception will be returned by calling `#cause` on the
   {Kafka::ProcessingError} instance.
 @return [nil]