Kafka

Generated from the type stubs and docstrings. Do not edit by hand.

Types

Acl

Acl(principal: str, host: str, operation: AclOperation, permission_type: AclPermissionType)

Fields:

• principal: str
• host: str
• operation: AclOperation
• permission_type: AclPermissionType

---

AclCreation

AclCreation(resource_type: AclResourceType, resource_name: str, principal: str, host: str, operation: AclOperation, permission_type: AclPermissionType, pattern_type: AclPatternType = ...)

Fields:

• resource_type: AclResourceType
• resource_name: str
• principal: str
• host: str
• operation: AclOperation
• permission_type: AclPermissionType
• pattern_type: AclPatternType

---

AclFilter

AclFilter(resource_type: AclResourceType = ..., resource_name: Optional[str] = None, principal: Optional[str] = None, host: Optional[str] = None, operation: AclOperation = ..., permission_type: AclPermissionType = ..., pattern_type: AclPatternType = ..., confirm_match_all: bool = False)

Fields:

• resource_type: AclResourceType
• resource_name: Optional[str]
• principal: Optional[str]
• host: Optional[str]
• operation: AclOperation
• permission_type: AclPermissionType
• pattern_type: AclPatternType
• confirm_match_all: bool

match_all

match_all(cls) -> Self

---

AclOperation

Fields:

• UNKNOWN: int
• ANY: int
• ALL: int
• READ: int
• WRITE: int
• CREATE: int
• DELETE: int
• ALTER: int
• DESCRIBE: int
• CLUSTER_ACTION: int
• DESCRIBE_CONFIGS: int
• ALTER_CONFIGS: int
• IDEMPOTENT_WRITE: int

---

AclPatternType

Fields:

• UNKNOWN: int
• ANY: int
• MATCH: int
• LITERAL: int
• PREFIXED: int

---

AclPermissionType

Fields:

• UNKNOWN: int
• ANY: int
• DENY: int
• ALLOW: int

---

AclResource

AclResource(resource_type: AclResourceType, resource_name: str, acls: IList[Acl] = ..., pattern_type: AclPatternType = ...)

Fields:

• resource_type: AclResourceType
• resource_name: str
• acls: IList[Acl]
• pattern_type: AclPatternType

---

AclResourceType

Fields:

• UNKNOWN: int
• ANY: int
• TOPIC: int
• GROUP: int
• CLUSTER: int
• TRANSACTIONAL_ID: int
• DELEGATION_TOKEN: int

---

AclResult

AclResult(error: str = '')

Fields:

• error: str

---

AdminError

---

AlterConfigOpType

Fields:

• SET: int
• DELETE: int
• APPEND: int
• SUBTRACT: int

---

ArrowBatch

Fields:

• num_rows: int

---

ArrowPollResult

Fields:

• batch: ArrowBatch
• offsets: dict[tuple[str, int], int]

---

AsyncKafkaConsumer

Delegates every call to a KafkaConsumer on a worker thread; each method returns exactly what its sync counterpart on KafkaConsumer returns. The worker serves concurrent coroutines one at a time, so they never meet the one-poller refusal.

AsyncKafkaConsumer(config: KafkaConsumerConfig)

close

close(self) -> None

poll_batch

poll_batch(self, max_records: int = 500, timeout_ms: int = 500) -> Either[KafkaBatch, KafkaError]

poll_bytes_batch

poll_bytes_batch(self, max_records: int = 500, timeout_ms: int = 500) -> Either[BytesBatch, KafkaError]

consume

consume(self, max_records: int = 500, timeout_ms: int = 500) -> AsyncIterator[KafkaRecord]

store_offsets

store_offsets(self, batch_or_offsets: KafkaBatch | BytesBatch | IMap[TopicPartition, int]) -> Either[None, KafkaError]

commit

commit(self, batch: Optional[KafkaBatch | BytesBatch | IMap[TopicPartition, int]] = None) -> Either[CommitReport, KafkaError]

commit_offsets

commit_offsets(self, offsets: IMap[TopicPartition, int]) -> Either[CommitReport, KafkaError]

subscribe

subscribe(self, topics: IList[str] | list[str] | str | re.Pattern[str], on_assign: Optional[Callable[[KafkaConsumer, IList[TopicPartition]], None]] = None, on_revoke: Optional[Callable[[KafkaConsumer, IList[TopicPartition]], None]] = None, on_lost: Optional[Callable[[KafkaConsumer, IList[TopicPartition]], None]] = None) -> Either[None, ConsumerError]

unsubscribe

unsubscribe(self) -> Either[None, ConsumerError]

assign

assign(self, partitions: IList[TopicPartition] | list[TopicPartition], start_offsets: Optional[IMap[TopicPartition, int] | dict[TopicPartition, int]] = None) -> Either[None, ConsumerError]

unassign

unassign(self) -> Either[None, ConsumerError]

assignment

assignment(self) -> IList[TopicPartition]

seek

seek(self, topic_partition: TopicPartition, offset: int) -> Either[None, ConsumerError]

seek_to_beginning

seek_to_beginning(self, topic_partition: TopicPartition) -> Either[None, ConsumerError]

seek_to_end

seek_to_end(self, topic_partition: TopicPartition) -> Either[None, ConsumerError]

rewind

rewind(self, batch: KafkaBatch) -> Either[None, ConsumerError]

committed

committed(self, topic_partition: TopicPartition) -> Option[int]
committed(self, topic_partition: IList[TopicPartition] | list[TopicPartition]) -> IMap[TopicPartition, Option[int]]

position

position(self, topic_partition: TopicPartition) -> int

get_watermark_offsets

get_watermark_offsets(self, topic_partition: TopicPartition, timeout_ms: Optional[int] = None) -> Either[tuple[int, int], ConsumerError]

offsets_for_times

