def commit(*msgs)
      tp = prepare_for_commit(*msgs)

      tp.each do |topic, po|
        po.each do |partition, offset|
          Rafka.wrap_errors do
            @redis.rpush("acks", "#{topic}:#{partition}:#{offset}")
          end
        end
      end

      tp
    end
def commit(*msgs)
      tp = prepare_for_commit(*msgs)

      tp.each do |topic, po|
        po.each do |partition, offset|
          Rafka.wrap_errors do
            @redis.rpush("acks", "#{topic}:#{partition}:#{offset}")
          end
        end
      end

      tp
    end
Commit offsets for the given messages.

 If more than one messages refer to the same topic/partition pair,
 only the largest offset amongst them is committed.

 @note This is non-blocking operation; a successful server reply means
   offsets are received by the server and will _eventually_ be submitted
   to Kafka. It is not guaranteed that offsets will be actually committed
   in case of failures.

 @param msgs [Array<Message>] any number of messages for which to commit
   offsets

 @raise [ConsumeError] if there was a server error

 @return [Hash{String=>Hash{Fixnum=>Fixnum}}] the actual offsets sent
   to the server for commit. Keys contain topics while values contain
   the respective partition/offset pairs.