def group_by_topic_and_partition(tuples):
    out = collections.defaultdict(dict)
    for t in tuples:
        out[t.topic][t.partition] = t
    return out def read_int_string(data, cur):
    if len(data) < cur + 4:
        raise _buffer_underflow('long string length', data, cur, 4)

    (strlen,) = struct.unpack('>i', data[cur:cur + 4])
    if strlen == -1:
        return None, cur + 4

    cur += 4
    if len(data) < cur + strlen:
        raise _buffer_underflow('long string', data, cur, strlen)

    out = data[cur:cur + strlen]
    return out, cur + strlen def read_short_ascii(data, cur):
    b, cur = read_short_bytes(data, cur)
    return b.decode('ascii'), cur def read_short_bytes(data, cur):
    if len(data) < cur + 2:
        raise _buffer_underflow('short string length', data, cur, 2)

    (strlen,) = struct.unpack('>h', data[cur:cur + 2])
    if strlen == -1:
        return None, cur + 2

    cur += 2
    if len(data) < cur + strlen:
        raise _buffer_underflow('short string', data, cur, strlen)

    out = data[cur:cur + strlen]
    return out, cur + strlen def read_short_text(data, cur):
    b, cur = read_short_bytes(data, cur)
    return b.decode('utf-8'), cur def relative_unpack(fmt, data, cur):
    size = struct.calcsize(fmt)
    if len(data) < cur + size:
        raise _buffer_underflow(fmt, data, cur, size)

    out = struct.unpack(fmt, data[cur:cur + size])
    return out, cur + size def write_int_string(s):
    if s is None:
        return struct.pack('>i', -1)
    return struct.pack('>i', len(s)) + s def write_short_ascii(s):
    """
    Encode a Kafka short string which represents ASCII text.

    :param str s:
        Text string or ``None``. The string will be ASCII-encoded.

    :returns: length-prefixed `bytes`
    :raises:
        `struct.error` for strings longer than 32767 characters
    """
    if s is None:
        return _NULL_SHORT_STRING
    if not isinstance(s, str):
        raise TypeError('{!r} is not text'.format(s))
    return write_short_bytes(s.encode('ascii')) def write_short_bytes(b):
    """
    Encode a Kafka short string which contains arbitrary bytes. A short string
    is limited to 32767 bytes in length by the signed 16-bit length prefix.
    A length prefix of -1 indicates ``null``, represented as ``None`` in
    Python.

    :param bytes b:
        No more than 32767 bytes, or ``None`` for the null encoding.
    :return: length-prefixed `bytes`
    :raises:
        `struct.error` for strings longer than 32767 characters
    """
    if b is None:
        return _NULL_SHORT_STRING
    if not isinstance(b, bytes):
        raise TypeError('{!r} is not bytes'.format(b))
    elif len(b) > 32767:
        raise struct.error(len(b))
    else:
        return struct.pack('>h', len(b)) + b def write_short_text(s):
    """
    Encode a Kafka short string which represents Unicode text.

    :param str s:
        Text string or ``None``. The string will be UTF-8 encoded.

    :returns: length-prefixed `bytes`
    :raises:
        `struct.error` when the UTF-8 encoded form of the string exceeds
        32767 bytes.
    """
    if s is None:
        return _NULL_SHORT_STRING
    if not isinstance(s, str):
        raise TypeError('{!r} is not text'.format(s))
    return write_short_bytes(s.encode('utf-8')) def gzip_decode(payload):
    buffer = BytesIO(payload)
    handle = gzip.GzipFile(fileobj=buffer, mode='r')
    result = handle.read()
    handle.close()
    buffer.close()
    return result def gzip_encode(payload):
    buffer = BytesIO()
    handle = gzip.GzipFile(fileobj=buffer, mode="w")
    handle.write(payload)
    handle.close()
    return buffer.getvalue() def snappy_decode(payload):
    if not has_snappy():
        raise NotImplementedError("Snappy codec is not available")

    if payload.startswith(_XERIAL_HEADER):
        # TODO ? Should become a fileobj ?
        view = memoryview(payload)
        out = []
        length = len(payload)

        cursor = 16
        while cursor < length:
            block_size = struct.unpack_from('!i', view, cursor)[0]
            # Skip the block size
            cursor += 4
            end = cursor + block_size
            # XXX snappy requires a bytes-like object but doesn't accept
            # a memoryview, so we must copy.
            out.append(snappy.decompress(view[cursor:end].tobytes()))
            cursor = end

        # See https://atleastfornow.net/blog/not-all-bytes/
        return b''.join(out)
    else:
        return snappy.decompress(payload) def snappy_encode(payload, xerial_compatible=False,
                  xerial_blocksize=32 * 1024):
    """
    Compress the given data with the Snappy algorithm.

    :param bytes payload: Data to compress.
    :param bool xerial_compatible:
        If set then the stream is broken into length-prefixed blocks in
        a fashion compatible with the xerial snappy library.

        The format winds up being::

            +-------------+------------+--------------+------------+--------------+
            |   Header    | Block1_len | Block1 data  | BlockN len | BlockN data  |
            |-------------+------------+--------------+------------+--------------|
            |  16 bytes   |  BE int32  | snappy bytes |  BE int32  | snappy bytes |
            +-------------+------------+--------------+------------+--------------+

    :param int xerial_blocksize:
        Number of bytes per chunk to independently Snappy encode. 32k is the
        default in the xerial library.

    :returns: Compressed bytes.
    :rtype: :class:`bytes`
    """
    if not has_snappy():  # FIXME This should be static, not checked every call.
        raise NotImplementedError("Snappy codec is not available")

    if xerial_compatible:
        def _chunker():
            for i in range(0, len(payload), xerial_blocksize):
                yield payload[i:i+xerial_blocksize]

        out = BytesIO()
        out.write(_XERIAL_HEADER)

        for chunk in _chunker():
            block = snappy.compress(chunk)
            out.write(struct.pack('!i', len(block)))
            out.write(block)

        out.seek(0)
        return out.read()

    else:
        return snappy.compress(payload) OFFSET_EARLIEST = -2  # From the docs for OffsetRequest
