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

Fields:

• num_rows: int

---

ArrowPollResult

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) -> AsyncIterator[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

metrics

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

---

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

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

metrics

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

---

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

list_consumer_group_offsets

list_consumer_group_offsets(self, group_id: str, topics: Optional[Any] = None) -> Any

alter_consumer_group_offsets

alter_consumer_group_offsets(self, group_id: str, offsets: Any) -> Any

create_acls

create_acls(self, acls: Any) -> Any

describe_acls

describe_acls(self, resource_type: int = 0, resource_name: Optional[str] = None, principal: Optional[str] = None, host: Optional[str] = None, operation: int = 0, permission_type: int = 0) -> Any

delete_acls

delete_acls(self, filters: Any) -> 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
• max_request_bytes: int
• coalesce_produce_requests: bool
• raw_connections_per_broker: 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]
• ssl_ca_location: Optional[str]
• ssl_certificate_location: Optional[str]
• ssl_key_location: Optional[str]
• error_cb: Optional[Any]
• logger: Optional[Any]
• statistics_interval_ms: int
• stats_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_bytes_packed

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

poll_bytes_batch_spans

poll_bytes_batch_spans(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

metrics

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

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_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[Any]
• ssl_ca_location: Optional[str]
• ssl_certificate_location: Optional[str]
• ssl_key_location: Optional[str]
• error_cb: Optional[Any]
• logger: Optional[Any]
• statistics_interval_ms: int
• stats_cb: Optional[Any]
• extra: Any

---

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

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

metrics

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

test_connections_opened

test_connections_opened(self) -> int

test_multipartition_requests

test_multipartition_requests(self) -> int

close

close(self) -> None

---

KafkaProducerConfig

Alias of KafkaConfig.

Fields:

• bootstrap_servers: str
• client_id: str
• acks: str | int
• compression: str
• linger_ms: int
• batch_size_bytes: int
• max_request_bytes: int
• coalesce_produce_requests: bool
• raw_connections_per_broker: 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]
• ssl_ca_location: Optional[str]
• ssl_certificate_location: Optional[str]
• ssl_key_location: Optional[str]
• error_cb: Optional[Any]
• logger: Optional[Any]
• statistics_interval_ms: int
• stats_cb: Optional[Any]
• extra: Any

---

KafkaRecord

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

---

PackedBytesBatch

Fields:

• packed: bytes
• offsets: Any
• count: int

to_bytes_list

to_bytes_list(self) -> Any

---

PartitionMetadata

Fields:

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

---

PollError

---

PollTimeoutExceeded

---

ProduceError

---

ProduceReceipt

get

get(self) -> Any

wait

wait(self) -> Any

try_get

try_get(self) -> Any

---

ProduceReport

Fields:

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

---

QueueFull

---

RebalanceError

---

ResourceType

Fields:

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

---

TopicDescription

Alias of TopicMetadata.

Fields:

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

---

TopicMetadata

Fields:

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

---

TopicPartition

Fields:

• topic: str
• partition: int

---

TopicPartitionInfo

Alias of PartitionMetadata.

Fields:

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

---

TopicResult

Fields:

• name: str
• error: str

---

TopicSpec

Fields:

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

Constants

• 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