Kafka

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

Types

`Acl`

Fields:

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

`AclCreation`

Fields:

resource_type: AclResourceType
resource_name: str
principal: str
host: str
operation: AclOperation
permission_type: AclPermissionType

`AclFilter`

Fields:

resource_type: AclResourceType
resource_name: Optional[str]
principal: Optional[str]
host: Optional[str]
operation: AclOperation
permission_type: AclPermissionType

`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

`AclPermissionType`

Fields:

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

`AclResource`

Fields:

resource_type: AclResourceType
resource_name: str
acls: Any

`AclResourceType`

Fields:

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

`AclResult`

Fields:

error: str

`AdminError`

`AlterConfigOpType`

Fields:

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

`ArrowBatch`

Rust-decoded Arrow RecordBatch handle (from poll_arrow).

Exports the Arrow PyCapsule C-stream interface, so pyarrow/polars/duckdb/ deltalake consume it zero-copy. Re-consumable: each arrow_c_stream call clones the (Arc-backed) batch, so re-writing it on a sink retry is safe.

Fields:

num_rows: int

`ArrowPollResult`

Result of poll_arrow: the decoded batch plus consumed offsets.

Fields:

batch: ArrowBatch
offsets: Any

`AsyncKafkaConsumer`

`close`

close(self) -> None

`poll_batch`

poll_batch(self, max_records: int = 500, timeout_ms: int = 500) -> Any

`poll_bytes_batch`

poll_bytes_batch(self, max_records: int = 500, timeout_ms: int = 500) -> Any

`consume`

consume(self, max_records: int = 500, timeout_ms: int = 500) -> Any

`store_offsets`

store_offsets(self, batch_or_offsets: Any) -> Any

`commit`

commit(self, batch: Optional[Any] = None) -> Any

`commit_offsets`

commit_offsets(self, offsets: Any) -> Any

`subscribe`

subscribe(self, topics: Any, on_assign: Optional[Any] = None, on_revoke: Optional[Any] = None, on_lost: Optional[Any] = None) -> Any

`unsubscribe`

unsubscribe(self) -> Any

`assign`

assign(self, partitions: Any) -> Any

`unassign`

unassign(self) -> Any

`assignment`

assignment(self) -> Any

`seek`

seek(self, topic_partition: Any, offset: int) -> Any

`seek_to_beginning`

seek_to_beginning(self, topic_partition: Any) -> Any

`seek_to_end`

seek_to_end(self, topic_partition: Any) -> Any

`committed`

committed(self, topic_partition: Any) -> Any

`position`

position(self, topic_partition: Any) -> int

`get_watermark_offsets`

get_watermark_offsets(self, topic_partition: Any) -> Any

`offsets_for_times`

offsets_for_times(self, tp_timestamps: Any) -> Any

`pause`

pause(self, partitions: Optional[Any] = None) -> Any

`resume`

resume(self, partitions: Optional[Any] = None) -> Any

`list_topics`

list_topics(self) -> Any

`subscription`

subscription(self) -> Any

`lag`

lag(self) -> Any

`memberid`

memberid(self) -> str

`consumer_group_metadata`

consumer_group_metadata(self) -> Any

`AsyncKafkaProducer`

`close`

close(self) -> None

`produce`

produce(self, topic: str, value: bytes, key: Optional[bytes] = None) -> Any

`produce_many_bytes`

produce_many_bytes(self, topic: str, values: Any, keys: Optional[Any] = None) -> Any

`produce_many_json`

produce_many_json(self, topic: str, values: Any, keys: Optional[Any] = None) -> Any

`produce_many_avro`

produce_many_avro(self, topic: str, values: Any, serde: Any, keys: Optional[Any] = None) -> Any

`produce_many_bytes_sync`

produce_many_bytes_sync(self, topic: str, values: Any, keys: Optional[Any] = None) -> Any

`produce_many_json_sync`

produce_many_json_sync(self, topic: str, values: Any, keys: Optional[Any] = None) -> Any

`flush`

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

`purge`

purge(self) -> Any

`produce_many_records_sync`

produce_many_records_sync(self, records: Any) -> Any

`produce_many_records_txn`

produce_many_records_txn(self, records: Any) -> Any

`send_offsets_to_transaction`

send_offsets_to_transaction(self, offsets: Any, consumer_group_metadata: Any) -> Any

`BrokerMetadata`

Fields:

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

`BytesBatch`

Fields:

values: object
offsets: Any
count: int

`ClusterMetadata`

Fields:

brokers: Any
controller_id: int
topics: Any
cluster_id: Optional[str]

`CommitError`

`CommitReport`

Fields:

committed: Any

`ConfigEntry`

Fields:

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

`ConfigError`

`ConfigResource`

Fields:

resource_type: ResourceType
name: str
set_config: Any
incremental_configs: Any

`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`

Fields:

group_id: str
state: str
protocol_type: str
protocol: str
members: Any

`ConsumerGroupInfo`

Fields:

group_id: str
protocol_type: str

`ConsumerGroupMember`

Fields:

member_id: str
client_id: str
host: str
assignment: Any

`ConsumerGroupMetadata`

Fields:

group_id: str
coordinator_addr: str

`DeletedRecords`

Fields:

low_watermark: int

`DeliveryReport`

Fields:

topic: str
partition: int
offset: int

`DeliveryTimeout`

`FetchBufferFull`

`FlushReport`

Fields:

flushed: int
elapsed_ms: int

`KafkaAdminClient`

`create_topics`

create_topics(self, specs: Any) -> Any

`delete_topics`

delete_topics(self, topics: Any) -> Any

`list_consumer_groups`

list_consumer_groups(self) -> Any

`describe_consumer_groups`

describe_consumer_groups(self, group_ids: Any) -> Any

`delete_consumer_groups`

delete_consumer_groups(self, group_ids: Any) -> Any

`create_partitions`

create_partitions(self, new_partitions: Any, validate_only: bool = False, operation_timeout_ms: int = 30000) -> Any

`list_topics`

list_topics(self) -> Any

`describe_cluster`

describe_cluster(self, include_authorized_operations: bool = False) -> Any

`describe_topics`

describe_topics(self, topics: Any, include_authorized_operations: bool = False) -> Any

`describe_configs`

describe_configs(self, resources: Any) -> Any

`describe_topic_config`

describe_topic_config(self, topics: Any) -> Any

`incremental_alter_configs`

incremental_alter_configs(self, resources: Any, validate_only: bool = False) -> Any

`list_offsets`

list_offsets(self, offset_specs: Any, isolation_level: str = 'read_committed') -> Any

`delete_records`

delete_records(self, offsets: Any, operation_timeout_ms: int = 30000) -> Any

`KafkaBatch`

Fields:

records: Any
offsets: Any
count: int

`KafkaConfig`

Fields:

bootstrap_servers: str
client_id: str
acks: str | int
compression: str
linger_ms: int
batch_size_bytes: int
enable_idempotence: bool
message_timeout_ms: 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[Any]
extra: Any

`KafkaConnectionError`

`KafkaConsumer`

`poll_batch`

poll_batch(self, max_records: int = 10000, timeout_ms: int = 100) -> Any

`poll_bytes_batch`

poll_bytes_batch(self, max_records: int = 10000, timeout_ms: int = 100) -> Any

`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) -> Any