offsets_for_times(self, tp_timestamps: IMap[TopicPartition, int], timeout_ms: Optional[int] = None) -> Either[IMap[TopicPartition, int], ConsumerError]

pause

pause(self, partitions: Optional[IList[TopicPartition] | list[TopicPartition]] = None) -> Either[None, KafkaError]

resume

resume(self, partitions: Optional[IList[TopicPartition] | list[TopicPartition]] = None) -> Either[None, KafkaError]

list_topics

list_topics(self) -> Either[IList[str], ConsumerError]

subscription

subscription(self) -> IList[str]

lag

lag(self) -> Either[IMap[TopicPartition, int], KafkaError]

memberid

memberid(self) -> str

consumer_group_metadata

consumer_group_metadata(self) -> Option[ConsumerGroupMetadata]

metrics

metrics(self) -> dict[str, int]

---

AsyncKafkaProducer

Runs every call of a KafkaProducer on the event loop’s executor; each method returns exactly what its sync counterpart on KafkaProducer returns. Usable from several event loops and threads at once: calls awaited in sequence keep their order, concurrent coroutines do not, as threads calling KafkaProducer do not.

AsyncKafkaProducer(config: KafkaConfig)

close

close(self) -> None

produce

produce(self, topic: str, value: Optional[bytes], key: Optional[bytes] = None, *, partition: int = -1, headers: Optional[dict[str, bytes] | IMap[str, bytes] | Sequence[tuple[str, bytes]]] = None, timestamp_ms: Optional[int] = None) -> ProduceReceipt

produce_many_bytes

