Configuration

KafkaConfig and KafkaConsumerConfig are typed classes. bootstrap_servers is the only required field. Every field and default is in the API reference.

Producer

from pyrula.kafka import KafkaConfig

KafkaConfig(
    bootstrap_servers="localhost:9092",
    acks="all",                  # all | -1 | 1 | 0
    enable_idempotence=None,     # off; on when transactional_id is set
    compression="none",         # none | gzip | snappy | lz4 | zstd
    partitioner="murmur2",       # default
    linger_ms=5,                 # 0 sends at once
    batch_size_bytes=131_072,
    batch_num_messages=10_000,
    max_request_bytes=1_048_576,
    raw_connections_per_broker=8,
    message_timeout_ms=None,     # 120 s; 60 s (the txn timeout) with transactional_id
    request_timeout_ms=30_000,
)

message_timeout_ms bounds how long a message can sit before delivery fails. request_timeout_ms bounds one broker request. Transient rejections (leader catching up, leader move, not enough in-sync replicas) are retried until the produce lands or message_timeout_ms runs out; message_send_max_retries defaults to 2147483647, so the time bound ends a produce. Backoff starts at retry_backoff_ms (100 ms) and doubles to 1 s.

Batching, routing and connecting are on the Producer page. Set transactional_id to enable transactions.

Field	Notes
partitioner	"murmur2" (Java, Kafka Streams, kafka-python, aiokafka) or "consistent" (librdkafka’s CRC32). Producers sharing a topic must agree, or the same key lands on different partitions. Anything else is a ConfigError.
max_request_bytes	Caps a whole produce request (1 MiB); must be >= batch_size_bytes or construction raises ConfigError.
coalesce_produce_requests	False: one partition per request, which maximizes overlap on a fast local broker. True packs partitions sharing a broker into one request; helps on high-latency or multi-broker links.
raw_connections_per_broker	Connections per leader for non-transactional produce (8). Each partition sticks to one connection. 1 shares one socket (roughly 25-30% slower at acks=1).
queue_buffering_max_messages, queue_buffering_max_kbytes	Bound what a producer holds, from accept until the send is acked or failed. A produce that would exceed either is refused whole with QueueFull (BufferError through the confluent shim).

A transactional producer ignores coalesce_produce_requests and raw_connections_per_broker: it uses one connection and one request per broker.

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,
    heartbeat_interval_ms=3_000,
    request_timeout_ms=30_000,
    max_poll_interval_ms=300_000,
    prefetch_max_bytes=67_108_864,
)

Offsets move when you commit, not on a timer. isolation_level="read_committed" hides aborted transactional records. subscribe() never creates a topic.

Timeouts

• session_timeout_ms is the broker’s liveness window for the group member.
• request_timeout_ms bounds one broker request: heartbeat, commit, fetch, LeaveGroup, FindCoordinator, metadata and watermark lookups. get_watermark_offsets and offsets_for_times take their own timeout_ms.
• Opening a connection has its own 5 s bound per bootstrap address.
• rebalance_timeout_ms bounds a join, a rejoin waiting on a partition leader, and a grouped assign looking up its partitions.
• Commits and fetches wait out a loading or moved coordinator for request_timeout_ms plus 30 s.

A join or rejoin the coordinator cannot answer is retried with backoff (100 ms doubling to 2 s) indefinitely, as librdkafka does. Trouble reaches error_cb as a retriable KafkaConnectionError marked _transport (unreachable) or _timed_out (not ready), at most every 30 s. State methods such as assignment() wait for a rejoin in flight for at most rebalance_timeout_ms, then raise a ConsumerError marked _timed_out.

A partition whose start fails (a lookup gets no answer, or a manual assign names a missing topic or partition) stays assigned and is retried in the background, backing off from 1 s to 10 s. It is neither read nor committed until a retry succeeds. The next poll reports one non-fatal error naming the partition, and metrics() and stats_cb report the count as pending_starts.

Fetch tuning

Field	Default	What it does
max_poll_records	500	Records per poll_batch.
max_partition_fetch_bytes	1 MiB	Per-partition cap per fetch.
fetch_max_bytes	50 MiB	Cap across partitions per fetch.
fetch_min_bytes	1	Bytes the broker accumulates before answering.
fetch_wait_max_ms	500	Longest the broker holds a fetch.
fetch_mode	"auto"	per_partition, or per_broker (one request per leader). auto picks per_broker above 4 partitions.
prefetch_max_bytes	64 MiB	Cap on fetched-but-unpolled data.
partition_assignment_strategy	"range"	range, roundrobin or cooperative-sticky.

For one or a few hot partitions, raise max_partition_fetch_bytes to 4-16 MiB (130k to 138k rec/s at 1 KB locally; worse past ~16 MiB). per_broker wins from 6-8 partitions up. Lower prefetch_max_bytes to cap memory.

A librdkafka-only key in extra is rejected with ConfigError. SASL and TLS fields are in Authentication.

Worker threads

Each producer, consumer and admin client runs its own pool of two worker threads, stopped by close(). PYRULA_KAFKA_WORKER_THREADS sets the count as a positive integer, read once per process at first client creation. Raise it only if one client’s callbacks keep both workers busy.

Logging

Native warnings and errors reach the pyrula.kafka logger. Pass logger= on a client to route its messages elsewhere, or set pyrula.kafka to DEBUG for group-join details.

Errors

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

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

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

DeliveryTimeout means message_timeout_ms ended a send that went out, so the broker may have received it. Catch QueueFull to react to backpressure, or KafkaError for anything.