docs: def _get_zk_consumer_offsets(self, zk_hosts_ports, consumer_groups=None, zk_prefix=''):
          """
          Fetch Consumer Group offsets from Zookeeper.
  
          Also fetch consumer_groups, topics, and partitions if not
          already specified in consumer_groups.
  
          :param dict consumer_groups: The consumer groups, topics, and partitions
              that you want to fetch offsets for. If consumer_groups is None, will
              fetch offsets for all consumer_groups. For examples of what this
              dict can look like, see _validate_explicit_consumer_groups().
          """
          zk_consumer_offsets = {}
  
          # Construct the Zookeeper path pattern
          # /consumers/[groupId]/offsets/[topic]/[partitionId]
          zk_path_consumer = zk_prefix + '/consumers/'
          zk_path_topic_tmpl = zk_path_consumer + '{group}/offsets/'
          zk_path_partition_tmpl = zk_path_topic_tmpl + '{topic}/'
  
          zk_conn = KazooClient(zk_hosts_ports, timeout=self._zk_timeout)
          zk_conn.start()
          try:
              if consumer_groups is None:
                  # If consumer groups aren't specified, fetch them from ZK
                  consumer_groups = {
                      consumer_group: None
                      for consumer_group in self._get_zk_path_children(zk_conn, zk_path_consumer, 'consumer groups')
                  }
  
              for consumer_group, topics in iteritems(consumer_groups):
                  if topics is None:
                      # If topics are't specified, fetch them from ZK
                      zk_path_topics = zk_path_topic_tmpl.format(group=consumer_group)
                      topics = {topic: None for topic in self._get_zk_path_children(zk_conn, zk_path_topics, 'topics')}
                      consumer_groups[consumer_group] = topics
  
                  for topic, partitions in iteritems(topics):
                      if partitions is not None:
                          partitions = set(partitions)  # defend against bad user input
                      else:
                          # If partitions aren't specified, fetch them from ZK
                          zk_path_partitions = zk_path_partition_tmpl.format(group=consumer_group, topic=topic)
                          # Zookeeper returns the partition IDs as strings because
                          # they are extracted from the node path
                          partitions = [
                              int(x) for x in self._get_zk_path_children(zk_conn, zk_path_partitions, 'partitions')
                          ]
                          consumer_groups[consumer_group][topic] = partitions
  
                      # Fetch consumer offsets for each partition from ZK
                      for partition in partitions:
                          zk_path = (zk_path_partition_tmpl + '{partition}/').format(
                              group=consumer_group, topic=topic, partition=partition
                          )
                          try:
                              consumer_offset = int(zk_conn.get(zk_path)[0])
                              key = (consumer_group, topic, partition)
                              zk_consumer_offsets[key] = consumer_offset
                          except NoNodeError:
                              self.log.info('No zookeeper node at %s', zk_path)
                          except Exception:
                              self.log.exception('Could not read consumer offset from %s', zk_path)
          finally:
              try:
                  zk_conn.stop()
                  zk_conn.close()
              except Exception:
                  self.log.exception('Error cleaning up Zookeeper connection')
          return zk_consumer_offsets, consumer_groups