lib/karafka/base_responder.rb
def validate_usage!
      registered_topics = self.class.topics.map do |name, topic|
        topic.to_h.merge!(
          usage_count: messages_buffer[name]&.count || 0
        )
      end

      used_topics = messages_buffer.map do |name, usage|
        topic = self.class.topics[name] || Responders::Topic.new(name, registered: false)
        topic.to_h.merge!(usage_count: usage.count)
      end

      result = Karafka::Schemas::ResponderUsage.call(
        registered_topics: registered_topics,
        used_topics: used_topics
      )

      return if result.success?

      raise Karafka::Errors::InvalidResponderUsageError, result.errors
    end
def validate_usage!
      registered_topics = self.class.topics.map do |name, topic|
        topic.to_h.merge!(
          usage_count: messages_buffer[name]&.count || 0
        )
      end

      used_topics = messages_buffer.map do |name, usage|
        topic = self.class.topics[name] || Responders::Topic.new(name, registered: false)
        topic.to_h.merge!(usage_count: usage.count)
      end

      result = Karafka::Schemas::ResponderUsage.call(
        registered_topics: registered_topics,
        used_topics: used_topics
      )

      return if result.success?

      raise Karafka::Errors::InvalidResponderUsageError, result.errors
    end
Checks if we met all the topics requirements. It will fail if we didn't send a message to
 a registered required topic, etc.