# 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/util.py
# NO_EXTENSIONS = bool(os.environ.get("AIOKAFKA_NO_EXTENSIONS"))


import struct
import time
import kafka.codec as codecs
from binascii import crc32
from aiokafka.errors import CorruptRecordException, UnsupportedCodecError
from aiokafka.util import NO_EXTENSIONS
from kafka.codec import (
    gzip_encode, snappy_encode, lz4_encode, lz4_encode_old_kafka,
    gzip_decode, snappy_decode, lz4_decode, lz4_decode_old_kafka
)
        from ._crecords import (
            LegacyRecordBatchBuilder as _LegacyRecordBatchBuilderCython,
            LegacyRecordMetadata as _LegacyRecordMetadataCython,
            LegacyRecordBatch as _LegacyRecordBatchCython,
            LegacyRecord as _LegacyRecordCython
        )
    RECORD_OVERHEAD_V1 = struct.calcsize(
        ">I"  # CRC
        "b"   # magic
        "b"   # attributes
        "q"   # timestamp
        "i"   # Key length
        "i"   # Value length
    )
    RECORD_OVERHEAD = {
        0: RECORD_OVERHEAD_V0,
        1: RECORD_OVERHEAD_V1,
    }

    KEY_OFFSET_V0 = HEADER_STRUCT_V0.size
    KEY_OFFSET_V1 = HEADER_STRUCT_V1.size
    KEY_LENGTH = VALUE_LENGTH = struct.calcsize(">i")  # Bytes length is Int32

    CODEC_MASK = 0x07
    CODEC_GZIP = 0x01
    CODEC_SNAPPY = 0x02
    CODEC_LZ4 = 0x03
    TIMESTAMP_TYPE_MASK = 0x08

    LOG_APPEND_TIME = 1
    CREATE_TIME = 0

    def _assert_has_codec(self, compression_type):
        if compression_type == self.CODEC_GZIP:
            checker, name = codecs.has_gzip, "gzip"
        elif compression_type == self.CODEC_SNAPPY:

# 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/util.py
# NO_EXTENSIONS = bool(os.environ.get("AIOKAFKA_NO_EXTENSIONS"))


# Path: aiokafka/record/legacy_records.py
import struct
import time
import kafka.codec as codecs
from binascii import crc32
from aiokafka.errors import CorruptRecordException, UnsupportedCodecError
from aiokafka.util import NO_EXTENSIONS
from kafka.codec import (
    gzip_encode, snappy_encode, lz4_encode, lz4_encode_old_kafka,
    gzip_decode, snappy_decode, lz4_decode, lz4_decode_old_kafka
)
        from ._crecords import (
            LegacyRecordBatchBuilder as _LegacyRecordBatchBuilderCython,
            LegacyRecordMetadata as _LegacyRecordMetadataCython,
            LegacyRecordBatch as _LegacyRecordBatchCython,
            LegacyRecord as _LegacyRecordCython
        )

    RECORD_OVERHEAD_V1 = struct.calcsize(
        ">I"  # CRC
        "b"   # magic
        "b"   # attributes
        "q"   # timestamp
        "i"   # Key length
        "i"   # Value length
    )
    RECORD_OVERHEAD = {
        0: RECORD_OVERHEAD_V0,
        1: RECORD_OVERHEAD_V1,
    }

    KEY_OFFSET_V0 = HEADER_STRUCT_V0.size
    KEY_OFFSET_V1 = HEADER_STRUCT_V1.size
    KEY_LENGTH = VALUE_LENGTH = struct.calcsize(">i")  # Bytes length is Int32

    CODEC_MASK = 0x07
    CODEC_GZIP = 0x01
    CODEC_SNAPPY = 0x02
    CODEC_LZ4 = 0x03
    TIMESTAMP_TYPE_MASK = 0x08

    LOG_APPEND_TIME = 1
    CREATE_TIME = 0

    def _assert_has_codec(self, compression_type):
        if compression_type == self.CODEC_GZIP:
            checker, name = codecs.has_gzip, "gzip"
        elif compression_type == self.CODEC_SNAPPY:

            checker, name = codecs.has_snappy, "snappy"