def update_metadata(self, metadata):
        """Update cluster state given a MetadataResponse.

        Arguments:
            metadata (MetadataResponse): broker response to a metadata request

        Returns: None
        """
        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)

        if metadata.API_VERSION < 2:
            _new_cluster_id = None
        else:
            _new_cluster_id = metadata.cluster_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 partition_data in partitions:
                    leader_epoch = -1
                    offline_replicas = []
                    if metadata.API_VERSION >= 7:
                        p_error, partition, leader, leader_epoch, replicas, isr, offline_replicas = partition_data
                    elif metadata.API_VERSION >= 5:
                        p_error, partition, leader, replicas, isr, offline_replicas = partition_data
                    else:
                        p_error, partition, leader, replicas, isr = partition_data

                    _new_partitions[topic][partition] = PartitionMetadata(
                        topic=topic, partition=partition,
                        leader=leader, leader_epoch=leader_epoch,
                        replicas=replicas, isr=isr, offline_replicas=offline_replicas,
                        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.cluster_id = _new_cluster_id
            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:
            # 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] != Errors.NoError.errno:
                error_code, topic = metadata.topics[0][:2]
                error = Errors.for_code(error_code)(topic)
                f.failure(error)
            else:
                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
Update cluster state given a MetadataResponse.

        Arguments:
            metadata (MetadataResponse): broker response to a metadata request

        Returns: None
        
kafka/cluster.py