`store_offsets`

store_offsets(self, batch_or_offsets: Any) -> Any

`commit`

commit(self, batch: Optional[Any] = None) -> Any

`commit_offsets`

commit_offsets(self, offsets: Any) -> Any

`subscribe`

subscribe(self, topics: Any, on_assign: Optional[Any] = None, on_revoke: Optional[Any] = None, on_lost: Optional[Any] = None) -> Any

`unsubscribe`

unsubscribe(self) -> Any

`assign`

assign(self, partitions: Any) -> Any

`unassign`

unassign(self) -> Any

`assignment`

assignment(self) -> Any

`incremental_assign`

incremental_assign(self, partitions: Any) -> Any

`incremental_unassign`

incremental_unassign(self, partitions: Any) -> Any

`seek`

seek(self, topic_partition: Any, offset: int) -> Any

`seek_to_beginning`

seek_to_beginning(self, topic_partition: Any) -> Any

`seek_to_end`

seek_to_end(self, topic_partition: Any) -> Any

`position`

position(self, topic_partition: Any) -> int

`committed`

committed(self, topic_partition: Any) -> Any

`get_watermark_offsets`

get_watermark_offsets(self, topic_partition: Any) -> Any

`offsets_for_times`

offsets_for_times(self, tp_timestamps: Any) -> Any

