def consume_each_message
      @consumer.each_message(**@message_processing_opts) do |message|
        message_processor = Phobos::Actions::ProcessMessage.new(
          listener: self,
          message: message,