code: def fetch_offsets(zk, consumer_group, topics):
      
      source_offsets = defaultdict(dict)
      for topic, partitions in six.iteritems(topics):
          for partition in partitions:
              offset, _ = zk.get(
                  "/consumers/{groupid}/offsets/{topic}/{partition}".format(
                      groupid=consumer_group,
                      topic=topic,
                      partition=partition,
                  )
              )
              source_offsets[topic][partition] = offset
      return source_offsets
docs: Fetch offsets for given topics of given consumer group.
  
      :param zk: Zookeeper client
      :param consumer_group: Consumer group id for given offsets
      :type consumer_group: int
      :rtype: dict(topic, dict(partition, offset))