# Path: aiokafka/structs.py
# class RecordMetadata(NamedTuple):
# class ConsumerRecord(Generic[KT, VT]):
# class OffsetAndTimestamp(NamedTuple):
# KT = TypeVar("KT")
# VT = TypeVar("VT")
# 
# Path: aiokafka/errors.py
# class CoordinatorNotAvailableError(GroupCoordinatorNotAvailableError):
# class NotCoordinatorError(NotCoordinatorForGroupError):
# class CoordinatorLoadInProgressError(GroupLoadInProgressError):
# class ConsumerStoppedError(Exception):
# class IllegalOperation(Exception):
# class NoOffsetForPartitionError(KafkaError):
# class RecordTooLargeError(KafkaError):
# class ProducerClosed(KafkaError):
# class ProducerFenced(KafkaError):
# class OutOfOrderSequenceNumber(BrokerResponseError):
# class DuplicateSequenceNumber(BrokerResponseError):
# class InvalidProducerEpoch(BrokerResponseError):
# class InvalidTxnState(BrokerResponseError):
# class InvalidProducerIdMapping(BrokerResponseError):
# class InvalidTransactionTimeout(BrokerResponseError):
# class ConcurrentTransactions(BrokerResponseError):
# class TransactionCoordinatorFenced(BrokerResponseError):
# class TransactionalIdAuthorizationFailed(BrokerResponseError):
# class SecurityDisabled(BrokerResponseError):
# class OperationNotAttempted(BrokerResponseError):
# class KafkaStorageError(BrokerResponseError):
# class LogDirNotFound(BrokerResponseError):
# class SaslAuthenticationFailed(BrokerResponseError):
# class UnknownProducerId(BrokerResponseError):
# class ReassignmentInProgress(BrokerResponseError):
# class DelegationTokenAuthDisabled(BrokerResponseError):
# class DelegationTokenNotFound(BrokerResponseError):
# class DelegationTokenOwnerMismatch(BrokerResponseError):
# class DelegationTokenRequestNotAllowed(BrokerResponseError):
# class DelegationTokenAuthorizationFailed(BrokerResponseError):
# class DelegationTokenExpired(BrokerResponseError):
# class InvalidPrincipalType(BrokerResponseError):
# class NonEmptyGroup(BrokerResponseError):
# class GroupIdNotFound(BrokerResponseError):
# class FetchSessionIdNotFound(BrokerResponseError):
# class InvalidFetchSessionEpoch(BrokerResponseError):
# class ListenerNotFound(BrokerResponseError):
#     def __init__(
#         self,
#         msg="There is a newer producer using the same transactional_id or"
#             "transaction timeout occurred (check that processing time is "
#             "below transaction_timeout_ms)"
#     ):
# def _iter_broker_errors():
# def for_code(error_code):


import collections
import logging
import time
from kafka.cluster import ClusterMetadata as BaseClusterMetadata
from aiokafka.structs import BrokerMetadata, PartitionMetadata, TopicPartition
from aiokafka import errors as Errors
        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))

            elif error_type is Errors.LeaderNotAvailableError:
                log.warning("Topic %s is not available during auto-create"
                            " initialization", topic)

# Path: aiokafka/structs.py
# class RecordMetadata(NamedTuple):
# class ConsumerRecord(Generic[KT, VT]):
# class OffsetAndTimestamp(NamedTuple):
# KT = TypeVar("KT")
# VT = TypeVar("VT")
# 
# Path: aiokafka/errors.py
# class CoordinatorNotAvailableError(GroupCoordinatorNotAvailableError):
# class NotCoordinatorError(NotCoordinatorForGroupError):
# class CoordinatorLoadInProgressError(GroupLoadInProgressError):
# class ConsumerStoppedError(Exception):
# class IllegalOperation(Exception):
# class NoOffsetForPartitionError(KafkaError):
# class RecordTooLargeError(KafkaError):
# class ProducerClosed(KafkaError):
# class ProducerFenced(KafkaError):
# class OutOfOrderSequenceNumber(BrokerResponseError):
# class DuplicateSequenceNumber(BrokerResponseError):
# class InvalidProducerEpoch(BrokerResponseError):
# class InvalidTxnState(BrokerResponseError):
# class InvalidProducerIdMapping(BrokerResponseError):
# class InvalidTransactionTimeout(BrokerResponseError):
# class ConcurrentTransactions(BrokerResponseError):
# class TransactionCoordinatorFenced(BrokerResponseError):
# class TransactionalIdAuthorizationFailed(BrokerResponseError):
# class SecurityDisabled(BrokerResponseError):
# class OperationNotAttempted(BrokerResponseError):
# class KafkaStorageError(BrokerResponseError):
# class LogDirNotFound(BrokerResponseError):
# class SaslAuthenticationFailed(BrokerResponseError):
# class UnknownProducerId(BrokerResponseError):
# class ReassignmentInProgress(BrokerResponseError):
# class DelegationTokenAuthDisabled(BrokerResponseError):
# class DelegationTokenNotFound(BrokerResponseError):
# class DelegationTokenOwnerMismatch(BrokerResponseError):
# class DelegationTokenRequestNotAllowed(BrokerResponseError):
# class DelegationTokenAuthorizationFailed(BrokerResponseError):
# class DelegationTokenExpired(BrokerResponseError):
# class InvalidPrincipalType(BrokerResponseError):
# class NonEmptyGroup(BrokerResponseError):
# class GroupIdNotFound(BrokerResponseError):
# class FetchSessionIdNotFound(BrokerResponseError):
# class InvalidFetchSessionEpoch(BrokerResponseError):
# class ListenerNotFound(BrokerResponseError):
#     def __init__(
#         self,
#         msg="There is a newer producer using the same transactional_id or"
#             "transaction timeout occurred (check that processing time is "
#             "below transaction_timeout_ms)"
#     ):
# def _iter_broker_errors():
# def for_code(error_code):


# Path: aiokafka/cluster.py
import collections
import logging
import time
from kafka.cluster import ClusterMetadata as BaseClusterMetadata
from aiokafka.structs import BrokerMetadata, PartitionMetadata, TopicPartition
from aiokafka import errors as Errors

        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))

            elif error_type is Errors.LeaderNotAvailableError:
                log.warning("Topic %s is not available during auto-create"
                            " initialization", topic)

            elif error_type is Errors.UnknownTopicOrPartitionError: