original_docstring: 
          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_consumer_groups().
          
docstring: Fetch Consumer Group offsets from Zookeeper.
  Also fetch consumer_groups, topics, and partitions if not
  already specified in consumer_groups.
docstring_tokens: Fetch Consumer Group offsets from Zookeeper . Also fetch consumer_groups topics and partitions if not already specified in consumer_groups .
code: 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_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 consumer_groups.iteritems():
                  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')}
  
                  for topic, partitions in topics.iteritems():
                      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')]
  
                      # 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
code_tokens: def _get_zk_consumer_offsets ( self , zk_hosts_ports , consumer_groups = None , zk_prefix = '' ) : zk_consumer_offsets = { } 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 : 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 consumer_groups . iteritems ( ) : if topics is None : 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' ) } for topic , partitions in topics . iteritems ( ) : if partitions is not None : partitions = set ( partitions ) else : zk_path_partitions = zk_path_partition_tmpl . format ( group = consumer_group , topic = topic ) partitions = [ int ( x ) for x in self . _get_zk_path_children ( zk_conn , zk_path_partitions , 'partitions' ) ] 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
short_docstring: Fetch Consumer Group offsets from Zookeeper.
short_docstring_tokens: Fetch Consumer Group offsets from Zookeeper .
comment: """
          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_consumer_groups().
          """$SEP$# Construct the Zookeeper path pattern$SEP$# /consumers/[groupId]/offsets/[topic]/[partitionId]$SEP$# If consumer groups aren't specified, fetch them from ZK$SEP$# If topics are't specified, fetch them from ZK$SEP$# defend against bad user input$SEP$# If partitions aren't specified, fetch them from ZK$SEP$# Zookeeper returns the partition IDs as strings because$SEP$# they are extracted from the node path$SEP$# Fetch consumer offsets for each partition from ZK