Configuration

Config is two typed dataclasses, not a string-keyed dict. bootstrap_servers is the only required field; everything else has a default that matches the runtime.

Producer

from pyrula.kafka import KafkaConfig

KafkaConfig(
    bootstrap_servers="localhost:9092",
    acks="all",                  # default
    enable_idempotence=False,
    compression="none",
    linger_ms=10,
    batch_size_bytes=131_072,
    message_timeout_ms=120_000,
    request_timeout_ms=30_000,
)

Every networked operation has an explicit timeout. message_timeout_ms bounds how long a message can sit before delivery fails; request_timeout_ms bounds a single broker request. Set transactional_id to enable transactions.

Consumer

from pyrula.kafka import KafkaConsumerConfig

KafkaConsumerConfig(
    bootstrap_servers="localhost:9092",
    topics=["events"],
    group_id="my-group",
    enable_auto_commit=False,    # default: you commit
    auto_offset_reset="earliest",
    isolation_level="read_committed",
    fetch_mode="per_partition",  # or "per_broker"
    max_poll_records=500,
    session_timeout_ms=10_000,
    max_poll_interval_ms=300_000,
)

enable_auto_commit is off by default, which is the safe choice: offsets move when you commit, not on a timer. isolation_level="read_committed" hides aborted transactional records; set "read_uncommitted" to skip that filter.

SASL and TLS fields (security_protocol, sasl_mechanism, oauth_cb, ssl_*) are covered in authentication.

Tuning consume throughput

Each fetch is one broker round-trip, and max_partition_fetch_bytes (default 1 MB, the same as confluent) caps how many records come back per round-trip. For a throughput-bound consumer reading one or a few high-volume partitions, raising it to 4-16 MB pulls more records per round-trip and cuts the number of fetches. In a 100k-record benchmark with 1 KB records on a local broker, that moved consume from roughly 130k to 138k records per second. Past about 16 MB it reverses as the batches get oversized. It costs more buffer memory per partition, so raise it for a measured need, not by default.

Fetches are pipelined: the next fetch is issued before the current batch is handed to your poll, so its round-trip overlaps your processing. That helps most over real network latency and very little on a local broker.

fetch_mode controls how those fetches are issued when a consumer holds many partitions. The default per_partition runs one fetch per assigned partition. per_broker coalesces all partitions led by the same broker into a single fetch request, the way librdkafka does, which scales better as the partition count climbs (it matches or passes confluent past about a dozen partitions). It is opt-in for now; leave it on per_partition unless you consume many partitions per process and have measured the difference.

The extra field is a validation guard: unknown keys are checked against a ban list and rejected with a ConfigError if they’re librdkafka-only options that the Rust client cannot use. It is not a tuning escape hatch and the values are not forwarded.

Errors

Calls return an Either, so the error is a value you inspect:

result = consumer.poll_batch()
if result.is_err():
    err = result.error

The error types form a hierarchy under KafkaError:

Error	When
ConfigError	Bad configuration. Not retriable.
KafkaConnectionError	Can’t reach the broker.
ProduceError	A produce failed. QueueFull and DeliveryTimeout subclass it.
ConsumerError / PollError	A consume or poll failed. FetchBufferFull and PollTimeoutExceeded subclass ConsumerError.
CommitError	An offset commit failed.
RebalanceError	A group rebalance failed.
AdminError	An admin operation failed.

ConfigError is the only non-retriable one, since retrying bad config won’t help. The rest are worth a retry. Catch a specific subclass like QueueFull when you want to react to backpressure, or KafkaError to handle anything.