def consume(timeout=5)
      raised = false
      msg = consume_one(timeout)

      return nil if !msg

      begin
        yield(msg) if block_given?
      rescue => e
        raised = true
        raise e
      end

      msg
    ensure
      commit(msg) if @rafka_opts[:auto_commit] && msg && !raised
    end
def consume(timeout=5)
      raised = false
      msg = consume_one(timeout)

      return nil if !msg

      begin
        yield(msg) if block_given?
      rescue => e
        raised = true
        raise e
      end

      msg
    ensure
      commit(msg) if @rafka_opts[:auto_commit] && msg && !raised
    end
Initialize a new consumer.

 @param [Hash] opts
 @option opts [String] :host ("localhost") server hostname
 @option opts [Fixnum] :port (6380) server port
 @option opts [String] :topic Kafka topic to consume (required)
 @option opts [String] :group Kafka consumer group name (required)
 @option opts [String] :id (random) Kafka consumer id
 @option opts [Boolean] :auto_commit (true) automatically commit
   offsets
 @option opts [Hash] :librdkafka ({}) librdkafka configuration. It will
   be merged over the existing configuration set in the server.
   See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
   for more information
 @option opts [Hash] :redis ({}) Configuration for the underlying Redis
   client. See {REDIS_DEFAULTS}

 @raise [RuntimeError] if a required option was not provided
   (see {REQUIRED_OPTS})

 @return [Consumer]
 Consumes the next message.

 If :auto_commit is true, offsets are committed automatically.
 In the block form, offsets are committed only if the block executes
 without raising any exceptions.

 If :auto_commit is false, offsets have to be committed manually using
 {#commit}.

 @param timeout [Fixnum] the time in seconds to wait for a message. If
   reached, {#consume} returns nil.

 @yieldparam [Message] msg the consumed message

 @raise [MalformedMessageError] if the message cannot be parsed
 @raise [ConsumeError] if there was a server error

 @return [nil, Message] the consumed message, or nil of there wasn't any

 @example Consume a message
   msg = consumer.consume
   msg.value # => "hi"

 @example Consume a message and commit offset if the block does not raise an exception
   consumer.consume { |msg| puts "I received #{msg.value}" }