def last_offsets_for(*topics)
      @cluster.add_target_topics(topics)
      topics.map {|topic|
        partition_ids = @cluster.partitions_for(topic).collect(&:partition_id)
        partition_offsets = @cluster.resolve_offsets(topic, partition_ids, :latest)
        [topic, partition_offsets.collect { |k, v| [k, v - 1] }.to_h]
      }.to_h
    end
def last_offsets_for(*topics)
      @cluster.add_target_topics(topics)
      topics.map {|topic|
        partition_ids = @cluster.partitions_for(topic).collect(&:partition_id)
        partition_offsets = @cluster.resolve_offsets(topic, partition_ids, :latest)
        [topic, partition_offsets.collect { |k, v| [k, v - 1] }.to_h]
      }.to_h
    end
Retrieve the offset of the last message in each partition of the specified topics.

 @param topics [Array<String>] topic names.
 @return [Hash<String, Hash<Integer, Integer>>]
 @example
   last_offsets_for('topic-1', 'topic-2') # =>
   # {
   #   'topic-1' => { 0 => 100, 1 => 100 },
   #   'topic-2' => { 0 => 100, 1 => 100 }
   # }