OFFSET_LATEST = -1  # From the docs for OffsetRequest
OFFSET_NOT_COMMITTED = -1  # Returned by kafka when no offset is stored
OFFSET_COMMITTED = -101  # Used to avoid possible additions from the Kafka team
TIMESTAMP_INVALID = -1  # Used to specify that the broker should set timestamp
KAFKA_SUCCESS = 0  # An 'error' of 0 is used to indicate success
PRODUCER_ACK_NOT_REQUIRED = 0  # No ack is required
PRODUCER_ACK_LOCAL_WRITE = 1  # Send response only after it is written to log
PRODUCER_ACK_ALL_REPLICAS = -1  # Response after data written by all replicas
CODEC_NONE = 0x00
CODEC_GZIP = 0x01
CODEC_SNAPPY = 0x02
CODEC_LZ4 = 0x03
_ALL_CODECS = (CODEC_NONE, CODEC_GZIP, CODEC_SNAPPY)
class _JoinGroupRequestProtocol(object):
class _JoinGroupProtocolMetadata(object):
class _JoinGroupRequest(object):
class _JoinGroupResponseMember(object):
class _JoinGroupResponse(object):
class _SyncGroupRequestMember(object):
class _SyncGroupMemberAssignment(object):
class _SyncGroupRequest(object):
class _SyncGroupResponse(object):
class Message(namedtuple("Message", ["magic", "attributes", "key", "value"])):
class KafkaError(Exception):
class ClientError(KafkaError):
class RestartError(ClientError):
class RestopError(ClientError):
class DuplicateRequestError(KafkaError):
class BrokerResponseError(KafkaError):
class RetriableBrokerResponseError(BrokerResponseError):
class UnknownError(BrokerResponseError):
class OffsetOutOfRangeError(BrokerResponseError):
class CorruptMessage(RetriableBrokerResponseError):
class UnknownTopicOrPartitionError(RetriableBrokerResponseError):
class InvalidFetchRequestError(BrokerResponseError):
class LeaderNotAvailableError(RetriableBrokerResponseError):
class NotLeaderForPartitionError(RetriableBrokerResponseError):
class RequestTimedOutError(RetriableBrokerResponseError):
class BrokerNotAvailableError(BrokerResponseError):
class ReplicaNotAvailableError(BrokerResponseError):
class MessageSizeTooLargeError(BrokerResponseError):
class StaleControllerEpochError(BrokerResponseError):
class OffsetMetadataTooLargeError(BrokerResponseError):
class NetworkException(RetriableBrokerResponseError):
class CoordinatorLoadInProgress(RetriableBrokerResponseError):
class CoordinatorNotAvailable(RetriableBrokerResponseError):
class NotCoordinator(RetriableBrokerResponseError):
class InvalidTopic(BrokerResponseError):
class RecordListTooLarge(BrokerResponseError):
class NotEnoughReplicas(RetriableBrokerResponseError):
class NotEnoughReplicasAfterAppend(RetriableBrokerResponseError):
class InvalidRequiredAcks(BrokerResponseError):
class IllegalGeneration(BrokerResponseError):
class InconsistentGroupProtocol(BrokerResponseError):
class InvalidGroupId(BrokerResponseError):
class UnknownMemberId(BrokerResponseError):
class InvalidSessionTimeout(BrokerResponseError):
class RebalanceInProgress(BrokerResponseError):
class InvalidCommitOffsetSize(BrokerResponseError):
class TopicAuthorizationFailed(BrokerResponseError):
class GroupAuthorizationFailed(BrokerResponseError):
class ClusterAuthorizationFailed(BrokerResponseError):
class InvalidTimestamp(BrokerResponseError):
class UnsupportedSaslMechanism(BrokerResponseError):
class IllegalSaslState(BrokerResponseError):
class UnsupportedVersion(BrokerResponseError):
class TopicAlreadyExists(BrokerResponseError):
class InvalidPartitions(BrokerResponseError):
class InvalidReplicationFactor(BrokerResponseError):
class InvalidReplicaAssignment(BrokerResponseError):
class InvalidConfig(BrokerResponseError):
class NotController(RetriableBrokerResponseError):
class InvalidRequest(BrokerResponseError):
class UnsupportedForMessageFormat(BrokerResponseError):
class PolicyViolation(BrokerResponseError):
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(RetriableBrokerResponseError):
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(RetriableBrokerResponseError):
class InvalidFetchSessionEpoch(RetriableBrokerResponseError):
class ListenerNotFound(RetriableBrokerResponseError):
class KafkaUnavailableError(KafkaError):
class LeaderUnavailableError(KafkaError):
class PartitionUnavailableError(KafkaError):
class FailedPayloadsError(KafkaError):
class ConnectionError(KafkaError):
class BufferUnderflowError(KafkaError):
class ChecksumError(KafkaError):
class ConsumerFetchSizeTooSmall(KafkaError):
class ProtocolError(KafkaError):
class UnsupportedCodecError(KafkaError):
class CancelledError(KafkaError):
class InvalidConsumerGroupError(KafkaError):
class NoResponseError(KafkaError):
class OperationInProgress(KafkaError):
    def __repr__(self):
    def __str__(self):
    def raise_for_errno(cls, errno, *args):
    def __init__(self, request_sent=None, message=None):
    def __str__(self):
    def __init__(self, deferred=None):
