kafka.protocol package¶

The body of FETCH_REQUEST_V6 is the same as FETCH_REQUEST_V5. The version number is bumped up to indicate that the client supports KafkaStorageException. The KafkaStorageException will be translated to NotLeaderForPartitionException in the response if version <= 5

API_KEY = 1¶

API_VERSION = 6¶

RESPONSE_TYPE¶: alias of FetchResponse_v6

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.fetch.FetchResponse_v0(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 1¶

API_VERSION = 0¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.fetch.FetchResponse_v1(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 1¶

API_VERSION = 1¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.fetch.FetchResponse_v2(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 1¶

API_VERSION = 2¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.fetch.FetchResponse_v3(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 1¶

API_VERSION = 3¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.fetch.FetchResponse_v4(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 1¶

API_VERSION = 4¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.fetch.FetchResponse_v5(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 1¶

API_VERSION = 5¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.fetch.FetchResponse_v6(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

Same as FetchResponse_v5. The version number is bumped up to indicate that the client supports KafkaStorageException. The KafkaStorageException will be translated to NotLeaderForPartitionException in the response if version <= 5

API_KEY = 1¶

API_VERSION = 6¶

SCHEMA = <kafka.protocol.types.Schema object>¶

kafka.protocol.group module¶

class kafka.protocol.group.HeartbeatRequest_v0(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

API_KEY = 12¶

API_VERSION = 0¶

RESPONSE_TYPE¶: alias of HeartbeatResponse_v0

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.HeartbeatRequest_v1(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

API_KEY = 12¶

API_VERSION = 1¶

RESPONSE_TYPE¶: alias of HeartbeatResponse_v1

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.HeartbeatResponse_v0(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 12¶

API_VERSION = 0¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.HeartbeatResponse_v1(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 12¶

API_VERSION = 1¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.JoinGroupRequest_v0(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

API_KEY = 11¶

API_VERSION = 0¶

RESPONSE_TYPE¶: alias of JoinGroupResponse_v0

SCHEMA = <kafka.protocol.types.Schema object>¶

UNKNOWN_MEMBER_ID = ''¶

class kafka.protocol.group.JoinGroupRequest_v1(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

API_KEY = 11¶

API_VERSION = 1¶

RESPONSE_TYPE¶: alias of JoinGroupResponse_v1

SCHEMA = <kafka.protocol.types.Schema object>¶

UNKNOWN_MEMBER_ID = ''¶

class kafka.protocol.group.JoinGroupRequest_v2(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

API_KEY = 11¶

API_VERSION = 2¶

RESPONSE_TYPE¶: alias of JoinGroupResponse_v2

SCHEMA = <kafka.protocol.types.Schema object>¶

UNKNOWN_MEMBER_ID = ''¶

class kafka.protocol.group.JoinGroupResponse_v0(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 11¶

API_VERSION = 0¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.JoinGroupResponse_v1(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 11¶

API_VERSION = 1¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.JoinGroupResponse_v2(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 11¶

API_VERSION = 2¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.LeaveGroupRequest_v0(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

API_KEY = 13¶

API_VERSION = 0¶

RESPONSE_TYPE¶: alias of LeaveGroupResponse_v0

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.LeaveGroupRequest_v1(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

API_KEY = 13¶

API_VERSION = 1¶

RESPONSE_TYPE¶: alias of LeaveGroupResponse_v1

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.LeaveGroupResponse_v0(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 13¶

API_VERSION = 0¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.LeaveGroupResponse_v1(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 13¶

API_VERSION = 1¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.MemberAssignment(*args, **kwargs)[source]¶

Bases: kafka.protocol.struct.Struct

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.ProtocolMetadata(*args, **kwargs)[source]¶

Bases: kafka.protocol.struct.Struct

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.SyncGroupRequest_v0(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

API_KEY = 14¶

API_VERSION = 0¶

RESPONSE_TYPE¶: alias of SyncGroupResponse_v0

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.SyncGroupRequest_v1(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

API_KEY = 14¶

API_VERSION = 1¶

RESPONSE_TYPE¶: alias of SyncGroupResponse_v1

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.SyncGroupResponse_v0(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 14¶

API_VERSION = 0¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.group.SyncGroupResponse_v1(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 14¶

API_VERSION = 1¶

SCHEMA = <kafka.protocol.types.Schema object>¶

kafka.protocol.legacy module¶

class kafka.protocol.legacy.KafkaProtocol[source]¶

Bases: 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.

CONSUMER_METADATA_KEY = 10¶

FETCH_KEY = 1¶

METADATA_KEY = 3¶

OFFSET_COMMIT_KEY = 8¶

OFFSET_FETCH_KEY = 9¶

OFFSET_KEY = 2¶

PRODUCE_KEY = 0¶

classmethod decode_consumer_metadata_response(data)[source]¶

Decode bytes to a kafka.structs.ConsumerMetadataResponse

Arguments:: data: bytes to decode

classmethod decode_fetch_response(response)[source]¶

Decode FetchResponse struct to FetchResponsePayloads

Arguments:: response: FetchResponse

classmethod decode_list_offset_response(response)[source]¶

Decode OffsetResponse_v2 into ListOffsetResponsePayloads

Arguments:: response: OffsetResponse_v2

Returns: list of ListOffsetResponsePayloads

classmethod decode_message_set(raw_data)[source]¶

classmethod decode_metadata_response(response)[source]¶

classmethod decode_offset_commit_response(response)[source]¶

Decode OffsetCommitResponse to an OffsetCommitResponsePayload

Arguments:: response: OffsetCommitResponse

classmethod decode_offset_fetch_response(response)[source]¶

Decode OffsetFetchResponse to OffsetFetchResponsePayloads

Arguments:: response: OffsetFetchResponse

classmethod decode_offset_response(response)[source]¶

Decode OffsetResponse into OffsetResponsePayloads

Arguments:: response: OffsetResponse

Returns: list of OffsetResponsePayloads

classmethod decode_produce_response(response)[source]¶

Decode ProduceResponse to ProduceResponsePayload

Arguments:: response: ProduceResponse

Return: list of ProduceResponsePayload

classmethod encode_consumer_metadata_request(client_id, correlation_id, payloads)[source]¶

Encode a ConsumerMetadataRequest

Arguments:: client_id: string correlation_id: int payloads: string (consumer group)

classmethod encode_fetch_request(payloads=(), max_wait_time=100, min_bytes=4096)[source]¶

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

classmethod encode_list_offset_request(payloads=())[source]¶

classmethod encode_metadata_request(topics=(), payloads=None)[source]¶

Encode a MetadataRequest

Arguments:: topics: list of strings

classmethod encode_offset_commit_request(group, payloads)[source]¶

Encode an OffsetCommitRequest struct

Arguments:: group: string, the consumer group you are committing offsets for payloads: list of OffsetCommitRequestPayload

classmethod encode_offset_fetch_request(group, payloads, from_kafka=False)[source]¶

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

classmethod encode_offset_request(payloads=())[source]¶

classmethod encode_produce_request(payloads=(), acks=1, timeout=1000)[source]¶

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

kafka.protocol.legacy.create_gzip_message(payloads, key=None, compresslevel=None)[source]¶

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)

kafka.protocol.legacy.create_message(payload, key=None)[source]¶

Construct a Message

Arguments:: payload: bytes, the payload to send to Kafka key: bytes, a key used for partition routing (optional)

kafka.protocol.legacy.create_message_set(messages, codec=0, key=None, compresslevel=None)[source]¶

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.

kafka.protocol.legacy.create_snappy_message(payloads, key=None)[source]¶

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)

kafka.protocol.message module¶

class kafka.protocol.message.Message(value, key=None, magic=0, attributes=0, crc=0, timestamp=None)[source]¶

Bases: kafka.protocol.struct.Struct

CODEC_GZIP = 1¶

CODEC_LZ4 = 3¶

CODEC_MASK = 7¶

CODEC_SNAPPY = 2¶

HEADER_SIZE = 22¶

SCHEMA = <kafka.protocol.types.Schema object>¶

SCHEMAS = [<kafka.protocol.types.Schema object>, <kafka.protocol.types.Schema object>]¶

TIMESTAMP_TYPE_MASK = 8¶

classmethod decode(data)[source]¶

decompress()[source]¶

is_compressed()[source]¶

timestamp_type¶

0 for CreateTime; 1 for LogAppendTime; None if unsupported.

Value is determined by broker; produced messages should always set to 0 Requires Kafka >= 0.10 / message version >= 1

validate_crc()[source]¶

class kafka.protocol.message.MessageSet[source]¶

Bases: kafka.protocol.abstract.AbstractType

HEADER_SIZE = 12¶

ITEM = <kafka.protocol.types.Schema object>¶

classmethod decode(data, bytes_to_read=None)[source]¶: Compressed messages should pass in bytes_to_read (via message size) otherwise, we decode from data as Int32

classmethod encode(items, prepend_size=True)[source]¶

classmethod repr(messages)[source]¶

class kafka.protocol.message.PartialMessage[source]¶: Bases: bytes

kafka.protocol.metadata module¶

class kafka.protocol.metadata.MetadataRequest_v0(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

ALL_TOPICS = None¶

API_KEY = 3¶

API_VERSION = 0¶

RESPONSE_TYPE¶: alias of MetadataResponse_v0

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.metadata.MetadataRequest_v1(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

ALL_TOPICS = -1¶

API_KEY = 3¶

API_VERSION = 1¶

NO_TOPICS = None¶

RESPONSE_TYPE¶: alias of MetadataResponse_v1

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.metadata.MetadataRequest_v2(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

ALL_TOPICS = -1¶

API_KEY = 3¶

API_VERSION = 2¶

NO_TOPICS = None¶

RESPONSE_TYPE¶: alias of MetadataResponse_v2

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.metadata.MetadataRequest_v3(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

ALL_TOPICS = -1¶

API_KEY = 3¶

API_VERSION = 3¶

NO_TOPICS = None¶

RESPONSE_TYPE¶: alias of MetadataResponse_v3

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.metadata.MetadataRequest_v4(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

ALL_TOPICS = -1¶

API_KEY = 3¶

API_VERSION = 4¶

NO_TOPICS = None¶

RESPONSE_TYPE¶: alias of MetadataResponse_v4

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.metadata.MetadataRequest_v5(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

The v5 metadata request is the same as v4. An additional field for offline_replicas has been added to the v5 metadata response

ALL_TOPICS = -1¶

API_KEY = 3¶

API_VERSION = 5¶

NO_TOPICS = None¶

RESPONSE_TYPE¶: alias of MetadataResponse_v5

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.metadata.MetadataResponse_v0(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 3¶

API_VERSION = 0¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.metadata.MetadataResponse_v1(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 3¶

API_VERSION = 1¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.metadata.MetadataResponse_v2(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 3¶

API_VERSION = 2¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.metadata.MetadataResponse_v3(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 3¶

API_VERSION = 3¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.metadata.MetadataResponse_v4(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 3¶

API_VERSION = 4¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.metadata.MetadataResponse_v5(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 3¶

API_VERSION = 5¶

SCHEMA = <kafka.protocol.types.Schema object>¶

kafka.protocol.offset module¶

class kafka.protocol.offset.OffsetRequest_v0(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

API_KEY = 2¶

API_VERSION = 0¶

DEFAULTS = {'replica_id': -1}¶

RESPONSE_TYPE¶: alias of OffsetResponse_v0

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.offset.OffsetRequest_v1(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

API_KEY = 2¶

API_VERSION = 1¶

DEFAULTS = {'replica_id': -1}¶

RESPONSE_TYPE¶: alias of OffsetResponse_v1

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.offset.OffsetRequest_v2(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Request

API_KEY = 2¶

API_VERSION = 2¶

DEFAULTS = {'replica_id': -1}¶

RESPONSE_TYPE¶: alias of OffsetResponse_v2

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.offset.OffsetResetStrategy[source]¶

Bases: object

EARLIEST = -2¶

LATEST = -1¶

NONE = 0¶

class kafka.protocol.offset.OffsetResponse_v0(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 2¶

API_VERSION = 0¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.offset.OffsetResponse_v1(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 2¶

API_VERSION = 1¶

SCHEMA = <kafka.protocol.types.Schema object>¶

class kafka.protocol.offset.OffsetResponse_v2(*args, **kwargs)[source]¶

Bases: kafka.protocol.api.Response

API_KEY = 2¶

API_VERSION = 2¶

SCHEMA = <kafka.protocol.types.Schema object>¶