# Kafka > Producers, consumers, admin, serdes, and config for pyrula.kafka. Source: https://docs.pyrula.io/reference/kafka/ Generated from the type stubs and docstrings. Do not edit by hand. ## Types ## `Acl` ```python Acl(principal: str, host: str, operation: AclOperation, permission_type: AclPermissionType) ``` **Fields:** - `principal: str` - `host: str` - `operation: AclOperation` - `permission_type: AclPermissionType` --- ## `AclCreation` ```python 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` ```python 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` ```python 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` ```python 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` ```python 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. ```python AsyncKafkaConsumer(config: KafkaConsumerConfig) ``` ### `close` ```python close(self) -> None ``` ### `poll_batch` ```python poll_batch(self, max_records: int = 500, timeout_ms: int = 500) -> Either[KafkaBatch, KafkaError] ``` ### `poll_bytes_batch` ```python poll_bytes_batch(self, max_records: int = 500, timeout_ms: int = 500) -> Either[BytesBatch, KafkaError] ``` ### `consume` ```python consume(self, max_records: int = 500, timeout_ms: int = 500) -> AsyncIterator[KafkaRecord] ``` ### `store_offsets` ```python store_offsets(self, batch_or_offsets: KafkaBatch | BytesBatch | IMap[TopicPartition, int]) -> Either[None, KafkaError] ``` ### `commit` ```python commit(self, batch: Optional[KafkaBatch | BytesBatch | IMap[TopicPartition, int]] = None) -> Either[CommitReport, KafkaError] ``` ### `commit_offsets` ```python commit_offsets(self, offsets: IMap[TopicPartition, int]) -> Either[CommitReport, KafkaError] ``` ### `subscribe` ```python 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` ```python unsubscribe(self) -> Either[None, ConsumerError] ``` ### `assign` ```python assign(self, partitions: IList[TopicPartition] | list[TopicPartition], start_offsets: Optional[IMap[TopicPartition, int] | dict[TopicPartition, int]] = None) -> Either[None, ConsumerError] ``` ### `unassign` ```python unassign(self) -> Either[None, ConsumerError] ``` ### `assignment` ```python assignment(self) -> IList[TopicPartition] ``` ### `seek` ```python seek(self, topic_partition: TopicPartition, offset: int) -> Either[None, ConsumerError] ``` ### `seek_to_beginning` ```python seek_to_beginning(self, topic_partition: TopicPartition) -> Either[None, ConsumerError] ``` ### `seek_to_end` ```python seek_to_end(self, topic_partition: TopicPartition) -> Either[None, ConsumerError] ``` ### `rewind` ```python rewind(self, batch: KafkaBatch) -> Either[None, ConsumerError] ``` ### `committed` ```python committed(self, topic_partition: TopicPartition) -> Option[int] committed(self, topic_partition: IList[TopicPartition] | list[TopicPartition]) -> IMap[TopicPartition, Option[int]] ``` ### `position` ```python position(self, topic_partition: TopicPartition) -> int ``` ### `get_watermark_offsets` ```python get_watermark_offsets(self, topic_partition: TopicPartition, timeout_ms: Optional[int] = None) -> Either[tuple[int, int], ConsumerError] ``` ### `offsets_for_times` ```python offsets_for_times(self, tp_timestamps: IMap[TopicPartition, int], timeout_ms: Optional[int] = None) -> Either[IMap[TopicPartition, int], ConsumerError] ``` ### `pause` ```python pause(self, partitions: Optional[IList[TopicPartition] | list[TopicPartition]] = None) -> Either[None, KafkaError] ``` ### `resume` ```python resume(self, partitions: Optional[IList[TopicPartition] | list[TopicPartition]] = None) -> Either[None, KafkaError] ``` ### `list_topics` ```python list_topics(self) -> Either[IList[str], ConsumerError] ``` ### `subscription` ```python subscription(self) -> IList[str] ``` ### `lag` ```python lag(self) -> Either[IMap[TopicPartition, int], KafkaError] ``` ### `memberid` ```python memberid(self) -> str ``` ### `consumer_group_metadata` ```python consumer_group_metadata(self) -> Option[ConsumerGroupMetadata] ``` ### `metrics` ```python 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. ```python AsyncKafkaProducer(config: KafkaConfig) ``` ### `close` ```python close(self) -> None ``` ### `produce` ```python 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` ```python produce_many_bytes(self, topic: str, values: BytesList | IList[bytes] | list[bytes], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> ProduceReceipt ``` ### `produce_many_json` ```python produce_many_json(self, topic: str, values: IList[dict] | list[dict], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> ProduceReceipt ``` ### `produce_many_avro` ```python 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` ```python 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` ```python produce_many_json_sync(self, topic: str, values: IList[dict] | list[dict], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> Either[ProduceReport, KafkaError] ``` ### `flush` ```python flush(self, timeout_ms: Optional[int] = None) -> Either[FlushReport, KafkaError] ``` ### `purge` ```python purge(self, in_queue: bool = True, in_flight: bool = True, blocking: bool = True) -> Either[int, KafkaError] ``` ### `produce_many_records_sync` ```python produce_many_records_sync(self, records: IList[KafkaRecord] | list[KafkaRecord]) -> Either[ProduceReport, KafkaError] ``` ### `init_transactions` ```python init_transactions(self, timeout_ms: Optional[int] = None) -> Either[None, KafkaError] ``` ### `begin_transaction` ```python begin_transaction(self) -> Either[None, KafkaError] ``` ### `commit_transaction` ```python commit_transaction(self, timeout_ms: Optional[int] = None) -> Either[None, KafkaError] ``` ### `abort_transaction` ```python abort_transaction(self, timeout_ms: Optional[int] = None) -> Either[None, KafkaError] ``` ### `produce_many_records_txn` ```python produce_many_records_txn(self, records: IList[KafkaRecord] | list[KafkaRecord]) -> ProduceReceipt ``` ### `send_offsets_to_transaction` ```python send_offsets_to_transaction(self, offsets: IMap[TopicPartition, int], consumer_group_metadata: ConsumerGroupMetadata | Option[ConsumerGroupMetadata], timeout_ms: Optional[int] = None) -> Either[None, KafkaError] ``` ### `metrics` ```python metrics(self) -> dict[str, int] ``` --- ## `BrokerMetadata` ```python BrokerMetadata(node_id: int, host: str, port: int, rack: Optional[str] = None) ``` **Fields:** - `node_id: int` - `host: str` - `port: int` - `rack: Optional[str]` --- ## `BytesBatch` ```python BytesBatch(values: object, offsets: IMap[TopicPartition, int], count: int) ``` **Fields:** - `values: object` - `offsets: IMap[TopicPartition, int]` - `count: int` --- ## `ClusterMetadata` ```python 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` ```python CommitReport(committed: IMap[TopicPartition, int]) ``` **Fields:** - `committed: IMap[TopicPartition, int]` --- ## `ConfigEntry` ```python 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` ```python 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` ```python 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` ```python ConsumerGroupInfo(group_id: str, protocol_type: str) ``` **Fields:** - `group_id: str` - `protocol_type: str` --- ## `ConsumerGroupMember` ```python 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` ```python ConsumerGroupMetadata(group_id: str, coordinator_addr: str) ``` **Fields:** - `group_id: str` - `coordinator_addr: str` --- ## `DeletedRecords` ```python 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. ```python 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). ```python DeliveryReport(topic: str, partition: int, offset: int) ``` **Fields:** - `topic: str` - `partition: int` - `offset: int` --- ## `DeliveryTimeout` --- ## `FlushReport` ```python 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. ```python KafkaAdminClient(config: KafkaConfig) ``` ### `create_topics` ```python 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` ```python delete_topics(self, topics: IList[str] | list[str]) -> Either[IList[TopicResult], AdminError] ``` ### `list_consumer_groups` ```python list_consumer_groups(self) -> Either[IList[ConsumerGroupInfo], AdminError] ``` ### `describe_consumer_groups` ```python describe_consumer_groups(self, group_ids: IList[str] | list[str]) -> Either[IList[ConsumerGroupDetail], AdminError] ``` ### `delete_consumer_groups` ```python delete_consumer_groups(self, group_ids: IList[str] | list[str]) -> Either[IList[TopicResult], AdminError] ``` ### `create_partitions` ```python create_partitions(self, new_partitions: IList[NewPartitions] | list[NewPartitions], validate_only: bool = False, operation_timeout_ms: int = 30000) -> Either[IList[TopicResult], AdminError] ``` ### `list_topics` ```python list_topics(self) -> Either[IList[str], AdminError] ``` ### `describe_cluster` ```python describe_cluster(self, include_authorized_operations: bool = False) -> Either[ClusterMetadata, AdminError] ``` ### `describe_topics` ```python describe_topics(self, topics: IList[str] | list[str] | IList[TopicMetadata] | list[TopicMetadata], include_authorized_operations: bool = False) -> Either[IList[TopicMetadata], AdminError] ``` ### `describe_configs` ```python describe_configs(self, resources: IList[ConfigResource] | list[ConfigResource] | IList[str] | list[str]) -> Either[IMap[str, IMap[str, str]], AdminError] ``` ### `describe_topic_config` ```python describe_topic_config(self, topics: IList[str] | list[str]) -> Either[IMap[str, IMap[str, str]], AdminError] ``` ### `incremental_alter_configs` ```python incremental_alter_configs(self, resources: IList[ConfigResource] | list[ConfigResource], validate_only: bool = False) -> Either[None, AdminError] ``` ### `list_offsets` ```python list_offsets(self, offset_specs: IMap[TopicPartition, OffsetSpec], isolation_level: str = 'read_committed') -> Either[IMap[TopicPartition, ListOffsetsResultInfo], AdminError] ``` ### `delete_records` ```python delete_records(self, offsets: IMap[TopicPartition, int], operation_timeout_ms: int = 30000) -> Either[IMap[TopicPartition, DeletedRecords], AdminError] ``` ### `list_consumer_group_offsets` ```python list_consumer_group_offsets(self, group_id: str, topics: Optional[IList[TopicPartition] | list[TopicPartition]] = None) -> Either[IMap[TopicPartition, int], AdminError] ``` ### `alter_consumer_group_offsets` ```python alter_consumer_group_offsets(self, group_id: str, offsets: IMap[TopicPartition, int]) -> Either[None, AdminError] ``` ### `create_acls` ```python create_acls(self, acls: IList[AclCreation] | list[AclCreation]) -> Either[IList[AclResult], AdminError] ``` ### `describe_acls` ```python 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` ```python delete_acls(self, filters: IList[AclFilter] | list[AclFilter]) -> Either[IList[AclResult], AdminError] ``` --- ## `KafkaBatch` ```python KafkaBatch(records: IList[KafkaRecord], offsets: IMap[TopicPartition, int], count: int) ``` **Fields:** - `records: IList[KafkaRecord]` - `offsets: IMap[TopicPartition, int]` - `count: int` --- ## `KafkaConfig` ```python 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. ```python KafkaConsumer(config: KafkaConsumerConfig) ``` ### `poll_batch` ```python poll_batch(self, max_records: int = 10000, timeout_ms: int = 100) -> Either[KafkaBatch, KafkaError] ``` ### `poll_bytes_batch` ```python poll_bytes_batch(self, max_records: int = 10000, timeout_ms: int = 100) -> Either[BytesBatch, KafkaError] ``` ### `poll_bytes_packed` ```python poll_bytes_packed(self, max_records: int = 10000, timeout_ms: int = 100) -> Either[PackedBytesBatch, KafkaError] ``` ### `poll_bytes_batch_spans` ```python poll_bytes_batch_spans(self, max_records: int = 10000, timeout_ms: int = 100) -> Either[tuple[BytesBatch, list[tuple[int, int, int]]], KafkaError] ``` ### `poll_arrow` ```python 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` ```python store_offsets(self, batch_or_offsets: KafkaBatch | BytesBatch | IMap[TopicPartition, int]) -> Either[None, KafkaError] ``` ### `commit` ```python commit(self, batch: Optional[KafkaBatch | BytesBatch | IMap[TopicPartition, int]] = None) -> Either[CommitReport, KafkaError] ``` ### `commit_offsets` ```python commit_offsets(self, offsets: IMap[TopicPartition, int]) -> Either[CommitReport, KafkaError] ``` ### `subscribe` ```python 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` ```python unsubscribe(self) -> Either[None, ConsumerError] ``` ### `assign` ```python assign(self, partitions: IList[TopicPartition] | list[TopicPartition], start_offsets: Optional[IMap[TopicPartition, int] | dict[TopicPartition, int]] = None) -> Either[None, ConsumerError] ``` ### `unassign` ```python unassign(self) -> Either[None, ConsumerError] ``` ### `assignment` ```python assignment(self) -> IList[TopicPartition] ``` ### `incremental_assign` ```python incremental_assign(self, partitions: IList[TopicPartition] | list[TopicPartition], start_offsets: Optional[IMap[TopicPartition, int] | dict[TopicPartition, int]] = None) -> Either[None, ConsumerError] ``` ### `incremental_unassign` ```python incremental_unassign(self, partitions: IList[TopicPartition] | list[TopicPartition]) -> Either[None, ConsumerError] ``` ### `seek` ```python seek(self, topic_partition: TopicPartition, offset: int) -> Either[None, ConsumerError] ``` ### `seek_to_beginning` ```python seek_to_beginning(self, topic_partition: TopicPartition) -> Either[None, ConsumerError] ``` ### `seek_to_end` ```python seek_to_end(self, topic_partition: TopicPartition) -> Either[None, ConsumerError] ``` ### `rewind` ```python rewind(self, batch: KafkaBatch) -> Either[None, ConsumerError] ``` ### `position` ```python position(self, topic_partition: TopicPartition) -> int ``` ### `committed` ```python committed(self, topic_partition: TopicPartition) -> Option[int] committed(self, topic_partition: IList[TopicPartition] | list[TopicPartition]) -> IMap[TopicPartition, Option[int]] ``` ### `get_watermark_offsets` ```python get_watermark_offsets(self, topic_partition: TopicPartition, timeout_ms: Optional[int] = None) -> Either[tuple[int, int], ConsumerError] ``` ### `offsets_for_times` ```python offsets_for_times(self, tp_timestamps: IMap[TopicPartition, int], timeout_ms: Optional[int] = None) -> Either[IMap[TopicPartition, int], ConsumerError] ``` ### `pause` ```python pause(self, partitions: Optional[IList[TopicPartition] | list[TopicPartition]] = None) -> Either[None, KafkaError] ``` ### `resume` ```python resume(self, partitions: Optional[IList[TopicPartition] | list[TopicPartition]] = None) -> Either[None, KafkaError] ``` ### `list_topics` ```python list_topics(self) -> Either[IList[str], ConsumerError] ``` ### `subscription` ```python subscription(self) -> IList[str] ``` ### `lag` ```python lag(self) -> Either[IMap[TopicPartition, int], KafkaError] ``` ### `memberid` ```python memberid(self) -> str ``` ### `consumer_group_metadata` ```python consumer_group_metadata(self) -> Option[ConsumerGroupMetadata] ``` ### `metrics` ```python metrics(self) -> dict[str, int] ``` ### `close` ```python close(self) -> None ``` --- ## `KafkaConsumerConfig` ```python 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. ```python KafkaProducer(config: KafkaConfig) ``` ### `produce` ```python 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` ```python produce_many_bytes(self, topic: str, values: BytesList | IList[bytes] | list[bytes], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> ProduceReceipt ``` ### `produce_many_json` ```python produce_many_json(self, topic: str, values: IList[dict] | list[dict], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> ProduceReceipt ``` ### `produce_many_avro` ```python produce_many_avro(self, topic: str, values: IList[dict] | list[dict], serde: AvroSerde, keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> ProduceReceipt ``` ### `produce_many` ```python produce_many(self, records: IList[KafkaRecord] | list[KafkaRecord]) -> ProduceReceipt ``` ### `produce_many_bytes_sync` ```python 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` ```python produce_many_json_sync(self, topic: str, values: IList[dict] | list[dict], keys: Optional[BytesList | IList[bytes] | list[bytes]] = None) -> Either[ProduceReport, KafkaError] ``` ### `flush` ```python flush(self, timeout_ms: Optional[int] = None) -> Either[FlushReport, KafkaError] ``` ### `purge` ```python purge(self, in_queue: bool = True, in_flight: bool = True, blocking: bool = True) -> Either[int, KafkaError] ``` ### `init_transactions` ```python init_transactions(self, timeout_ms: Optional[int] = None) -> Either[None, KafkaError] ``` ### `begin_transaction` ```python begin_transaction(self) -> Either[None, KafkaError] ``` ### `commit_transaction` ```python commit_transaction(self, timeout_ms: Optional[int] = None) -> Either[None, KafkaError] ``` ### `abort_transaction` ```python abort_transaction(self, timeout_ms: Optional[int] = None) -> Either[None, KafkaError] ``` ### `produce_many_bytes_txn` ```python 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` ```python produce_many_records_txn(self, records: IList[KafkaRecord] | list[KafkaRecord]) -> ProduceReceipt ``` ### `send_offsets_to_transaction` ```python 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` ```python list_topics(self) -> Either[IList[str], KafkaError] ``` ### `metrics` ```python metrics(self) -> dict[str, int] ``` ### `close` ```python close(self, timeout_ms: Optional[int] = None) -> None ``` --- ## `KafkaProducerConfig` Alias of `KafkaConfig`. ```python 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. ```python 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` ```python ListOffsetsResultInfo(offset: int, timestamp_ms: int) ``` **Fields:** - `offset: int` - `timestamp_ms: int` --- ## `MaxPollIntervalExceeded` --- ## `NewPartitions` ```python NewPartitions(topic: str, total_count: int) ``` **Fields:** - `topic: str` - `total_count: int` --- ## `OffsetSpec` ```python OffsetSpec(kind: str, timestamp_ms: int = -1) ``` **Fields:** - `kind: str` - `timestamp_ms: int` ### `earliest` ```python earliest(cls) -> OffsetSpec ``` ### `latest` ```python latest(cls) -> OffsetSpec ``` ### `max_timestamp` ```python max_timestamp(cls) -> OffsetSpec ``` ### `for_timestamp` ```python for_timestamp(cls, timestamp_ms: int) -> OffsetSpec ``` --- ## `PackedBytesBatch` ```python PackedBytesBatch(packed: bytes, offsets: IMap[TopicPartition, int], count: int) ``` **Fields:** - `packed: bytes` - `offsets: IMap[TopicPartition, int]` - `count: int` ### `to_bytes_list` ```python to_bytes_list(self) -> BytesList ``` --- ## `PartitionMetadata` ```python 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` ```python get(self) -> Either[ProduceReport, KafkaError] ``` ### `wait` ```python wait(self) -> Either[ProduceReport, KafkaError] ``` ### `try_get` ```python try_get(self) -> Optional[Either[ProduceReport, KafkaError]] ``` --- ## `ProduceReport` ```python 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`. ```python 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` ```python 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` ```python TopicPartition(topic: str, partition: int) ``` **Fields:** - `topic: str` - `partition: int` --- ## `TopicPartitionInfo` Alias of `PartitionMetadata`. ```python 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` ```python TopicResult(name: str, error: str = '', code: int = 0, message: Optional[str] = None) ``` **Fields:** - `name: str` - `error: str` - `code: int` - `message: Optional[str]` --- ## `TopicSpec` ```python 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](https://docs.pyrula.io/kafka/guarantees.md) - `OFFSET_BEGINNING: int` - `OFFSET_END: int` - `OFFSET_INVALID: int` - `OFFSET_STORED: int` ## Functions ### `avro_schema_dict` ```python avro_schema_dict(model: type, *, name: Optional[str] = None) -> dict ``` ### `avro_schema_json` ```python avro_schema_json(schema: Any) -> str ``` ### `consume` ```python 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` ```python is_schema_type(t: Any) -> bool ``` ### `json_schema_json` ```python json_schema_json(schema: Any) -> str ```