Producer

KafkaProducer takes a KafkaConfig. Use it as a context manager.

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), or produce_many(records) for KafkaRecords. One record:

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:

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.

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). For exactly-once across a consume/produce cycle, use Transactions.

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.

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().