def consume_all(topic)
      config = @config.clone
      config[:group_id] = SecureRandom.uuid
      consumer = build_consumer(topic, true, config)
      consumer.each_message do |message|
        yield decode(message), consumer
      end
    ensure
      consumer.stop
    end
def consume_all(topic)
      config = @config.clone
      config[:group_id] = SecureRandom.uuid
      consumer = build_consumer(topic, true, config)
      consumer.each_message do |message|
        yield decode(message), consumer
      end
    ensure
      consumer.stop
    end
Consume all records from a topic. Each record will be yielded
 to block along with consumer instance. Will listen to topic
 after all records have been consumed.

 topic  - topic to consume from [String]
 yields - record, consumer [String, Kafka::Consumer]
 return - [nil]