def lag(topic_partition_list, watermark_timeout_ms=100)
      out = {}
      topic_partition_list.to_h.each do |topic, partitions|
        # Query high watermarks for this topic's partitions
        # and compare to the offset in the list.
        topic_out = {}
        partitions.each do |p|
          next if p.offset.nil?
          low, high = query_watermark_offsets(
            topic,
            p.partition,
            watermark_timeout_ms
          )
          topic_out[p.partition] = high - p.offset
        end
        out[topic] = topic_out
      end
      out
    end
def lag(topic_partition_list, watermark_timeout_ms=100)
      out = {}
      topic_partition_list.to_h.each do |topic, partitions|
        # Query high watermarks for this topic's partitions
        # and compare to the offset in the list.
        topic_out = {}
        partitions.each do |p|
          next if p.offset.nil?
          low, high = query_watermark_offsets(
            topic,
            p.partition,
            watermark_timeout_ms
          )
          topic_out[p.partition] = high - p.offset
        end
        out[topic] = topic_out
      end
      out
    end
Calculate the consumer lag per partition for the provided topic partition list.
 You can get a suitable list by calling {committed} or {position} (TODO). It is also
 possible to create one yourself, in this case you have to provide a list that
 already contains all the partitions you need the lag for.

 @param topic_partition_list [TopicPartitionList] The list to calculate lag for.
 @param watermark_timeout_ms [Integer] The timeout for each query watermark call.

 @raise [RdkafkaError] When querying the broker fails.

 @return [Hash<String, Hash<Integer, Integer>>] A hash containing all topics with the lag per partition