`pause`

pause(self, partitions: Optional[Any] = None) -> Any

`resume`

resume(self, partitions: Optional[Any] = None) -> Any

`list_topics`

list_topics(self) -> Any

`subscription`

subscription(self) -> Any

`lag`

lag(self) -> Any

`memberid`

memberid(self) -> str

`consumer_group_metadata`

consumer_group_metadata(self) -> Any

`close`

close(self) -> None

`KafkaConsumerConfig`

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
partition_assignment_strategy: str
rebalance_timeout_ms: 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_mode: str
security_protocol: str
sasl_mechanism: str
sasl_username: Optional[str]
sasl_password: Optional[str]
oauth_cb: Optional[Any]
extra: Any

`KafkaError`

`KafkaProducer`

`produce`

produce(self, topic: str, value: bytes, key: Optional[bytes] = None) -> ProduceReceipt

`produce_many_bytes`

produce_many_bytes(self, topic: str, values: Any, keys: Optional[Any] = None) -> ProduceReceipt

`produce_many_json`

produce_many_json(self, topic: str, values: Any, keys: Optional[Any] = None) -> ProduceReceipt

`produce_many_avro`

produce_many_avro(self, topic: str, values: Any, serde: Any, keys: Optional[Any] = None) -> ProduceReceipt

`produce_many`

produce_many(self, records: Any) -> ProduceReceipt

`produce_many_bytes_sync`

produce_many_bytes_sync(self, topic: str, values: Any, keys: Optional[Any] = None) -> Any

`produce_many_json_sync`

produce_many_json_sync(self, topic: str, values: Any, keys: Optional[Any] = None) -> Any

`flush`

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

`purge`

purge(self) -> Any

`init_transactions`

init_transactions(self) -> Any

`begin_transaction`

begin_transaction(self) -> Any

`commit_transaction`

commit_transaction(self) -> Any

`abort_transaction`

abort_transaction(self) -> Any

`produce_many_bytes_txn`

produce_many_bytes_txn(self, topic: str, values: Any, keys: Optional[Any] = None, partition: int = 0) -> Any

`produce_many_records_txn`

produce_many_records_txn(self, records: Any) -> Any

`send_offsets_to_transaction`

send_offsets_to_transaction(self, offsets: Any, consumer_group_metadata: Any) -> Any

`list_topics`

list_topics(self) -> Any

`close`

close(self) -> None

`KafkaRecord`

Native (Rust) record. Frozen. partition/offset are 0/-1 on the produce path.

key/timestamp_ms read back as Option (Some/Nothing); headers as IMap[str, bytes]. Ctor accepts Some(x)/Nothing OR raw values/None for key and timestamp_ms, and IMap or plain dict for headers. key/headers/timestamp_ms/value materialize lazily on first access (cached).

Fields:

topic: str
value: bytes
partition: int
offset: int
key: object
headers: Any
timestamp_ms: object

`ListOffsetsResultInfo`

Fields:

offset: int
timestamp_ms: int

`NewPartitions`

Fields:

topic: str
total_count: int

`OffsetSpec`

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

`PartitionMetadata`

Fields:

partition_index: int
leader_id: int
replicas: Any
isr: Any

`PollError`

`PollTimeoutExceeded`

`ProduceError`

`ProduceReceipt`

`get`

get(self) -> Any

`wait`

wait(self) -> Any

`ProduceReport`

Fields:

accepted: int
delivered: int
failed: int
errors: list

`QueueFull`

`RebalanceError`

`ResourceType`

Fields:

UNKNOWN: int
TOPIC: int
BROKER: int

`TopicMetadata`

Fields:

name: str
partitions: Any
is_internal: bool
error: Optional[str]

`TopicPartition`

Fields:

topic: str
partition: int

`TopicResult`

Fields:

name: str
error: str

`TopicSpec`

Fields:

name: str
num_partitions: int
replication_factor: int
config: Any

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