class_docstr: A class to manage kafka cluster metadata.
  
  This class does not perform any IO. It simply updates internal state
  given API responses (MetadataResponse, GroupCoordinatorResponse).
  
  Keyword Arguments:
      retry_backoff_ms (int): Milliseconds to backoff when retrying on
          errors. Default: 100.
      metadata_max_age_ms (int): The period of time in milliseconds after
          which we force a refresh of metadata even if we haven't seen any
          partition leadership changes to proactively discover any new
          brokers or partitions. Default: 300000
      bootstrap_servers: 'host[:port]' string (or list of 'host[:port]'
          strings) that the client should contact to bootstrap initial
          cluster metadata. This does not have to be the full node list.
          It just needs to have at least one broker that will respond to a
          Metadata API Request. Default port is 9092. If no servers are
          specified, will default to localhost:9092.
class_docstr_tok: ["A","class","to","manage","kafka","cluster","metadata",".","This","class","does","not","perform","any","IO",".","It","simply","updates","internal","state","given","API","responses","(","MetadataResponse",",","GroupCoordinatorResponse",")",".","Keyword","Arguments",":","retry_backoff_ms","(","int",")",":","Milliseconds","to","backoff","when","retrying","on","errors",".","Default",":","100.","metadata_max_age_ms","(","int",")",":","The","period","of","time","in","milliseconds","after","which","we","force","a","refresh","of","metadata","even","if","we","have","n't","seen","any","partition","leadership","changes","to","proactively","discover","any","new","brokers","or","partitions",".","Default",":","300000","bootstrap_servers",":","'host","[",":","port","]","'","string","(","or","list","of","'host","[",":","port","]","'","strings",")","that","the","client","should","contact","to","bootstrap","initial","cluster","metadata",".","This","does","not","have","to","be","the","full","node","list",".","It","just","needs","to","have","at","least","one","broker","that","will","respond","to","a","Metadata","API","Request",".","Default","port","is","9092",".","If","no","servers","are","specified",",","will","default","to","localhost:9092","."]
docstr: Update cluster state given a MetadataResponse.
  
  Arguments:
      metadata (MetadataResponse): broker response to a metadata request
  
  Returns: None
docstr_tok: ["Update","cluster","state","given","a","MetadataResponse",".","Arguments",":","metadata","(","MetadataResponse",")",":","broker","response","to","a","metadata","request","Returns",":","None"]
code: def update_metadata(self, metadata):
      """Update cluster state given a MetadataResponse.
  
      Arguments:
          metadata (MetadataResponse): broker response to a metadata request
  
      Returns: None
      """
      # In the common case where we ask for a single topic and get back an
      # error, we should fail the future
      if len(metadata.topics) == 1 and metadata.topics[0][0] != 0:
          error_code, topic = metadata.topics[0][:2]
          error = Errors.for_code(error_code)(topic)
          return self.failed_update(error)
  
      if not metadata.brokers:
          log.warning(
              "No broker metadata found in MetadataResponse -- ignoring."
          )
          return self.failed_update(
              Errors.MetadataEmptyBrokerList(metadata)
          )
  
      _new_brokers = {}
      for broker in metadata.brokers:
          if metadata.API_VERSION == 0:
              node_id, host, port = broker
              rack = None
          else:
              node_id, host, port, rack = broker
          _new_brokers.update(
              {node_id: BrokerMetadata(node_id, host, port, rack)}
          )
  
      if metadata.API_VERSION == 0:
          _new_controller = None
      else:
          _new_controller = _new_brokers.get(metadata.controller_id)
  
      _new_partitions = {}
      _new_broker_partitions = collections.defaultdict(set)
      _new_unauthorized_topics = set()
      _new_internal_topics = set()
  
      for topic_data in metadata.topics:
          if metadata.API_VERSION == 0:
              error_code, topic, partitions = topic_data
              is_internal = False
          else:
              error_code, topic, is_internal, partitions = topic_data
          if is_internal:
              _new_internal_topics.add(topic)
          error_type = Errors.for_code(error_code)
          if error_type is Errors.NoError:
              _new_partitions[topic] = {}
              for (
                  p_error,
                  partition,
                  leader,
                  replicas,
                  isr,
              ) in partitions:
                  _new_partitions[topic][partition] = PartitionMetadata(
                      topic=topic,
                      partition=partition,
                      leader=leader,
                      replicas=replicas,
                      isr=isr,
                      error=p_error,
                  )
                  if leader != -1:
                      _new_broker_partitions[leader].add(
                          TopicPartition(topic, partition)
                      )
  
          # Specific topic errors can be ignored if this is a full metadata fetch
          elif self.need_all_topic_metadata:
              continue
  
          elif error_type is Errors.LeaderNotAvailableError:
              log.warning(
                  "Topic %s is not available during auto-create"
                  " initialization",
                  topic,
              )
          elif error_type is Errors.UnknownTopicOrPartitionError:
              log.error("Topic %s not found in cluster metadata", topic)
          elif error_type is Errors.TopicAuthorizationFailedError:
              log.error(
                  "Topic %s is not authorized for this client", topic
              )
              _new_unauthorized_topics.add(topic)
          elif error_type is Errors.InvalidTopicError:
              log.error("'%s' is not a valid topic name", topic)
          else:
              log.error(
                  "Error fetching metadata for topic %s: %s",
                  topic,
                  error_type,
              )
  
      with self._lock:
          self._brokers = _new_brokers
          self.controller = _new_controller
          self._partitions = _new_partitions
          self._broker_partitions = _new_broker_partitions
          self.unauthorized_topics = _new_unauthorized_topics
          self.internal_topics = _new_internal_topics
          f = None
          if self._future:
              f = self._future
          self._future = None
          self._need_update = False
  
      now = time.time() * 1000
      self._last_refresh_ms = now
      self._last_successful_refresh_ms = now
  
      if f:
          f.success(self)
      log.debug("Updated cluster metadata to %s", self)
  
      for listener in self._listeners:
          listener(self)
  
      if self.need_all_topic_metadata:
          # the listener may change the interested topics,
          # which could cause another metadata refresh.
          # If we have already fetched all topics, however,
          # another fetch should be unnecessary.
          self._need_update = False
  
imports: from __future__ import absolute_import
  import collections
  import copy
  import logging
  import threading
  import time
  from kafka.vendor import six
  from kafka import errors
  from kafka.conn import collect_hosts
  from kafka.future import Future
  from kafka.structs import BrokerMetadata, PartitionMetadata, TopicPartition
label_desc: Use image node_id 16 for calling the ClusterMetadata obj's underlying member method code with example usage: obj.update_metadata(metadata) and returns: self, self