# 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/record/memory_records.py
# class _MemoryRecordsPy:
#     LENGTH_OFFSET = struct.calcsize(">q")
#     LOG_OVERHEAD = struct.calcsize(">qi")
#     MAGIC_OFFSET = struct.calcsize(">qii")
#     MIN_SLICE = LOG_OVERHEAD + LegacyRecordBatch.RECORD_OVERHEAD_V0
#     def __init__(self, bytes_data):
#     def size_in_bytes(self):
#     def _cache_next(self, len_offset=LENGTH_OFFSET, log_overhead=LOG_OVERHEAD):
#     def has_next(self):
#     def next_batch(self, _min_slice=MIN_SLICE,
#                    _magic_offset=MAGIC_OFFSET):
# 
# Path: aiokafka/record/control_record.py
# class ControlRecord:
# 
#     def __init__(self, version, type_):
#         self._version = version
#         self._type = type_
# 
#     @property
#     def version(self):
#         return self._version
# 
#     @property
#     def type_(self):
#         return self._type
# 
#     def __eq__(self, other):
#         if isinstance(other, ControlRecord):
#             return other._version == self._version and \
#                 other._type == self._type
#         return False
# 
#     @classmethod
#     def parse(cls, data: bytes, _schema=struct.Struct(">HH")):
#         version, type_ = _schema.unpack_from(data)
#         return cls(version, type_)
# 
#     def __repr__(self):
#         return "ControlRecord(version={}, type_={})".format(
#             self._version, self._type)
# 
# ABORT_MARKER = ControlRecord(0, 0)
# 
# Path: aiokafka/structs.py
# class RecordMetadata(NamedTuple):
# class ConsumerRecord(Generic[KT, VT]):
# class OffsetAndTimestamp(NamedTuple):
# KT = TypeVar("KT")
# VT = TypeVar("VT")
# 
# Path: aiokafka/util.py
# def create_future(loop: AbstractEventLoop = None) -> "asyncio.Future[T]":
#     if loop is None:
#         loop = get_running_loop()
#     return loop.create_future()
# 
# def create_task(coro: Awaitable[T]) -> "asyncio.Task[T]":
#     loop = get_running_loop()
#     return loop.create_task(coro)


import asyncio
import collections
import logging
import random
import time
import async_timeout
import aiokafka.errors as Errors
from itertools import chain
from kafka.protocol.offset import OffsetRequest
from kafka.protocol.fetch import FetchRequest
from aiokafka.errors import (
    ConsumerStoppedError, RecordTooLargeError, KafkaTimeoutError)
from aiokafka.record.memory_records import MemoryRecords
from aiokafka.record.control_record import ControlRecord, ABORT_MARKER
from aiokafka.structs import OffsetAndTimestamp, TopicPartition, ConsumerRecord
from aiokafka.util import create_future, create_task
class PartitionRecords:

    def __init__(
            self, tp, records, aborted_transactions, fetch_offset,
            key_deserializer, value_deserializer, check_crcs, isolation_level):
        self._tp = tp
        self._records = records
        self._aborted_transactions = sorted(
            aborted_transactions or [], key=lambda x: x[1])
        self._aborted_producers = set()
        self._key_deserializer = key_deserializer
        self._value_deserializer = value_deserializer
        self._check_crcs = check_crcs
        self._isolation_level = isolation_level

        # Even without consuming any records we may need to force position to
        # a next offset. In cases like aborted transactions, control batches,
        # empty compacted batches, etc.
        self.next_fetch_offset = fetch_offset

        self._records_iterator = self._unpack_records()

    def __iter__(self):
        return self

    def __next__(self):
        try:
            return next(self._records_iterator)
        except StopIteration:
            # Break reference cycle just in case

# 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/record/memory_records.py
# class _MemoryRecordsPy:
#     LENGTH_OFFSET = struct.calcsize(">q")
#     LOG_OVERHEAD = struct.calcsize(">qi")
#     MAGIC_OFFSET = struct.calcsize(">qii")
#     MIN_SLICE = LOG_OVERHEAD + LegacyRecordBatch.RECORD_OVERHEAD_V0
#     def __init__(self, bytes_data):
#     def size_in_bytes(self):
#     def _cache_next(self, len_offset=LENGTH_OFFSET, log_overhead=LOG_OVERHEAD):
#     def has_next(self):
#     def next_batch(self, _min_slice=MIN_SLICE,
#                    _magic_offset=MAGIC_OFFSET):
# 
# Path: aiokafka/record/control_record.py
# class ControlRecord:
# 
#     def __init__(self, version, type_):
#         self._version = version
#         self._type = type_
# 
#     @property
#     def version(self):
#         return self._version
# 
#     @property
#     def type_(self):
#         return self._type
# 
#     def __eq__(self, other):
#         if isinstance(other, ControlRecord):
#             return other._version == self._version and \
#                 other._type == self._type
#         return False
# 
#     @classmethod
#     def parse(cls, data: bytes, _schema=struct.Struct(">HH")):
#         version, type_ = _schema.unpack_from(data)
#         return cls(version, type_)
# 
#     def __repr__(self):
#         return "ControlRecord(version={}, type_={})".format(
#             self._version, self._type)
# 
# ABORT_MARKER = ControlRecord(0, 0)
# 
# Path: aiokafka/structs.py
# class RecordMetadata(NamedTuple):
# class ConsumerRecord(Generic[KT, VT]):
# class OffsetAndTimestamp(NamedTuple):
# KT = TypeVar("KT")
# VT = TypeVar("VT")
# 
# Path: aiokafka/util.py
# def create_future(loop: AbstractEventLoop = None) -> "asyncio.Future[T]":
#     if loop is None:
#         loop = get_running_loop()
#     return loop.create_future()
# 
# def create_task(coro: Awaitable[T]) -> "asyncio.Task[T]":
#     loop = get_running_loop()
#     return loop.create_task(coro)


# Path: aiokafka/consumer/fetcher.py
import asyncio
import collections
import logging
import random
import time
import async_timeout
import aiokafka.errors as Errors
from itertools import chain
from kafka.protocol.offset import OffsetRequest
from kafka.protocol.fetch import FetchRequest
from aiokafka.errors import (
    ConsumerStoppedError, RecordTooLargeError, KafkaTimeoutError)
from aiokafka.record.memory_records import MemoryRecords
from aiokafka.record.control_record import ControlRecord, ABORT_MARKER
from aiokafka.structs import OffsetAndTimestamp, TopicPartition, ConsumerRecord
from aiokafka.util import create_future, create_task

class PartitionRecords:

    def __init__(
            self, tp, records, aborted_transactions, fetch_offset,
            key_deserializer, value_deserializer, check_crcs, isolation_level):
        self._tp = tp
        self._records = records
        self._aborted_transactions = sorted(
            aborted_transactions or [], key=lambda x: x[1])
        self._aborted_producers = set()
        self._key_deserializer = key_deserializer
        self._value_deserializer = value_deserializer
        self._check_crcs = check_crcs
        self._isolation_level = isolation_level

        # Even without consuming any records we may need to force position to
        # a next offset. In cases like aborted transactions, control batches,
        # empty compacted batches, etc.
        self.next_fetch_offset = fetch_offset

        self._records_iterator = self._unpack_records()

    def __iter__(self):
        return self

    def __next__(self):
        try:
            return next(self._records_iterator)
        except StopIteration:
            # Break reference cycle just in case

            self._records_iterator = None