diff options
Diffstat (limited to 'kafka/protocol/legacy.py')
| -rw-r--r-- | kafka/protocol/legacy.py | 474 |
1 files changed, 0 insertions, 474 deletions
diff --git a/kafka/protocol/legacy.py b/kafka/protocol/legacy.py deleted file mode 100644 index 2e8f5bc..0000000 --- a/kafka/protocol/legacy.py +++ /dev/null @@ -1,474 +0,0 @@ -from __future__ import absolute_import - -import logging -import struct - -from kafka.vendor import six # pylint: disable=import-error - -import kafka.protocol.commit -import kafka.protocol.fetch -import kafka.protocol.message -import kafka.protocol.metadata -import kafka.protocol.offset -import kafka.protocol.produce -import kafka.structs - -from kafka.codec import gzip_encode, snappy_encode -from kafka.errors import ProtocolError, UnsupportedCodecError -from kafka.util import ( - crc32, read_short_string, relative_unpack, - write_int_string, group_by_topic_and_partition) -from kafka.protocol.message import MessageSet - - -log = logging.getLogger(__name__) - -ATTRIBUTE_CODEC_MASK = 0x03 -CODEC_NONE = 0x00 -CODEC_GZIP = 0x01 -CODEC_SNAPPY = 0x02 -ALL_CODECS = (CODEC_NONE, CODEC_GZIP, CODEC_SNAPPY) - - -class KafkaProtocol(object): - """ - Class to encapsulate all of the protocol encoding/decoding. - This class does not have any state associated with it, it is purely - for organization. - """ - PRODUCE_KEY = 0 - FETCH_KEY = 1 - OFFSET_KEY = 2 - METADATA_KEY = 3 - OFFSET_COMMIT_KEY = 8 - OFFSET_FETCH_KEY = 9 - CONSUMER_METADATA_KEY = 10 - - ################### - # Private API # - ################### - - @classmethod - def _encode_message_header(cls, client_id, correlation_id, request_key, - version=0): - """ - Encode the common request envelope - """ - return struct.pack('>hhih%ds' % len(client_id), - request_key, # ApiKey - version, # ApiVersion - correlation_id, # CorrelationId - len(client_id), # ClientId size - client_id) # ClientId - - @classmethod - def _encode_message_set(cls, messages): - """ - Encode a MessageSet. Unlike other arrays in the protocol, - MessageSets are not length-prefixed - - Format - ====== - MessageSet => [Offset MessageSize Message] - Offset => int64 - MessageSize => int32 - """ - message_set = [] - for message in messages: - encoded_message = KafkaProtocol._encode_message(message) - message_set.append(struct.pack('>qi%ds' % len(encoded_message), 0, - len(encoded_message), - encoded_message)) - return b''.join(message_set) - - @classmethod - def _encode_message(cls, message): - """ - Encode a single message. - - The magic number of a message is a format version number. - The only supported magic number right now is zero - - Format - ====== - Message => Crc MagicByte Attributes Key Value - Crc => int32 - MagicByte => int8 - Attributes => int8 - Key => bytes - Value => bytes - """ - if message.magic == 0: - msg = b''.join([ - struct.pack('>BB', message.magic, message.attributes), - write_int_string(message.key), - write_int_string(message.value) - ]) - crc = crc32(msg) - msg = struct.pack('>i%ds' % len(msg), crc, msg) - else: - raise ProtocolError("Unexpected magic number: %d" % message.magic) - return msg - - ################## - # Public API # - ################## - - @classmethod - def encode_produce_request(cls, payloads=(), acks=1, timeout=1000): - """ - Encode a ProduceRequest struct - - Arguments: - payloads: list of ProduceRequestPayload - acks: How "acky" you want the request to be - 1: written to disk by the leader - 0: immediate response - -1: waits for all replicas to be in sync - timeout: Maximum time (in ms) the server will wait for replica acks. - This is _not_ a socket timeout - - Returns: ProduceRequest - """ - if acks not in (1, 0, -1): - raise ValueError('ProduceRequest acks (%s) must be 1, 0, -1' % acks) - - topics = [] - for topic, topic_payloads in group_by_topic_and_partition(payloads).items(): - topic_msgs = [] - for partition, payload in topic_payloads.items(): - partition_msgs = [] - for msg in payload.messages: - m = kafka.protocol.message.Message( - msg.value, key=msg.key, - magic=msg.magic, attributes=msg.attributes - ) - partition_msgs.append((0, m.encode())) - topic_msgs.append((partition, MessageSet.encode(partition_msgs, prepend_size=False))) - topics.append((topic, topic_msgs)) - - - return kafka.protocol.produce.ProduceRequest[0]( - required_acks=acks, - timeout=timeout, - topics=topics - ) - - @classmethod - def decode_produce_response(cls, response): - """ - Decode ProduceResponse to ProduceResponsePayload - - Arguments: - response: ProduceResponse - - Return: list of ProduceResponsePayload - """ - return [ - kafka.structs.ProduceResponsePayload(topic, partition, error, offset) - for topic, partitions in response.topics - for partition, error, offset in partitions - ] - - @classmethod - def encode_fetch_request(cls, payloads=(), max_wait_time=100, min_bytes=4096): - """ - Encodes a FetchRequest struct - - Arguments: - payloads: list of FetchRequestPayload - max_wait_time (int, optional): ms to block waiting for min_bytes - data. Defaults to 100. - min_bytes (int, optional): minimum bytes required to return before - max_wait_time. Defaults to 4096. - - Return: FetchRequest - """ - return kafka.protocol.fetch.FetchRequest[0]( - replica_id=-1, - max_wait_time=max_wait_time, - min_bytes=min_bytes, - topics=[( - topic, - [( - partition, - payload.offset, - payload.max_bytes) - for partition, payload in topic_payloads.items()]) - for topic, topic_payloads in group_by_topic_and_partition(payloads).items()]) - - @classmethod - def decode_fetch_response(cls, response): - """ - Decode FetchResponse struct to FetchResponsePayloads - - Arguments: - response: FetchResponse - """ - return [ - kafka.structs.FetchResponsePayload( - topic, partition, error, highwater_offset, [ - offset_and_msg - for offset_and_msg in cls.decode_message_set(messages)]) - for topic, partitions in response.topics - for partition, error, highwater_offset, messages in partitions - ] - - @classmethod - def decode_message_set(cls, raw_data): - messages = MessageSet.decode(raw_data, bytes_to_read=len(raw_data)) - for offset, _, message in messages: - if isinstance(message, kafka.protocol.message.Message) and message.is_compressed(): - inner_messages = message.decompress() - for (inner_offset, _msg_size, inner_msg) in inner_messages: - yield kafka.structs.OffsetAndMessage(inner_offset, inner_msg) - else: - yield kafka.structs.OffsetAndMessage(offset, message) - - @classmethod - def encode_offset_request(cls, payloads=()): - return kafka.protocol.offset.OffsetRequest[0]( - replica_id=-1, - topics=[( - topic, - [( - partition, - payload.time, - payload.max_offsets) - for partition, payload in six.iteritems(topic_payloads)]) - for topic, topic_payloads in six.iteritems(group_by_topic_and_partition(payloads))]) - - @classmethod - def decode_offset_response(cls, response): - """ - Decode OffsetResponse into OffsetResponsePayloads - - Arguments: - response: OffsetResponse - - Returns: list of OffsetResponsePayloads - """ - return [ - kafka.structs.OffsetResponsePayload(topic, partition, error, tuple(offsets)) - for topic, partitions in response.topics - for partition, error, offsets in partitions - ] - - @classmethod - def encode_list_offset_request(cls, payloads=()): - return kafka.protocol.offset.OffsetRequest[1]( - replica_id=-1, - topics=[( - topic, - [( - partition, - payload.time) - for partition, payload in six.iteritems(topic_payloads)]) - for topic, topic_payloads in six.iteritems(group_by_topic_and_partition(payloads))]) - - @classmethod - def decode_list_offset_response(cls, response): - """ - Decode OffsetResponse_v2 into ListOffsetResponsePayloads - - Arguments: - response: OffsetResponse_v2 - - Returns: list of ListOffsetResponsePayloads - """ - return [ - kafka.structs.ListOffsetResponsePayload(topic, partition, error, timestamp, offset) - for topic, partitions in response.topics - for partition, error, timestamp, offset in partitions - ] - - - @classmethod - def encode_metadata_request(cls, topics=(), payloads=None): - """ - Encode a MetadataRequest - - Arguments: - topics: list of strings - """ - if payloads is not None: - topics = payloads - - return kafka.protocol.metadata.MetadataRequest[0](topics) - - @classmethod - def decode_metadata_response(cls, response): - return response - - @classmethod - def encode_consumer_metadata_request(cls, client_id, correlation_id, payloads): - """ - Encode a ConsumerMetadataRequest - - Arguments: - client_id: string - correlation_id: int - payloads: string (consumer group) - """ - message = [] - message.append(cls._encode_message_header(client_id, correlation_id, - KafkaProtocol.CONSUMER_METADATA_KEY)) - message.append(struct.pack('>h%ds' % len(payloads), len(payloads), payloads)) - - msg = b''.join(message) - return write_int_string(msg) - - @classmethod - def decode_consumer_metadata_response(cls, data): - """ - Decode bytes to a kafka.structs.ConsumerMetadataResponse - - Arguments: - data: bytes to decode - """ - ((correlation_id, error, nodeId), cur) = relative_unpack('>ihi', data, 0) - (host, cur) = read_short_string(data, cur) - ((port,), cur) = relative_unpack('>i', data, cur) - - return kafka.structs.ConsumerMetadataResponse(error, nodeId, host, port) - - @classmethod - def encode_offset_commit_request(cls, group, payloads): - """ - Encode an OffsetCommitRequest struct - - Arguments: - group: string, the consumer group you are committing offsets for - payloads: list of OffsetCommitRequestPayload - """ - return kafka.protocol.commit.OffsetCommitRequest[0]( - consumer_group=group, - topics=[( - topic, - [( - partition, - payload.offset, - payload.metadata) - for partition, payload in six.iteritems(topic_payloads)]) - for topic, topic_payloads in six.iteritems(group_by_topic_and_partition(payloads))]) - - @classmethod - def decode_offset_commit_response(cls, response): - """ - Decode OffsetCommitResponse to an OffsetCommitResponsePayload - - Arguments: - response: OffsetCommitResponse - """ - return [ - kafka.structs.OffsetCommitResponsePayload(topic, partition, error) - for topic, partitions in response.topics - for partition, error in partitions - ] - - @classmethod - def encode_offset_fetch_request(cls, group, payloads, from_kafka=False): - """ - Encode an OffsetFetchRequest struct. The request is encoded using - version 0 if from_kafka is false, indicating a request for Zookeeper - offsets. It is encoded using version 1 otherwise, indicating a request - for Kafka offsets. - - Arguments: - group: string, the consumer group you are fetching offsets for - payloads: list of OffsetFetchRequestPayload - from_kafka: bool, default False, set True for Kafka-committed offsets - """ - version = 1 if from_kafka else 0 - return kafka.protocol.commit.OffsetFetchRequest[version]( - consumer_group=group, - topics=[( - topic, - list(topic_payloads.keys())) - for topic, topic_payloads in six.iteritems(group_by_topic_and_partition(payloads))]) - - @classmethod - def decode_offset_fetch_response(cls, response): - """ - Decode OffsetFetchResponse to OffsetFetchResponsePayloads - - Arguments: - response: OffsetFetchResponse - """ - return [ - kafka.structs.OffsetFetchResponsePayload( - topic, partition, offset, metadata, error - ) - for topic, partitions in response.topics - for partition, offset, metadata, error in partitions - ] - - -def create_message(payload, key=None): - """ - Construct a Message - - Arguments: - payload: bytes, the payload to send to Kafka - key: bytes, a key used for partition routing (optional) - - """ - return kafka.structs.Message(0, 0, key, payload) - - -def create_gzip_message(payloads, key=None, compresslevel=None): - """ - Construct a Gzipped Message containing multiple Messages - - The given payloads will be encoded, compressed, and sent as a single atomic - message to Kafka. - - Arguments: - payloads: list(bytes), a list of payload to send be sent to Kafka - key: bytes, a key used for partition routing (optional) - - """ - message_set = KafkaProtocol._encode_message_set( - [create_message(payload, pl_key) for payload, pl_key in payloads]) - - gzipped = gzip_encode(message_set, compresslevel=compresslevel) - codec = ATTRIBUTE_CODEC_MASK & CODEC_GZIP - - return kafka.structs.Message(0, 0x00 | codec, key, gzipped) - - -def create_snappy_message(payloads, key=None): - """ - Construct a Snappy Message containing multiple Messages - - The given payloads will be encoded, compressed, and sent as a single atomic - message to Kafka. - - Arguments: - payloads: list(bytes), a list of payload to send be sent to Kafka - key: bytes, a key used for partition routing (optional) - - """ - message_set = KafkaProtocol._encode_message_set( - [create_message(payload, pl_key) for payload, pl_key in payloads]) - - snapped = snappy_encode(message_set) - codec = ATTRIBUTE_CODEC_MASK & CODEC_SNAPPY - - return kafka.structs.Message(0, 0x00 | codec, key, snapped) - - -def create_message_set(messages, codec=CODEC_NONE, key=None, compresslevel=None): - """Create a message set using the given codec. - - If codec is CODEC_NONE, return a list of raw Kafka messages. Otherwise, - return a list containing a single codec-encoded message. - """ - if codec == CODEC_NONE: - return [create_message(m, k) for m, k in messages] - elif codec == CODEC_GZIP: - return [create_gzip_message(messages, key, compresslevel)] - elif codec == CODEC_SNAPPY: - return [create_snappy_message(messages, key)] - else: - raise UnsupportedCodecError("Codec 0x%02x unsupported" % (codec,)) |