A producer may be shared by any number of fibers in the Eio domain that created it (via create's ~sw). It is not domain-safe: sharing one t across multiple Eio domains is unsupported and unguarded.

create cfg ~sw creates a producer and starts delivery and poll fibers in sw. When sw is cancelled the fibers stop and the producer is closed.

Every operation below returns Error Kafka_error.Destroy (or, for produce_await, a promise already resolved to it) once close has been called, instead of touching the underlying (possibly destroyed) handle. close itself also resolves any produce_await promises still awaiting a delivery receipt to Error Kafka_error.Destroy, so a fiber awaiting one cannot hang forever past shutdown.

create_topic t ~topic_name ~partitions ~replication_factor creates a topic via librdkafka's admin API, reusing this producer's handle. Treats an already-existing topic as success. Does not leak Kafka_raw.kafka_handle the way an earlier raw_handle accessor did — Kafka_raw is internal to kafka-eio-core and not part of this package's public surface.

Enqueue a message and return immediately. No delivery confirmation. value = None sends a Kafka tombstone (a NULL payload — the delete marker for a key on a compacted topic), distinct from Some Bytes.empty, a genuine zero-length value. value is required (not optional) so a caller must choose explicitly, rather than a forgotten ~value silently sending a tombstone. ~key, if given, is used exactly as passed — including a genuine zero-length key, which Kafka partitions differently (hashed) from no key at all (round- robin/sticky); omitting ~key entirely means no key. A header's value is string option: None sends a NULL-valued header, distinct from Some "".

Can return Error Kafka_error.Queue_full if librdkafka's local send queue is full — this call does not block or retry for you. Callers producing at a high rate must handle it themselves (drop, retry after a delay, or apply their own backpressure); there is currently no Eio-native blocking/backpressure variant of produce.

The trailing unit is required by OCaml's optional-argument erasure rules.

Enqueue a message and return a promise that resolves when the broker acknowledges delivery (or reports an error). value = None sends a tombstone, and ~key/~headers behave exactly as in produce. Can resolve to Error Kafka_error.Queue_full the same way produce can return it — see produce. The trailing unit is required by OCaml's optional-argument erasure rules.

Block until all enqueued messages have been delivered.

A transactional-API call (begin/commit/abort/send-offsets) itself failed. is_fatal/is_retriable/requires_abort are librdkafka's per-error-instance flags — not derivable from error alone — telling the caller whether the producer must be retired, whether the same call can be retried, and whether the transaction had to be aborted (already done by with_transaction when this is true).

with_transaction t ?consumer_offsets f runs f inside a Kafka transaction. Commits on Ok, aborts on Error or exception. Requires delivery_mode = Exactly_once.

consumer_offsets, if given, is (consumer_handle, offsets) where offsets are the exact (topic, partition, offset-of-last-processed- message) tuples f processed from that consumer — typically built from the consumed messages' topic/partition/offset fields. Only these offsets are committed as part of the transaction; the consumer's current assignment or position is never read, so a transaction can never advance past a message f did not actually process.