# Configuration

> KafkaConfig and KafkaConsumerConfig fields, timeouts, and the typed error model.

Source: https://docs.pyrula.io/kafka/configuration/

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

## Producer

```python
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](https://docs.pyrula.io/kafka/producer.md#batching) page.
Set `transactional_id` to enable [transactions](https://docs.pyrula.io/kafka/transactions.md).

| 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

```python
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](https://docs.pyrula.io/kafka/consumer.md), 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](https://docs.pyrula.io/kafka/authentication.md).

## 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`](https://docs.pyrula.io/primitives/either.md), so the error is a value:

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