lib/karafka/base_responder.rb
def deliver!
      messages_buffer.each_value do |data_elements|
        data_elements.each do |data, options|
          # We map this topic name, so it will match namespaced/etc topic in Kafka
          # @note By default will not change topic (if default mapper used)
          mapped_topic = Karafka::App.config.topic_mapper.outgoing(options[:topic])
          external_options = options.merge(topic: mapped_topic)
          producer(options).call(data, external_options)
        end
      end
    end
def deliver!
      messages_buffer.each_value do |data_elements|
        data_elements.each do |data, options|
          # We map this topic name, so it will match namespaced/etc topic in Kafka
          # @note By default will not change topic (if default mapper used)
          mapped_topic = Karafka::App.config.topic_mapper.outgoing(options[:topic])
          external_options = options.merge(topic: mapped_topic)
          producer(options).call(data, external_options)
        end
      end
    end
Takes all the messages from the buffer and delivers them one by one
 @note This method is executed after the validation, so we're sure that
   what we send is legit and it will go to a proper topics