produce_many_bytes(self, topic: str, values: BytesList | IList[bytes] | list[bytes], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> ProduceReceipt

produce_many_json

produce_many_json(self, topic: str, values: IList[dict] | list[dict], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> ProduceReceipt

produce_many_avro

produce_many_avro(self, topic: str, values: IList[dict] | list[dict], serde: AvroSerde, keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> ProduceReceipt

produce_many_bytes_sync

produce_many_bytes_sync(self, topic: str, values: BytesList | IList[bytes] | list[bytes], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> Either[ProduceReport, KafkaError]

produce_many_json_sync

produce_many_json_sync(self, topic: str, values: IList[dict] | list[dict], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> Either[ProduceReport, KafkaError]

flush

flush(self, timeout_ms: Optional[int] = None) -> Either[FlushReport, KafkaError]

purge

purge(self, in_queue: bool = True, in_flight: bool = True, blocking: bool = True) -> Either[int, KafkaError]

produce_many_records_sync

produce_many_records_sync(self, records: IList[KafkaRecord] | list[KafkaRecord]) -> Either[ProduceReport, KafkaError]

init_transactions

init_transactions(self, timeout_ms: Optional[int] = None) -> Either[None, KafkaError]

begin_transaction

begin_transaction(self) -> Either[None, KafkaError]

commit_transaction

commit_transaction(self, timeout_ms: Optional[int] = None) -> Either[None, KafkaError]

abort_transaction

abort_transaction(self, timeout_ms: Optional[int] = None) -> Either[None, KafkaError]

produce_many_records_txn

produce_many_records_txn(self, records: IList[KafkaRecord] | list[KafkaRecord]) -> ProduceReceipt

send_offsets_to_transaction

send_offsets_to_transaction(self, offsets: IMap[TopicPartition, int], consumer_group_metadata: ConsumerGroupMetadata | Option[ConsumerGroupMetadata], timeout_ms: Optional[int] = None) -> Either[None, KafkaError]

metrics

metrics(self) -> dict[str, int]

---

BrokerMetadata

BrokerMetadata(node_id: int, host: str, port: int, rack: Optional[str] = None)

Fields:

• node_id: int
• host: str
• port: int
• rack: Optional[str]

---

BytesBatch

BytesBatch(values: object, offsets: IMap[TopicPartition, int], count: int)

Fields:

• values: object
• offsets: IMap[TopicPartition, int]
• count: int

---

ClusterMetadata

ClusterMetadata(brokers: IList[BrokerMetadata] = ..., controller_id: int = -1, topics: IList[TopicMetadata] = ..., cluster_id: Optional[str] = None)

Fields:

• brokers: IList[BrokerMetadata]
• controller_id: int
• topics: IList[TopicMetadata]
• cluster_id: Optional[str]

---

CommitError

---

CommitReport

CommitReport(committed: IMap[TopicPartition, int])

Fields:

• committed: IMap[TopicPartition, int]

---

ConfigEntry

ConfigEntry(name: str, value: Optional[str] = None, incremental_operation: Optional[AlterConfigOpType] = None, source: ConfigSource = ..., is_default: bool = False, is_sensitive: bool = False, read_only: bool = False)

Fields:

• name: str
• value: Optional[str]
• incremental_operation: Optional[AlterConfigOpType]
• source: ConfigSource
• is_default: bool
• is_sensitive: bool
• read_only: bool

---

ConfigError

---

ConfigResource

ConfigResource(resource_type: ResourceType, name: str, set_config: IMap[str, str] = ..., incremental_configs: IList[ConfigEntry] = ...)

Fields:

• resource_type: ResourceType
• name: str
• set_config: IMap[str, str]
• incremental_configs: IList[ConfigEntry]

---

ConfigSource

Fields:

• UNKNOWN: int
• DYNAMIC_TOPIC_CONFIG: int
• DYNAMIC_BROKER_CONFIG: int
• DYNAMIC_DEFAULT_BROKER_CONFIG: int
• STATIC_BROKER_CONFIG: int
• DEFAULT_CONFIG: int
• DYNAMIC_BROKER_LOGGER_CONFIG: int

---

ConsumerError

---

ConsumerGroupDetail

ConsumerGroupDetail(group_id: str, state: str, protocol_type: str, protocol: str, members: IList[ConsumerGroupMember])

Fields:

• group_id: str
• state: str
• protocol_type: str
• protocol: str
• members: IList[ConsumerGroupMember]

---

ConsumerGroupInfo

ConsumerGroupInfo(group_id: str, protocol_type: str)

Fields:

• group_id: str
• protocol_type: str

---

ConsumerGroupMember

ConsumerGroupMember(member_id: str, client_id: str, host: str, assignment: IList[TopicPartition])

Fields:

• member_id: str
• client_id: str
• host: str
• assignment: IList[TopicPartition]

---

ConsumerGroupMetadata

ConsumerGroupMetadata(group_id: str, coordinator_addr: str)

Fields:

• group_id: str
• coordinator_addr: str

---

DeletedRecords

DeletedRecords(low_watermark: int)

Fields:

• low_watermark: int

---

Deliveries

Native sequence, one slot per accepted record in produce order: a DeliveryReport where the record landed (offset -1 when the broker did not state it: acks=0), None where it did not - exact on every path, with or without idempotence, whatever acks is, and in a transaction. Built by the Rust core; the constructor exists for tests and the empty default, with -2**63 in offsets marking a record that did not land.

Deliveries(topic: str, partitions: list[int], offsets: list[int])

---

DeliveryReport

Where one produced record landed: an element of ProduceReport.deliveries. offset is -1 when the broker confirmed delivery without an offset (acks=0, or a duplicate-sequence acknowledgement).

DeliveryReport(topic: str, partition: int, offset: int)

Fields:

• topic: str
• partition: int
• offset: int

---

DeliveryTimeout

---

FlushReport

FlushReport(flushed: int, elapsed_ms: int)

Fields:

• flushed: int
• elapsed_ms: int

---

KafkaAdminClient

Kafka admin client. Safe to share between threads; calls from several threads run at once, as confluent-kafka’s AdminClient allows.

KafkaAdminClient(config: KafkaConfig)

create_topics

create_topics(self, specs: IList[TopicSpec] | list[TopicSpec], validate_only: bool = False, operation_timeout_ms: int = 30000, request_timeout_ms: Optional[int] = None) -> Either[IList[TopicResult], AdminError]

delete_topics

delete_topics(self, topics: IList[str] | list[str]) -> Either[IList[TopicResult], AdminError]

list_consumer_groups

list_consumer_groups(self) -> Either[IList[ConsumerGroupInfo], AdminError]

describe_consumer_groups

describe_consumer_groups(self, group_ids: IList[str] | list[str]) -> Either[IList[ConsumerGroupDetail], AdminError]

delete_consumer_groups

delete_consumer_groups(self, group_ids: IList[str] | list[str]) -> Either[IList[TopicResult], AdminError]

create_partitions

create_partitions(self, new_partitions: IList[NewPartitions] | list[NewPartitions], validate_only: bool = False, operation_timeout_ms: int = 30000) -> Either[IList[TopicResult], AdminError]

list_topics

list_topics(self) -> Either[IList[str], AdminError]

describe_cluster

describe_cluster(self, include_authorized_operations: bool = False) -> Either[ClusterMetadata, AdminError]

describe_topics

describe_topics(self, topics: IList[str] | list[str] | IList[TopicMetadata] | list[TopicMetadata], include_authorized_operations: bool = False) -> Either[IList[TopicMetadata], AdminError]

describe_configs

describe_configs(self, resources: IList[ConfigResource] | list[ConfigResource] | IList[str] | list[str]) -> Either[IMap[str, IMap[str, str]], AdminError]

describe_topic_config

describe_topic_config(self, topics: IList[str] | list[str]) -> Either[IMap[str, IMap[str, str]], AdminError]

incremental_alter_configs

incremental_alter_configs(self, resources: IList[ConfigResource] | list[ConfigResource], validate_only: bool = False) -> Either[None, AdminError]

list_offsets

list_offsets(self, offset_specs: IMap[TopicPartition, OffsetSpec], isolation_level: str = 'read_committed') -> Either[IMap[TopicPartition, ListOffsetsResultInfo], AdminError]

delete_records

delete_records(self, offsets: IMap[TopicPartition, int], operation_timeout_ms: int = 30000) -> Either[IMap[TopicPartition, DeletedRecords], AdminError]

list_consumer_group_offsets

list_consumer_group_offsets(self, group_id: str, topics: Optional[IList[TopicPartition] | list[TopicPartition]] = None) -> Either[IMap[TopicPartition, int], AdminError]

alter_consumer_group_offsets

alter_consumer_group_offsets(self, group_id: str, offsets: IMap[TopicPartition, int]) -> Either[None, AdminError]

create_acls

create_acls(self, acls: IList[AclCreation] | list[AclCreation]) -> Either[IList[AclResult], AdminError]

describe_acls

describe_acls(self, resource_type: int = 1, resource_name: Optional[str] = None, principal: Optional[str] = None, host: Optional[str] = None, operation: int = 1, permission_type: int = 1, pattern_type: int = 1) -> Either[IList[AclResource], AdminError]

delete_acls

delete_acls(self, filters: IList[AclFilter] | list[AclFilter]) -> Either[IList[AclResult], AdminError]

---

KafkaBatch

KafkaBatch(records: IList[KafkaRecord], offsets: IMap[TopicPartition, int], count: int)

Fields:

• records: IList[KafkaRecord]
• offsets: IMap[TopicPartition, int]
• count: int

---

KafkaConfig

KafkaConfig(bootstrap_servers: str, client_id: str = 'pyrula', acks: str | int = 'all', compression: str = 'none', partitioner: str = 'murmur2', linger_ms: int = 5, batch_size_bytes: int = 131072, batch_num_messages: int = 10000, sticky_partitioning_linger_ms: Optional[int] = None, max_in_flight_requests_per_connection: Optional[int] = None, topic_metadata_refresh_interval_ms: int = 300000, connect_in_background: bool = False, max_request_bytes: int = 1048576, coalesce_produce_requests: bool = False, raw_connections_per_broker: int = 8, enable_idempotence: Optional[bool] = None, message_timeout_ms: Optional[int] = None, request_timeout_ms: int = 30000, retry_backoff_ms: int = 100, message_send_max_retries: int = 2147483647, queue_buffering_max_messages: int = 100000, queue_buffering_max_kbytes: int = 1048576, transactional_id: str = '', security_protocol: str = 'PLAINTEXT', sasl_mechanism: str = 'PLAIN', sasl_username: Optional[str] = None, sasl_password: Optional[str] = None, oauth_cb: Optional[Callable[[], tuple[str, float] | tuple[str, float, Optional[str]] | tuple[str, float, Optional[str], Optional[dict[str, str]]]]] = None, ssl_ca_location: Optional[str] = None, ssl_certificate_location: Optional[str] = None, ssl_key_location: Optional[str] = None, error_cb: Optional[Callable[[Exception], None]] = None, logger: Optional[logging.Logger] = None, statistics_interval_ms: int = 0, stats_cb: Optional[Callable[[dict[str, int]], None]] = None, extra: Any = ...)

Fields:

• bootstrap_servers: str
• client_id: str
• acks: str | int
• compression: str
• partitioner: str
• linger_ms: int
• batch_size_bytes: int
• batch_num_messages: int
• sticky_partitioning_linger_ms: Optional[int]
• max_in_flight_requests_per_connection: Optional[int]
• topic_metadata_refresh_interval_ms: int
• connect_in_background: bool
• max_request_bytes: int
• coalesce_produce_requests: bool
• raw_connections_per_broker: int
• enable_idempotence: Optional[bool]
• message_timeout_ms: Optional[int]
• request_timeout_ms: int
• retry_backoff_ms: int
• message_send_max_retries: int
• queue_buffering_max_messages: int
• queue_buffering_max_kbytes: int
• transactional_id: str
• security_protocol: str
• sasl_mechanism: str
• sasl_username: Optional[str]
• sasl_password: Optional[str]
• oauth_cb: Optional[Callable[[], tuple[str, float] | tuple[str, float, Optional[str]] | tuple[str, float, Optional[str], Optional[dict[str, str]]]]]
• ssl_ca_location: Optional[str]
• ssl_certificate_location: Optional[str]
• ssl_key_location: Optional[str]
• error_cb: Optional[Callable[[Exception], None]]
• logger: Optional[logging.Logger]
• statistics_interval_ms: int
• stats_cb: Optional[Callable[[dict[str, int]], None]]
• extra: Any

---

KafkaConnectionError

---

KafkaConsumer

Blocking Kafka consumer.

Other threads may call it while one thread polls, as confluent-kafka’s consumer allows: commits, stored offsets, seeks, pauses, positions and the rest are served at once. One thread polls at a time; a second concurrent poll_*() raises ConsumerError (_conflicting_use). close() from another thread wakes a blocked poll, which returns an empty batch, then leaves the group once. Rebalance callbacks run on the thread whose call drove the rebalance and may call back into the consumer; stats_cb, error_cb and a logger may too, except to poll or close from inside this thread’s own poll, or to change the subscription from inside a rebalance this thread is running (_conflicting_use). A background error_cb runs on a client thread, where consumer calls are refused. metrics() answers from any thread at any time. Dropping the consumer without close() frees it but does not leave the group promptly: the group waits out the session, as with confluent-kafka. Call close() to leave and commit staged offsets.

KafkaConsumer(config: KafkaConsumerConfig)

poll_batch

poll_batch(self, max_records: int = 10000, timeout_ms: int = 100) -> Either[KafkaBatch, KafkaError]

poll_bytes_batch

poll_bytes_batch(self, max_records: int = 10000, timeout_ms: int = 100) -> Either[BytesBatch, KafkaError]

poll_bytes_packed

poll_bytes_packed(self, max_records: int = 10000, timeout_ms: int = 100) -> Either[PackedBytesBatch, KafkaError]

poll_bytes_batch_spans

poll_bytes_batch_spans(self, max_records: int = 10000, timeout_ms: int = 100) -> Either[tuple[BytesBatch, list[tuple[int, int, int]]], KafkaError]

poll_arrow

poll_arrow(self, max_records: int = 10000, timeout_ms: int = 100, value_format: str = 'raw', schema: Optional[str] = None, include_metadata: bool = True) -> Either[ArrowPollResult, KafkaError]

store_offsets

store_offsets(self, batch_or_offsets: KafkaBatch | BytesBatch | IMap[TopicPartition, int]) -> Either[None, KafkaError]

commit

commit(self, batch: Optional[KafkaBatch | BytesBatch | IMap[TopicPartition, int]] = None) -> Either[CommitReport, KafkaError]

commit_offsets

commit_offsets(self, offsets: IMap[TopicPartition, int]) -> Either[CommitReport, KafkaError]

subscribe

subscribe(self, topics: IList[str] | list[str] | str | re.Pattern[str], on_assign: Optional[Callable[[KafkaConsumer, IList[TopicPartition]], None]] = None, on_revoke: Optional[Callable[[KafkaConsumer, IList[TopicPartition]], None]] = None, on_lost: Optional[Callable[[KafkaConsumer, IList[TopicPartition]], None]] = None) -> Either[None, ConsumerError]

unsubscribe

unsubscribe(self) -> Either[None, ConsumerError]

assign

assign(self, partitions: IList[TopicPartition] | list[TopicPartition], start_offsets: Optional[IMap[TopicPartition, int] | dict[TopicPartition, int]] = None) -> Either[None, ConsumerError]

unassign

unassign(self) -> Either[None, ConsumerError]

assignment

assignment(self) -> IList[TopicPartition]

incremental_assign

incremental_assign(self, partitions: IList[TopicPartition] | list[TopicPartition], start_offsets: Optional[IMap[TopicPartition, int] | dict[TopicPartition, int]] = None) -> Either[None, ConsumerError]

incremental_unassign

incremental_unassign(self, partitions: IList[TopicPartition] | list[TopicPartition]) -> Either[None, ConsumerError]

seek

seek(self, topic_partition: TopicPartition, offset: int) -> Either[None, ConsumerError]

seek_to_beginning

seek_to_beginning(self, topic_partition: TopicPartition) -> Either[None, ConsumerError]

seek_to_end

seek_to_end(self, topic_partition: TopicPartition) -> Either[None, ConsumerError]

rewind

rewind(self, batch: KafkaBatch) -> Either[None, ConsumerError]

position

position(self, topic_partition: TopicPartition) -> int

committed

committed(self, topic_partition: TopicPartition) -> Option[int]
committed(self, topic_partition: IList[TopicPartition] | list[TopicPartition]) -> IMap[TopicPartition, Option[int]]

get_watermark_offsets

get_watermark_offsets(self, topic_partition: TopicPartition, timeout_ms: Optional[int] = None) -> Either[tuple[int, int], ConsumerError]

offsets_for_times

offsets_for_times(self, tp_timestamps: IMap[TopicPartition, int], timeout_ms: Optional[int] = None) -> Either[IMap[TopicPartition, int], ConsumerError]

pause

pause(self, partitions: Optional[IList[TopicPartition] | list[TopicPartition]] = None) -> Either[None, KafkaError]

resume

resume(self, partitions: Optional[IList[TopicPartition] | list[TopicPartition]] = None) -> Either[None, KafkaError]

list_topics

list_topics(self) -> Either[IList[str], ConsumerError]

subscription

subscription(self) -> IList[str]

lag

lag(self) -> Either[IMap[TopicPartition, int], KafkaError]

memberid

memberid(self) -> str

consumer_group_metadata

consumer_group_metadata(self) -> Option[ConsumerGroupMetadata]

metrics

metrics(self) -> dict[str, int]

close

close(self) -> None

---

KafkaConsumerConfig

KafkaConsumerConfig(bootstrap_servers: str, topics: list[str], group_id: str = '', group_instance_id: str = '', enable_auto_commit: bool = False, auto_commit_interval_ms: int = 5000, auto_offset_reset: str = 'earliest', max_poll_interval_ms: int = 300000, session_timeout_ms: int = 10000, heartbeat_interval_ms: int = 3000, request_timeout_ms: int = 30000, partition_assignment_strategy: str = 'range', rebalance_timeout_ms: Optional[int] = None, topic_metadata_refresh_interval_ms: int = 300000, topic_metadata_refresh_fast_interval_ms: int = 250, isolation_level: str = 'read_committed', max_poll_records: int = 500, fetch_max_bytes: int = 52428800, fetch_min_bytes: int = 1, fetch_wait_max_ms: int = 500, max_partition_fetch_bytes: int = 1048576, fetch_mode: str = 'auto', prefetch_max_bytes: int = 67108864, security_protocol: str = 'PLAINTEXT', sasl_mechanism: str = 'PLAIN', sasl_username: Optional[str] = None, sasl_password: Optional[str] = None, oauth_cb: Optional[Callable[[], tuple[str, float] | tuple[str, float, Optional[str]] | tuple[str, float, Optional[str], Optional[dict[str, str]]]]] = None, ssl_ca_location: Optional[str] = None, ssl_certificate_location: Optional[str] = None, ssl_key_location: Optional[str] = None, error_cb: Optional[Callable[[Exception], None]] = None, logger: Optional[logging.Logger] = None, statistics_interval_ms: int = 0, stats_cb: Optional[Callable[[dict[str, int]], None]] = None, extra: Any = ..., client_id: str = 'pyrula', allow_auto_create_topics: bool = False)

Fields:

• bootstrap_servers: str
• topics: list[str]
• group_id: str
• group_instance_id: str
• enable_auto_commit: bool
• auto_commit_interval_ms: int
• auto_offset_reset: str
• max_poll_interval_ms: int
• session_timeout_ms: int
• heartbeat_interval_ms: int
• request_timeout_ms: int
• partition_assignment_strategy: str
• rebalance_timeout_ms: Optional[int]
• topic_metadata_refresh_interval_ms: int
• topic_metadata_refresh_fast_interval_ms: int
• isolation_level: str
• max_poll_records: int
• fetch_max_bytes: int
• fetch_min_bytes: int
• fetch_wait_max_ms: int
• max_partition_fetch_bytes: int
• fetch_mode: str
• prefetch_max_bytes: int
• security_protocol: str
• sasl_mechanism: str
• sasl_username: Optional[str]
• sasl_password: Optional[str]
• oauth_cb: Optional[Callable[[], tuple[str, float] | tuple[str, float, Optional[str]] | tuple[str, float, Optional[str], Optional[dict[str, str]]]]]
• ssl_ca_location: Optional[str]
• ssl_certificate_location: Optional[str]
• ssl_key_location: Optional[str]
• error_cb: Optional[Callable[[Exception], None]]
• logger: Optional[logging.Logger]
• statistics_interval_ms: int
• stats_cb: Optional[Callable[[dict[str, int]], None]]
• extra: Any
• client_id: str
• allow_auto_create_topics: bool

---

KafkaError

---

KafkaErrorCode

Named Kafka broker error codes (confluent-kafka compatible).

Fields:

• UNKNOWN_SERVER_ERROR: int
• NO_ERROR: int
• OFFSET_OUT_OF_RANGE: int
• CORRUPT_MESSAGE: int
• UNKNOWN_TOPIC_OR_PARTITION: int
• INVALID_FETCH_SIZE: int
• LEADER_NOT_AVAILABLE: int
• NOT_LEADER_OR_FOLLOWER: int
• REQUEST_TIMED_OUT: int
• BROKER_NOT_AVAILABLE: int
• REPLICA_NOT_AVAILABLE: int
• MESSAGE_TOO_LARGE: int
• STALE_CONTROLLER_EPOCH: int
• OFFSET_METADATA_TOO_LARGE: int
• NETWORK_EXCEPTION: int
• COORDINATOR_LOAD_IN_PROGRESS: int
• COORDINATOR_NOT_AVAILABLE: int
• NOT_COORDINATOR: int
• INVALID_TOPIC_EXCEPTION: int
• RECORD_LIST_TOO_LARGE: int
• NOT_ENOUGH_REPLICAS: int
• NOT_ENOUGH_REPLICAS_AFTER_APPEND: int
• INVALID_REQUIRED_ACKS: int
• ILLEGAL_GENERATION: int
• INCONSISTENT_GROUP_PROTOCOL: int
• INVALID_GROUP_ID: int
• UNKNOWN_MEMBER_ID: int
• INVALID_SESSION_TIMEOUT: int
• REBALANCE_IN_PROGRESS: int
• INVALID_COMMIT_OFFSET_SIZE: int
• TOPIC_AUTHORIZATION_FAILED: int
• GROUP_AUTHORIZATION_FAILED: int
• CLUSTER_AUTHORIZATION_FAILED: int
• INVALID_TIMESTAMP: int
• UNSUPPORTED_SASL_MECHANISM: int
• ILLEGAL_SASL_STATE: int
• UNSUPPORTED_VERSION: int
• TOPIC_ALREADY_EXISTS: int
• INVALID_PARTITIONS: int
• INVALID_REPLICATION_FACTOR: int
• INVALID_REPLICA_ASSIGNMENT: int
• INVALID_CONFIG: int
• NOT_CONTROLLER: int
• INVALID_REQUEST: int
• UNSUPPORTED_FOR_MESSAGE_FORMAT: int
• POLICY_VIOLATION: int
• OUT_OF_ORDER_SEQUENCE_NUMBER: int
• DUPLICATE_SEQUENCE_NUMBER: int
• INVALID_PRODUCER_EPOCH: int
• INVALID_TXN_STATE: int
• INVALID_PRODUCER_ID_MAPPING: int
• INVALID_TRANSACTION_TIMEOUT: int
• CONCURRENT_TRANSACTIONS: int
• TRANSACTION_COORDINATOR_FENCED: int
• TRANSACTIONAL_ID_AUTHORIZATION_FAILED: int
• SECURITY_DISABLED: int
• OPERATION_NOT_ATTEMPTED: int
• KAFKA_STORAGE_ERROR: int
• LOG_DIR_NOT_FOUND: int
• SASL_AUTHENTICATION_FAILED: int
• UNKNOWN_PRODUCER_ID: int
• REASSIGNMENT_IN_PROGRESS: int
• DELEGATION_TOKEN_AUTH_DISABLED: int
• DELEGATION_TOKEN_NOT_FOUND: int
• DELEGATION_TOKEN_OWNER_MISMATCH: int
• DELEGATION_TOKEN_REQUEST_NOT_ALLOWED: int
• DELEGATION_TOKEN_AUTHORIZATION_FAILED: int
• DELEGATION_TOKEN_EXPIRED: int
• INVALID_PRINCIPAL_TYPE: int
• NON_EMPTY_GROUP: int
• GROUP_ID_NOT_FOUND: int
• FETCH_SESSION_ID_NOT_FOUND: int
• INVALID_FETCH_SESSION_EPOCH: int
• LISTENER_NOT_FOUND: int
• TOPIC_DELETION_DISABLED: int
• FENCED_LEADER_EPOCH: int
• UNKNOWN_LEADER_EPOCH: int
• UNSUPPORTED_COMPRESSION_TYPE: int
• STALE_BROKER_EPOCH: int
• OFFSET_NOT_AVAILABLE: int
• MEMBER_ID_REQUIRED: int
• PREFERRED_LEADER_NOT_AVAILABLE: int
• GROUP_MAX_SIZE_REACHED: int
• FENCED_INSTANCE_ID: int
• ELIGIBLE_LEADERS_NOT_AVAILABLE: int
• ELECTION_NOT_NEEDED: int
• NO_REASSIGNMENT_IN_PROGRESS: int
• GROUP_SUBSCRIBED_TO_TOPIC: int
• INVALID_RECORD: int
• UNSTABLE_OFFSET_COMMIT: int
• THROTTLING_QUOTA_EXCEEDED: int
• PRODUCER_FENCED: int
• RESOURCE_NOT_FOUND: int
• DUPLICATE_RESOURCE: int
• UNACCEPTABLE_CREDENTIAL: int
• INCONSISTENT_VOTER_SET: int
• INVALID_UPDATE_VERSION: int
• FEATURE_UPDATE_FAILED: int
• PRINCIPAL_DESERIALIZATION_FAILURE: int
• SNAPSHOT_NOT_FOUND: int
• POSITION_OUT_OF_RANGE: int
• UNKNOWN_TOPIC_ID: int
• DUPLICATE_BROKER_REGISTRATION: int
• BROKER_ID_NOT_REGISTERED: int
• INCONSISTENT_TOPIC_ID: int
• INCONSISTENT_CLUSTER_ID: int
• TRANSACTIONAL_ID_NOT_FOUND: int

---

KafkaProducer

Produces records in batches; see the producer guide. Safe from any thread: every method may be called from several threads at once, and a producer may be closed or freed on a thread other than the one that made it. One transactional call runs at a time - a different one made meanwhile returns an error marked _conflicting_use, the same one from another thread _prev_in_progress, both retriable. Every blocking call releases the GIL while it waits, and one made from error_cb, stats_cb or the logger (which run on the producer’s own worker thread) raises ProduceError marked _conflicting_use instead of blocking that thread on itself. A refused send raises rather than returning an error, from produce* and the _sync variants alike: a closed producer, an oversized record, a full queue (QueueFull), and an idempotent producer that has failed fatally (ProduceError marked _producer_fatal; close it and make a new one). The Either a _sync call returns carries delivery failures.

KafkaProducer(config: KafkaConfig)

produce

produce(self, topic: str, value: Optional[bytes], key: Optional[bytes] = None, *, partition: int = -1, headers: Optional[dict[str, bytes] | IMap[str, bytes] | Sequence[tuple[str, bytes]]] = None, timestamp_ms: Optional[int] = None) -> ProduceReceipt

produce_many_bytes

produce_many_bytes(self, topic: str, values: BytesList | IList[bytes] | list[bytes], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> ProduceReceipt

produce_many_json

produce_many_json(self, topic: str, values: IList[dict] | list[dict], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> ProduceReceipt

produce_many_avro

produce_many_avro(self, topic: str, values: IList[dict] | list[dict], serde: AvroSerde, keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> ProduceReceipt

produce_many

produce_many(self, records: IList[KafkaRecord] | list[KafkaRecord]) -> ProduceReceipt

produce_many_bytes_sync

produce_many_bytes_sync(self, topic: str, values: BytesList | IList[bytes] | list[bytes], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> Either[ProduceReport, KafkaError]

produce_many_json_sync

produce_many_json_sync(self, topic: str, values: IList[dict] | list[dict], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> Either[ProduceReport, KafkaError]

flush

flush(self, timeout_ms: Optional[int] = None) -> Either[FlushReport, KafkaError]

purge

purge(self, in_queue: bool = True, in_flight: bool = True, blocking: bool = True) -> Either[int, KafkaError]

init_transactions

init_transactions(self, timeout_ms: Optional[int] = None) -> Either[None, KafkaError]

begin_transaction

begin_transaction(self) -> Either[None, KafkaError]

commit_transaction

commit_transaction(self, timeout_ms: Optional[int] = None) -> Either[None, KafkaError]

abort_transaction

abort_transaction(self, timeout_ms: Optional[int] = None) -> Either[None, KafkaError]

produce_many_bytes_txn

produce_many_bytes_txn(self, topic: str, values: BytesList | IList[bytes] | list[bytes], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None, partition: int = -1) -> ProduceReceipt

produce_many_records_txn

produce_many_records_txn(self, records: IList[KafkaRecord] | list[KafkaRecord]) -> ProduceReceipt

send_offsets_to_transaction

send_offsets_to_transaction(self, offsets: IMap[TopicPartition, int], consumer_group_metadata: ConsumerGroupMetadata | Option[ConsumerGroupMetadata], timeout_ms: Optional[int] = None) -> Either[None, KafkaError]

list_topics

list_topics(self) -> Either[IList[str], KafkaError]

metrics

metrics(self) -> dict[str, int]

close

close(self, timeout_ms: Optional[int] = None) -> None

---

KafkaProducerConfig

Alias of KafkaConfig.

KafkaProducerConfig(bootstrap_servers: str, client_id: str = 'pyrula', acks: str | int = 'all', compression: str = 'none', partitioner: str = 'murmur2', linger_ms: int = 5, batch_size_bytes: int = 131072, batch_num_messages: int = 10000, sticky_partitioning_linger_ms: Optional[int] = None, max_in_flight_requests_per_connection: Optional[int] = None, topic_metadata_refresh_interval_ms: int = 300000, connect_in_background: bool = False, max_request_bytes: int = 1048576, coalesce_produce_requests: bool = False, raw_connections_per_broker: int = 8, enable_idempotence: Optional[bool] = None, message_timeout_ms: Optional[int] = None, request_timeout_ms: int = 30000, retry_backoff_ms: int = 100, message_send_max_retries: int = 2147483647, queue_buffering_max_messages: int = 100000, queue_buffering_max_kbytes: int = 1048576, transactional_id: str = '', security_protocol: str = 'PLAINTEXT', sasl_mechanism: str = 'PLAIN', sasl_username: Optional[str] = None, sasl_password: Optional[str] = None, oauth_cb: Optional[Callable[[], tuple[str, float] | tuple[str, float, Optional[str]] | tuple[str, float, Optional[str], Optional[dict[str, str]]]]] = None, ssl_ca_location: Optional[str] = None, ssl_certificate_location: Optional[str] = None, ssl_key_location: Optional[str] = None, error_cb: Optional[Callable[[Exception], None]] = None, logger: Optional[logging.Logger] = None, statistics_interval_ms: int = 0, stats_cb: Optional[Callable[[dict[str, int]], None]] = None, extra: Any = ...)

Fields:

• bootstrap_servers: str
• client_id: str
• acks: str | int
• compression: str
• partitioner: str
• linger_ms: int
• batch_size_bytes: int
• batch_num_messages: int
• sticky_partitioning_linger_ms: Optional[int]
• max_in_flight_requests_per_connection: Optional[int]
• topic_metadata_refresh_interval_ms: int
• connect_in_background: bool
• max_request_bytes: int
• coalesce_produce_requests: bool
• raw_connections_per_broker: int
• enable_idempotence: Optional[bool]
• message_timeout_ms: Optional[int]
• request_timeout_ms: int
• retry_backoff_ms: int
• message_send_max_retries: int
• queue_buffering_max_messages: int
• queue_buffering_max_kbytes: int
• transactional_id: str
• security_protocol: str
• sasl_mechanism: str
• sasl_username: Optional[str]
• sasl_password: Optional[str]
• oauth_cb: Optional[Callable[[], tuple[str, float] | tuple[str, float, Optional[str]] | tuple[str, float, Optional[str], Optional[dict[str, str]]]]]
• ssl_ca_location: Optional[str]
• ssl_certificate_location: Optional[str]
• ssl_key_location: Optional[str]
• error_cb: Optional[Callable[[Exception], None]]
• logger: Optional[logging.Logger]
• statistics_interval_ms: int
• stats_cb: Optional[Callable[[dict[str, int]], None]]
• extra: Any

---

KafkaRecord

-1 on the produce path means route by key (round-robin without one); consumed records always carry their partition.

KafkaRecord(topic: str, value: Optional[bytes], partition: int = -1, offset: int = -1, key: Option[bytes] | bytes | None = ..., headers: Optional[dict[str, bytes] | IMap[str, bytes] | Sequence[tuple[str, bytes]]] = ..., timestamp_ms: Option[int] | int | None = ...)

Fields:

• topic: str
• value: Optional[bytes]
• partition: int
• offset: int
• key: Option[bytes]
• headers: IMap[str, bytes]
• header_list: list[tuple[str, bytes]]
• timestamp_ms: Option[int]

---

ListOffsetsResultInfo

ListOffsetsResultInfo(offset: int, timestamp_ms: int)

Fields:

• offset: int
• timestamp_ms: int

---

MaxPollIntervalExceeded

---

NewPartitions

NewPartitions(topic: str, total_count: int)

Fields:

• topic: str
• total_count: int

---

OffsetSpec

OffsetSpec(kind: str, timestamp_ms: int = -1)

Fields:

• kind: str
• timestamp_ms: int

earliest

earliest(cls) -> OffsetSpec

latest

latest(cls) -> OffsetSpec

max_timestamp

max_timestamp(cls) -> OffsetSpec

for_timestamp

for_timestamp(cls, timestamp_ms: int) -> OffsetSpec

---

PackedBytesBatch

PackedBytesBatch(packed: bytes, offsets: IMap[TopicPartition, int], count: int)

Fields:

• packed: bytes
• offsets: IMap[TopicPartition, int]
• count: int

to_bytes_list

to_bytes_list(self) -> BytesList

---

PartitionMetadata

PartitionMetadata(partition_index: int, leader_id: int, replicas: IList[int] = ..., isr: IList[int] = ..., error_code: Optional[int] = None)

Fields:

• partition_index: int
• leader_id: int
• replicas: IList[int]
• isr: IList[int]
• error_code: Optional[int]

---

PollError

---

ProduceError

Fields:

• report: ProduceReport
• delivered: int
• failed: int

---

ProduceReceipt

get

get(self) -> Either[ProduceReport, KafkaError]

wait

wait(self) -> Either[ProduceReport, KafkaError]

try_get

try_get(self) -> Optional[Either[ProduceReport, KafkaError]]

---

ProduceReport

ProduceReport(accepted: int, delivered: int, failed: int, errors: list[ProduceError] = ..., deliveries: Deliveries = ...)

Fields:

• accepted: int
• delivered: int
• failed: int
• errors: list[ProduceError]
• deliveries: Deliveries

---

QueueFull

---

RebalanceError

---

ResourceType

Fields:

• UNKNOWN: int
• TOPIC: int
• BROKER: int

---

TopicDescription

Alias of TopicMetadata.

TopicDescription(name: str, partitions: IList[PartitionMetadata] = ..., is_internal: bool = False, error: Optional[str] = None, error_code: Optional[int] = None)

Fields:

• name: str
• partitions: IList[PartitionMetadata]
• is_internal: bool
• error: Optional[str]
• error_code: Optional[int]

---

TopicMetadata

TopicMetadata(name: str, partitions: IList[PartitionMetadata] = ..., is_internal: bool = False, error: Optional[str] = None, error_code: Optional[int] = None)

Fields:

• name: str
• partitions: IList[PartitionMetadata]
• is_internal: bool
• error: Optional[str]
• error_code: Optional[int]

---

TopicPartition

TopicPartition(topic: str, partition: int)

Fields:

• topic: str
• partition: int

---

TopicPartitionInfo

Alias of PartitionMetadata.

TopicPartitionInfo(partition_index: int, leader_id: int, replicas: IList[int] = ..., isr: IList[int] = ..., error_code: Optional[int] = None)

Fields:

• partition_index: int
• leader_id: int
• replicas: IList[int]
• isr: IList[int]
• error_code: Optional[int]

---

TopicResult

TopicResult(name: str, error: str = '', code: int = 0, message: Optional[str] = None)

Fields:

• name: str
• error: str
• code: int
• message: Optional[str]

---

TopicSpec

TopicSpec(name: str, num_partitions: int = 1, replication_factor: int = 1, config: IMap[str, str] = ...)

Fields:

• name: str
• num_partitions: int
• replication_factor: int
• config: IMap[str, str]

Constants

• CAPABILITIES: tuple[str, ...]
• CAPABILITY_DEFINITIONS: dict[str, str]
• GUARANTEES: dict[str, Guarantee] - see the client guarantee matrix
• OFFSET_BEGINNING: int
• OFFSET_END: int
• OFFSET_INVALID: int
• OFFSET_STORED: int

Functions

avro_schema_dict

avro_schema_dict(model: type, *, name: Optional[str] = None) -> dict

avro_schema_json

avro_schema_json(schema: Any) -> str

consume

consume(consumer: KafkaConsumer, *, max_records: int = 500, timeout_ms: int = 500) -> Generator[KafkaRecord]

Convenience generator: yields KafkaRecord objects one at a time.

Usage: for record in consume(consumer, max_records=100): process(record)

Stops only on StopIteration (never blocks forever); caller must break out.

is_schema_type

is_schema_type(t: Any) -> bool

json_schema_json

json_schema_json(schema: Any) -> str