def prepare_for_commit(*msgs)
      tp = Hash.new { |h, k| h[k] = Hash.new(0) }

      msgs.each do |msg|
        if msg.offset >= tp[msg.topic][msg.partition]
          tp[msg.topic][msg.partition] = msg.offset
        end
      end

      tp
    end
def prepare_for_commit(*msgs)
      tp = Hash.new { |h, k| h[k] = Hash.new(0) }

      msgs.each do |msg|
        if msg.offset >= tp[msg.topic][msg.partition]
          tp[msg.topic][msg.partition] = msg.offset
        end
      end

      tp
    end
Accepts one or more messages and prepare them for commit.

 @param msgs [Array<Message>]

 @return [Hash{String=>Hash{Fixnum=>Fixnum}}] the offsets to be committed.
   Keys denote the topics while values contain the partition=>offset pairs.