# Path: aiokafka/record/util.py
# def encode_varint_py(value, write):
# def size_of_varint_py(value):
# def decode_varint_py(buffer, pos=0):
# def calc_crc32c_py(memview):
# 
# 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 .util import decode_varint, encode_varint, calc_crc32c, size_of_varint
from aiokafka.errors import CorruptRecordException, UnsupportedCodecError
from aiokafka.util import NO_EXTENSIONS
from kafka.codec import (
    gzip_encode, snappy_encode, lz4_encode, zstd_encode,
    gzip_decode, snappy_decode, lz4_decode, zstd_decode
)
        from ._crecords import (
            DefaultRecordBatchBuilder as _DefaultRecordBatchBuilderCython,
            DefaultRecordMetadata as _DefaultRecordMetadataCython,
            DefaultRecordBatch as _DefaultRecordBatchCython,
            DefaultRecord as _DefaultRecordCython,
        )
    AFTER_LEN_OFFSET = struct.calcsize(">qi")

    CODEC_MASK = 0x07
    CODEC_NONE = 0x00
    CODEC_GZIP = 0x01
    CODEC_SNAPPY = 0x02
    CODEC_LZ4 = 0x03
    CODEC_ZSTD = 0x04
    TIMESTAMP_TYPE_MASK = 0x08
    TRANSACTIONAL_MASK = 0x10
    CONTROL_MASK = 0x20

    LOG_APPEND_TIME = 1
    CREATE_TIME = 0

    NO_PARTITION_LEADER_EPOCH = -1

    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"
        elif compression_type == self.CODEC_LZ4:
            checker, name = codecs.has_lz4, "lz4"
        elif compression_type == self.CODEC_ZSTD:
            checker, name = codecs.has_zstd, "zstd"
        else:
            raise UnsupportedCodecError(
                f"Unknown compression codec {compression_type:#04x}")
        if not checker():

# Path: aiokafka/record/util.py
# def encode_varint_py(value, write):
# def size_of_varint_py(value):
# def decode_varint_py(buffer, pos=0):
# def calc_crc32c_py(memview):
# 
# 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/default_records.py
import struct
import time
import kafka.codec as codecs
from .util import decode_varint, encode_varint, calc_crc32c, size_of_varint
from aiokafka.errors import CorruptRecordException, UnsupportedCodecError
from aiokafka.util import NO_EXTENSIONS
from kafka.codec import (
    gzip_encode, snappy_encode, lz4_encode, zstd_encode,
    gzip_decode, snappy_decode, lz4_decode, zstd_decode
)
        from ._crecords import (
            DefaultRecordBatchBuilder as _DefaultRecordBatchBuilderCython,
            DefaultRecordMetadata as _DefaultRecordMetadataCython,
            DefaultRecordBatch as _DefaultRecordBatchCython,
            DefaultRecord as _DefaultRecordCython,
        )

    AFTER_LEN_OFFSET = struct.calcsize(">qi")

    CODEC_MASK = 0x07
    CODEC_NONE = 0x00
    CODEC_GZIP = 0x01
    CODEC_SNAPPY = 0x02
    CODEC_LZ4 = 0x03
    CODEC_ZSTD = 0x04
    TIMESTAMP_TYPE_MASK = 0x08
    TRANSACTIONAL_MASK = 0x10
    CONTROL_MASK = 0x20

    LOG_APPEND_TIME = 1
    CREATE_TIME = 0

    NO_PARTITION_LEADER_EPOCH = -1

    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"
        elif compression_type == self.CODEC_LZ4:
            checker, name = codecs.has_lz4, "lz4"
        elif compression_type == self.CODEC_ZSTD:
            checker, name = codecs.has_zstd, "zstd"
        else:
            raise UnsupportedCodecError(
                f"Unknown compression codec {compression_type:#04x}")
        if not checker():

            raise UnsupportedCodecError(