# 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
        a new one was elected for the same purpose (For example group
        coordinator for group X).
        """
        if purpose in self._coordinator_by_key:
            old_id = self._coordinator_by_key.pop(purpose)
            del self._coordinators[old_id]

        self._coordinators[node_id] = BrokerMetadata(node_id, host, port, rack)
        self._coordinator_by_key[purpose] = node_id

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

        _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({

# 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

        a new one was elected for the same purpose (For example group
        coordinator for group X).
        """
        if purpose in self._coordinator_by_key:
            old_id = self._coordinator_by_key.pop(purpose)
            del self._coordinators[old_id]

        self._coordinators[node_id] = BrokerMetadata(node_id, host, port, rack)
        self._coordinator_by_key[purpose] = node_id

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

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