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=5,                 # default (librdkafka parity)
    batch_size_bytes=131_072,
    max_request_bytes=1_048_576,
    coalesce_produce_requests=False,
    raw_connections_per_broker=8,
    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. max_request_bytes caps the whole produced request (parity with librdkafka’s message.max.bytes / Java’s max.request.size, default 1 MiB) and must be >= batch_size_bytes; construction raises ConfigError otherwise. Set transactional_id to enable transactions.

coalesce_produce_requests controls wire-request packing granularity per broker and defaults to False: each produce request carries a single partition’s fragment, which maximizes cross-partition overlap and matches the throughput of independent per-partition sends. Set it to True to pack partitions that share a broker into multi-partition requests up to max_request_bytes — fewer wire requests, which helps on high-latency or multi-broker links where request count (not local packing) is the bottleneck — but doing so can reduce cross-partition overlap and lower throughput on a fast/local broker (this is why it is opt-in rather than the default). Duplicate-prevention and sequence-continuity guarantees are identical in both modes; only the number of partitions riding in one wire request changes.

raw_connections_per_broker (default 8) sets how many TCP connections the raw (non-default acks) and idempotent produce paths open per broker leader. A single shared connection serializes every partition’s requests onto one socket, which costs roughly 25-30% throughput on the raw acks=1 path versus spreading partitions across several connections. Each (topic, partition) is assigned to exactly one connection by a deterministic hash — sticky for the life of the producer — so raising this only adds cross-partition parallelism: a single partition’s requests always travel the same connection in the same order they do today, and idempotent sequence numbers stay continuous exactly as before. Set it to 1 to reproduce the original single-shared- connection behavior. The trade-off is more broker connections and file descriptors per producer; must be >= 1 or construction raises ConfigError.

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="auto",           # auto | per_partition | per_broker
    max_poll_records=500,
    session_timeout_ms=10_000,
    max_poll_interval_ms=300_000,
    prefetch_max_bytes=1_073_741_824,  # 1 GiB prefetch-memory safety-net
)

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 auto picks per_partition (one fetch per assigned partition) when the first assignment has four or fewer partitions, and per_broker above that. 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). You can pin either mode explicitly; auto suits almost every consumer.

prefetch_max_bytes (default 1 GiB) is a memory safety-net, not a throughput knob. Each partition prefetches a bounded number of batches ahead to hide broker latency; at very high partition counts the total (partitions × depth × max_partition_fetch_bytes) can grow large, so this budget caps it by trimming the per-partition prefetch depth (never below one batch per partition). At normal partition counts it never engages. Leave it at the default unless you assign hundreds of partitions per consumer and want a tighter memory ceiling.

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.