# Producer

> Produce batches of bytes, JSON, or Avro, and control when you block for delivery.

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

`KafkaProducer` takes a [`KafkaConfig`](https://docs.pyrula.io/kafka/configuration.md). Use it as a context manager.

```python
from pyrula.kafka import KafkaConfig, KafkaProducer

with KafkaProducer(KafkaConfig(bootstrap_servers="localhost:9092")) as producer:
    receipt = producer.produce_many_bytes("events", values=[b"a", b"b"], keys=[b"k1", b"k2"])
    receipt.get().unwrap_or_raise()
```

Leaving the block calls `close()`, which drains every queued record; whatever can't be
delivered fails its receipt with `ProduceError`. Call `producer.purge()` first to drop
instead. `close(timeout_ms=...)` returns at the timeout while the drain continues.

## Producing

`values` and `keys` are parallel lists. Use `produce_many_bytes`, `produce_many_json`,
`produce_many_avro` (see [Formats](https://docs.pyrula.io/kafka/formats.md)), or `produce_many(records)` for
`KafkaRecord`s. One record:

```python
with KafkaProducer(KafkaConfig(bootstrap_servers="localhost:9092")) as producer:
    producer.produce(
        "events", b"v", key=b"k",
        partition=-1,          # default: route by key; keyless records use the sticky partition
        headers={"trace": b"abc123"},
        timestamp_ms=None,     # default: now
    )
```

- **Partition:** `-1` routes by key, `>= 0` pins, below `-1` is a `ConfigError`.
- **Headers:** a dict, or a list of `(name, value)` pairs to repeat a name. A consumed
  record's `header_list` is wire order; `headers` keeps the last copy of each name.
- **Tombstones:** a `None` value is a Kafka null. `b""` is an empty value.
- **Size:** a record over `max_request_bytes` (key, value and headers plus 36 bytes) fails the
  call with `ProduceError` (`MESSAGE_TOO_LARGE`) and none of its records is queued.

## Batching

A batch goes out:

- `linger_ms` (default 5) after its first record; `0` sends as soon as the sender is free;
- at `batch_num_messages` records (10000) or `batch_size_bytes`;
- immediately while `flush()` is pending or the producer is closing.

Keyless records use librdkafka's sticky partitioner: one random partition per topic, kept
for `sticky_partitioning_linger_ms` (default `2 x linger_ms`) or until its leader is lost.
`0` is round-robin. `max_in_flight_requests_per_connection` is 5 with idempotence (more is
a `ConfigError`).

### Connecting

The constructor raises `KafkaConnectionError` if no bootstrap broker answers.
`connect_in_background=True` returns at once instead, as confluent-kafka does. Records wait,
each failed attempt reaches `error_cb` marked `_all_brokers_down` or `_authentication`, and
attempts retry without end. A record that never reaches a broker fails at
`message_timeout_ms` as a `DeliveryTimeout` with `_possibly_persisted=False`.

## Delivery

`produce_many_*` returns a `ProduceReceipt` at once:

```python
receipt = producer.produce_many_bytes("events", values=batch)
# ... other work ...
report = receipt.get().unwrap_or_raise()   # blocks until the broker acks
```

`receipt.get()` / `receipt.wait()` block for that call; `producer.flush(timeout_ms=...)`
waits for everything queued and returns `Ok(FlushReport(flushed, elapsed_ms))`.
`producer.purge()` drops unsent records, failing their receipts as purged, but can't reach
an idempotent batch that already has a sequence number.

Without idempotence, a retried batch can land after a later batch of the same partition.
A stuck partition holds only its own records; a receipt spanning healthy and stuck
partitions resolves when its last record does.

`report.deliveries` has one slot per record in produce order: a
`DeliveryReport(topic, partition, offset)`, or `None` where it didn't land.

```python
values = [b"a", b"b", b"c"]
with KafkaProducer(KafkaConfig(bootstrap_servers="localhost:9092")) as producer:
    receipt = producer.produce_many_bytes("events", values=values)
    report = receipt.get().value
for value, d in zip(values, report.deliveries):
    if d is not None:
        print(value, "->", d.partition, d.offset)
```

### Delivery settings

`acks="all"` (default) waits for the in-sync replicas. `enable_idempotence=True` makes
retries safe and keeps a partition's batches in order. `partitioner` must match across
producers sharing a topic (see [Configuration](https://docs.pyrula.io/kafka/configuration.md)). For exactly-once
across a consume/produce cycle, use [Transactions](https://docs.pyrula.io/kafka/transactions.md).

### When a produce fails

A receipt is `Ok(ProduceReport)` only when every record landed. Otherwise it is
`Err(ProduceError)` carrying the same report:

- `err.report.delivered` / `err.report.failed`: counts for the whole call.
- `err.report.errors`: one entry per failed partition (`topic()`, `partition()`), each with
  its own counts. A failure before routing gives one entry with no partition.

An idempotent producer stops a partition at its first failure, so resend each failed
partition's records from `entry.report.delivered` on; resending the whole batch duplicates
what landed.

| Situation | Error | May have landed? |
|---|---|---|
| Sent, no usable answer by `message_timeout_ms` or the last retry | `DeliveryTimeout` | yes |
| Never reached a broker (no leader, unreachable) | `DeliveryTimeout`, `_possibly_persisted=False` | no, safe to resend |
| Refused before anything was sent (unknown topic or partition) | `ProduceError` | no |
| `MessageTooLarge` from the broker | `ProduceError` (code 10); every record in that partition's batch fails | no |
| Non-idempotent, last attempt lost to the connection | `ProduceError` with the transport error | yes |
| Producer fenced (`ProducerFenced`, `InvalidProducerEpoch`, out-of-order sequence on the oldest batch) | fatal: `error_cb` gets one `KafkaError` marked `_producer_fatal`, then every produce raises `ProduceError` | - |

A fenced producer can't recover; create a new one.

`MessageTooLarge` means the batch is over the topic's `max.message.bytes` (the broker's
`message.max.bytes` when the topic sets none). Raise the limit on the topic or the broker,
or send smaller records. `max_request_bytes` is the client's own cap and refuses a record
before it is sent.

A topic that keeps answering
`UnknownTopicOrPartition`, or a partition missing from metadata, fails after a fixed 5 s
with a `ProduceError` naming it.

## Metrics

`producer.metrics()` returns a snapshot dict:

| Key | Meaning |
|---|---|
| `msg_cnt` / `msg_size` | records (and bytes) queued or in flight |
| `tx` / `tx_bytes` | records (and estimated bytes) acked since start |
| `txerrs` | records that failed terminally |
| `txretries` | produce retries scheduled |
| `producer_id` / `producer_epoch` | idempotent id (`-1` before the first send) and epoch |

Set `statistics_interval_ms` and `stats_cb` for periodic snapshots.

## Threads

A `KafkaProducer` is safe from any thread. Records one thread produces to a partition keep
its order. Blocking calls release the GIL.

- One transactional call runs at a time. A different call during one in progress returns an
  error marked `_conflicting_use`; the same call from another thread, `_prev_in_progress`.
  Both are retriable.
- `error_cb`, `stats_cb` and the logger run on the producer's worker thread. A blocking
  producer call from one raises `ProduceError` marked `_conflicting_use`; hand the work to
  another thread. The confluent shim serves callbacks on the polling thread instead.

## Async

`AsyncKafkaProducer` has the same surface with `await` and `async with`. Concurrent
coroutines don't keep their order.

```python
from pyrula.kafka import AsyncKafkaProducer

async with AsyncKafkaProducer(config) as producer:
    await producer.produce_many_bytes("events", values=batch)
    await producer.flush()
```

Each blocking call holds a thread of the loop's default executor until it returns. Bound
the timeouts or enlarge the pool with `loop.set_default_executor()`.