def consume_batch(timeout: 1.0, batch_size: 0, batching_max_sec: 0)
      if batch_size == 0 && batching_max_sec == 0
        raise ArgumentError, "one of batch_size or batching_max_sec must be greater than 0"
      end

      raised = false
      start_time = Time.now
      msgs = []

      loop do
        break if batch_size > 0 && msgs.size >= batch_size
        break if batching_max_sec > 0 && (Time.now - start_time >= batching_max_sec)
        msg = consume_one(timeout)
        msgs << msg if msg
      end

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

      msgs
    ensure
      commit(*msgs) if @rafka_opts[:auto_commit] && !raised
    end
def consume_batch(timeout: 1.0, batch_size: 0, batching_max_sec: 0)
      if batch_size == 0 && batching_max_sec == 0
        raise ArgumentError, "one of batch_size or batching_max_sec must be greater than 0"
      end

      raised = false
      start_time = Time.now
      msgs = []

      loop do
        break if batch_size > 0 && msgs.size >= batch_size
        break if batching_max_sec > 0 && (Time.now - start_time >= batching_max_sec)
        msg = consume_one(timeout)
        msgs << msg if msg
      end

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

      msgs
    ensure
      commit(*msgs) if @rafka_opts[:auto_commit] && !raised
    end
Consume a batch of messages.

 Messages are accumulated in a batch until (a) batch_size number of
 messages are accumulated or (b) batching_max_sec seconds have passed.
 When either of the conditions is met the batch is returned.

 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}.

 @note Either one of, or both batch_size and batching_max_sec may be
   provided, but not neither.

 @param timeout [Fixnum] the time in seconds to wait for each message
 @param batch_size [Fixnum] maximum number of messages to accumulate
   in the batch
 @param batching_max_sec [Fixnum] maximum time in seconds to wait for
   messages to accumulate in the batch

 @yieldparam [Array<Message>] msgs the batch

 @raise [MalformedMessageError] if a message cannot be parsed
 @raise [ConsumeError] if there was a server error
 @raise [ArgumentError] if neither batch_size nor batching_max_sec were
   provided

 @return [Array<Message>] the batch

 @example Consume a batch of 10 messages
   msgs = consumer.consume_batch(batch_size: 10)
   msgs.size # => 10

 @example Accumulate messages for 5 seconds and consume the batch
   msgs = consumer.consume_batch(batching_max_sec: 5)
   msgs.size # => 3813