class Karafka::Persistence::Consumer

def fetch(topic, partition)

Parameters:
  • partition (Integer) -- number of partition for which we want to cache
  • topic (Karafka::Routing::Topic) -- topic instance for which we might cache

Returns:
  • (Karafka::BaseConsumer) - base consumer descendant 
def fetch(topic, partition)
  # We always store a current instance for callback reasons
  if topic.persistent
    all[topic][partition] ||= topic.consumer.new
  else
    all[topic][partition] = topic.consumer.new
  end
end