def consume_one(topic)
      consumer = build_consumer(topic, true, @config)
      msg = nil
      consumer.each_message do |message|
        msg = decode(message)
        consumer.mark_message_as_processed(message)
        consumer.commit_offsets
        consumer.stop
      end
      msg
    ensure
      consumer.stop
    end
def consume_one(topic)
      consumer = build_consumer(topic, true, @config)
      msg = nil
      consumer.each_message do |message|
        msg = decode(message)
        consumer.mark_message_as_processed(message)
        consumer.commit_offsets
        consumer.stop
      end
      msg
    ensure
      consumer.stop
    end
avro   - Avro instance for deserializing records [AvroTurf::Messaging]
 kafka  - Kafka instance for creating consumers [Kafka]
 config - consumer configuration (optional) [Hash]
 Consume a single record from any partition.
 Will block indefinitely if no record present.

 topic  - topic to consume from [String]
 return - deserialized record [String]