def _pretty_errno(errno):
import logging
import struct
import zlib
from binascii import hexlify
from twisted.python.compat import nativeString
from ._util import (
    group_by_topic_and_partition, read_int_string, read_short_ascii,
    read_short_bytes, read_short_text, relative_unpack, write_int_string,
    write_short_ascii, write_short_bytes, write_short_text,
)
from .codec import gzip_decode, gzip_encode, snappy_decode, snappy_encode
from .common import (
    CODEC_GZIP, CODEC_NONE, CODEC_SNAPPY, BrokerMetadata, BufferUnderflowError,
    ChecksumError, ConsumerFetchSizeTooSmall, ConsumerMetadataResponse,
    FetchResponse, InvalidMessageError, Message, OffsetAndMessage,
    OffsetCommitResponse, OffsetFetchResponse, OffsetResponse,
    PartitionMetadata, ProduceResponse, ProtocolError, TopicMetadata,
    UnsupportedCodecError, _HeartbeatResponse, _JoinGroupProtocolMetadata,
    _JoinGroupResponse, _JoinGroupResponseMember, _LeaveGroupResponse,
    _SyncGroupMemberAssignment, _SyncGroupResponse,
)
                                             KafkaCodec.CONSUMER_METADATA_KEY)
        message += write_short_ascii(consumer_group)
        return message

    @classmethod
    def decode_consumermetadata_response(cls, data):
        """
        Decode bytes to a ConsumerMetadataResponse

        :param bytes data: bytes to decode
        """
        (correlation_id, error_code, node_id), cur = \
            relative_unpack('>ihi', data, 0)
        host, cur = read_short_ascii(data, cur)
        (port,), cur = relative_unpack('>i', data, cur)

        return ConsumerMetadataResponse(
            error_code, node_id, nativeString(host), port)

    @classmethod
    def encode_offset_commit_request(cls, client_id, correlation_id,
                                     group, group_generation_id, consumer_id,
                                     payloads):
        """
        Encode some OffsetCommitRequest structs (v1)

        :param bytes client_id: string
        :param int correlation_id: int
        :param str group: the consumer group to which you are committing offsets
        :param int group_generation_id: int32, generation ID of the group
        :param str consumer_id: string, Identifier for the consumer
        :param list payloads: list of :class:`OffsetCommitRequest`
        """
        assert consumer_id is not None
        grouped_payloads = group_by_topic_and_partition(payloads)

        message = cls._encode_message_header(
            client_id, correlation_id, KafkaCodec.OFFSET_COMMIT_KEY,
            api_version=1,
        )

        message += write_short_ascii(group)
        message += struct.pack('>i', group_generation_id)
        message += write_short_ascii(consumer_id)
        message += struct.pack('>i', len(grouped_payloads))

        for topic, topic_payloads in grouped_payloads.items():
            message += write_short_ascii(topic)
            message += struct.pack('>i', len(topic_payloads))

            for partition, payload in topic_payloads.items():
                message += struct.pack('>iqq', partition, payload.offset,
                                       payload.timestamp)
                message += write_short_bytes(payload.metadata)

        return message

    @classmethod
    def decode_offset_commit_response(cls, data):
        """
        Decode bytes to an OffsetCommitResponse

        :param bytes data: bytes to decode
        """
        ((correlation_id,), cur) = relative_unpack('>i', data, 0)
        ((num_topics,), cur) = relative_unpack('>i', data, cur)

        for _i in range(num_topics):
            (topic, cur) = read_short_ascii(data, cur)
            ((num_partitions,), cur) = relative_unpack('>i', data, cur)

            for _i in range(num_partitions):
                ((partition, error), cur) = relative_unpack('>ih', data, cur)
                yield OffsetCommitResponse(topic, partition, error)

    @classmethod
    def encode_offset_fetch_request(cls, client_id, correlation_id,
                                    group, payloads):
        """
        Encode some OffsetFetchRequest structs

        :param bytes client_id: string
        :param int correlation_id: int
        :param bytes group: string, the consumer group you are fetching offsets for
        :param list payloads: list of :class:`OffsetFetchRequest`
        """
        grouped_payloads = group_by_topic_and_partition(payloads)
        message = cls._encode_message_header(
            client_id, correlation_id, KafkaCodec.OFFSET_FETCH_KEY,
            api_version=1)

        message += write_short_ascii(group)
        message += struct.pack('>i', len(grouped_payloads))

        for topic, topic_payloads in grouped_payloads.items():
            message += write_short_ascii(topic)
            message += struct.pack('>i', len(topic_payloads))

            for partition in topic_payloads:
                message += struct.pack('>i', partition)

        return message

    @classmethod
    def decode_offset_fetch_response(cls, data):
        """
        Decode bytes to an OffsetFetchResponse

        :param bytes data: bytes to decode
        """

        ((correlation_id,), cur) = relative_unpack('>i', data, 0)
        ((num_topics,), cur) = relative_unpack('>i', data, cur)

        for _i in range(num_topics):
            (topic, cur) = read_short_ascii(data, cur)
            ((num_partitions,), cur) = relative_unpack('>i', data, cur)

            for _i in range(num_partitions):
                ((partition, offset), cur) = relative_unpack('>iq', data, cur)

                (metadata, cur) = read_short_bytes(data, cur)