# Client guarantees

> What the Kafka client guarantees, capability by capability, and the brokers it supports.

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

`pyrula-kafka` has no dependency on the workflow or agent packages. `yes` means the
guarantee holds; `partial` means it holds within the boundary under Limits; `n/a` means the
capability has no meaning for that surface.

The `compat.confluent` row is the [migration shim](https://docs.pyrula.io/kafka/confluent-compat.md): the native
rows' guarantees, observed through confluent-kafka's API.

| Surface | delivery | ordering | exit | offset commit | transactions | backpressure | rebalance |
|---|---|---|---|---|---|---|---|
| `producer` | yes | yes | yes | n/a | yes | yes | n/a |
| `consumer` | yes | n/a | yes | yes | yes | yes | yes |
| `compat.confluent` | partial | yes | yes | yes | yes | yes | yes |

## Column definitions

- **delivery:** A receipt resolves only after the broker acks the record per `acks` (default `all`).
  - A record the broker refuses fails its receipt with a typed error, never silently.
  - Receipts resolve without the caller polling.
- **ordering:** Records produced to one partition by one producer land in produce order, with `enable_idempotence=True`.
- **exit:** Closing settles what is in flight.
  - Producer: leaving the context manager (or `close()`) drains every queued record and waits for the actor to stop. `purge()` before close is the explicit drop.
  - Consumer: wherever a partition leaves (`close()`, `unsubscribe()`, `unassign()`, a revoke), an offset staged with `store_offsets()` is committed regardless of `enable_auto_commit`. The fetch position is committed too only when it is `True`.
- **offset commit:** Nothing is committed that was not polled.
  - Auto-commit is off by default (the `compat.confluent` shim defaults it on, as confluent-kafka does).
  - `commit(batch)` commits exactly that batch's offsets, synchronously, and reports failure.
  - `store_offsets` stages offsets for the next commit.
- **transactions:** `begin_transaction` / `produce_*_txn` / `send_offsets_to_transaction` / `commit_transaction` land as one unit.
  - `abort_transaction` discards both output and offsets.
  - A failure from a fenced producer is classified fatal and says so.
  - A `read_committed` consumer (the default) never sees an aborted record, including one that follows another transaction's commit marker.
- **backpressure:** A full queue refuses new work; it never drops accepted work.
  - A produce that would exceed `queue_buffering_max_messages` / `queue_buffering_max_kbytes` is refused whole with `QueueFull`, before any of its records is accepted.
  - `purge()` and `flush()` release capacity.
  - Consumer: `pause`/`resume` stop and restart fetch without losing position.
- **rebalance:** Callbacks fire in order (`on_revoke`/`on_lost` before partitions leave, `on_assign` after they arrive), with the consumer released.
  - A callback may `commit`, `store_offsets`, `seek`, `position`, `committed`, `pause`/`resume` and inspect the consumer it was raised from.
  - It may acknowledge the phase with `assign`/`unassign` (eager) or `incremental_assign`/`incremental_unassign` (cooperative-sticky), naming exactly the partitions it received.
  - `subscribe`, `unsubscribe`, `close`, any `poll_*`, and `commit` inside `on_lost` are refused with `RebalanceError`; the rebalance still completes.
  - Offsets staged for revoked partitions are committed before they leave, regardless of `enable_auto_commit` (the same rule as `close()`, `unsubscribe()`, `unassign()` and a replacing `subscribe()`). The fetch position commits too only when it is `True`.
  - Under cooperative-sticky assignment, unrevoked partitions keep flowing.
  - A consumer evicted for exceeding `max_poll_interval_ms` (callback time included) gets `MaxPollIntervalExceeded` from `poll_batch`, once.

## Install and tested versions

All surfaces: built in. Supported brokers: Kafka 3.9+ (tested on 3.9 and 4.4), Confluent Platform 7.9+.

## Notes

- **`producer`:** `KafkaProducer` and `AsyncKafkaProducer` share one Rust actor; the async class drives it from the event loop's executor and inherits every cell here
- **`consumer`:** the async consumer (`AsyncKafkaConsumer`) delegates every call to a `KafkaConsumer` on a worker thread and returns what the sync call returns
- **`compat.confluent`:** A translation over the native rows, matching confluent-kafka 2.15.1's API. Cells hold within the translated config.
  - `enable.auto.commit` defaults to `True`, confluent-kafka's default (native: `False`).
  - Under auto-commit, the tick, `close()` and a revoke commit what `poll()` handed out, never the buffered batch's fetch position. `enable.auto.offset.store` is honoured (default true); see [Migrating from confluent-kafka](https://docs.pyrula.io/kafka/confluent-compat.md).
  - `client.id` maps to `client_id` on the producer and the consumer.
  - A full queue raises `BufferError`, as confluent-shaped code expects.
  - confluent's `Producer` has no `close()`, so the `exit` cell is `flush()`.

## Limits

- **`producer` - offset commit, rebalance:** the producer has no offsets and no group; the column belongs to the consumer
- **`consumer` - ordering:** the consumer produces nothing; the column belongs to the producer
- **`compat.confluent` - delivery:** delivery callbacks fire on `poll()`/`flush()`, as confluent's do; the without-polling clause is native-only

## Failure behavior

- Delivery: a retried batch under `enable_idempotence=True` does not duplicate; a batch whose
  fate is unknown at the deadline or when a finite retry budget runs out fails its receipt
  with a `DeliveryTimeout` that says so, the producer epoch bumps and later sends continue;
  a send that partly lands fails its receipt with a `ProduceError` whose `report` counts
  what landed and names the partitions that failed. That count is exact on every path:
  `enable_idempotence=True`, `acks="all"`, `acks=1`, `acks=0` and inside a transaction. A
  produce that fails on some partitions retries those partitions only; records the broker
  already acknowledged are not sent again.
- Ordering: order holds after a retried/duplicate-sequence-number ack, not only across
  concurrent in-flight batches.
- Exit: the same bound `flush()` drains under (`message_timeout_ms`, within which a record
  is retried up to `message_send_max_retries` times) bounds the drain, and an undelivered record's receipt fails
  (logged at ERROR if nobody holds it).
- Transactions: after `abort_transaction()` the producer is reusable and the next
  transaction commits cleanly.
- Transactions: a worker process stopped mid-transaction past its session timeout, whose
  partition a second worker (same per-partition `transactional_id`) takes over and commits,
  is fenced when it resumes: its late commit and its late produce fail as fatal, a
  `read_committed` reader sees each input's output once and the group's offset covers
  every input. An at-least-once worker stopped the same way duplicates its held batch.

## confluent-kafka behavior

This table compares pyrula-kafka with confluent-kafka-python 2.15.1 and names the matrix
row and column that covers each behavior.

| You are used to | Here | Matrix cell |
|---|---|---|
| Records still queued when the producer is destroyed are dropped unless you `flush()` first | Leaving the context manager drains, under the same retry budget `flush()` uses; what cannot be delivered fails its receipt or is logged; `purge()` is the opt-out | `producer` - exit |
| Delivery callbacks fire only when you call `poll(0)` or `flush()` | Receipts resolve on the broker's ack without polling; the shim keeps confluent's poll-driven callbacks | `producer` - delivery |
| `BufferError` on a full queue, and it is on you to retry | The same refusal, as `QueueFull` natively and `BufferError` through the shim; nothing already accepted is dropped | `producer` - backpressure |
| A batch that expires after it was sent (`delivery.timeout.ms`) fails its callback, librdkafka bumps the producer epoch and carries on | The same KIP-360 behaviour: the unknown-fate batch's receipt fails with a `DeliveryTimeout` saying it may have landed, the epoch bumps, later sends continue on the same producer; the error's `report` says what landed | `producer` - delivery |
| Committing in `on_revoke` yourself when auto-commit is off, because only auto-commit commits stored offsets on revoke | Offsets staged with `store_offsets` are committed before the partition leaves, whether or not auto-commit is on; callbacks fire in order | `consumer` - rebalance |
| Calling `commit()`, `position()` on the consumer from inside a rebalance callback | Parity: the callback receives the consumer it was registered on and runs with it released, so `commit`, `position`, `committed`, `pause`/`resume`, `assignment`, `seek` and the acknowledgement calls (`assign`/`unassign` eager, `incremental_*` cooperative-sticky) all work inside it, through the shim too | `consumer` / `compat.confluent` - rebalance |
| `assign()` / `unassign()` / `incremental_assign()` / `incremental_unassign()` / `seek()`, standalone or inside a callback, with the `OFFSET_BEGINNING`/`OFFSET_END`/`OFFSET_STORED`/`OFFSET_INVALID` sentinels on the `TopicPartition`s you hand them | The same five calls, resolving the same sentinels; a `seek()`, `assign()` or `unassign()` also drops any record the shim had already read ahead of the old position, so nothing pre-seek is ever handed out and nothing commits below it, unless you staged a lower offset with `store_offsets()`, which stays in force across a seek as it does in confluent | `consumer` / `compat.confluent` - rebalance |
| `_MAX_POLL_EXCEEDED` arriving as an error `Message` you must check `msg.error()` for | The same eviction, surfaced as a typed error from `poll_batch` instead of a message you inspect; through the shim, `poll()` hands it back as an error `Message` and `consume()` raises, with `KafkaError._MAX_POLL_EXCEEDED` as the wrapped error's `code()` | `consumer` / `compat.confluent` - rebalance |
| `err.retriable()` / `err.txn_requires_abort()` / `err.fatal()` to decide what to do with a transaction failure | The same three, on the native error and on the shim's `KafkaError`: a failed transactional produce, partition registration or `send_offsets_to_transaction` leaves the transaction abortable, as librdkafka's does, and `commit_transaction` then fails with that failure's code, `txn_requires_abort()` true and neither fatal nor retriable, until you abort; a fenced producer is fatal. A client-side refusal is not abortable, as in confluent | `producer` - transactions |
| A delivery callback's `msg.partition()` / `msg.offset()`, or extra bookkeeping to learn where a record landed | The broker's real partition and offset - the same values `ProduceReport.deliveries[i]` names for that record | `producer` - delivery |
| `close()` commits when `enable.auto.commit` is on and not otherwise | Under auto-commit the same; with it off, offsets you staged with `store_offsets` are still committed at `close()`, a deliberate deviation in the safe direction - confluent would drop them | `consumer` - exit |

### Not yet through the shim, or that behaves differently by default

These confluent-kafka surfaces are unavailable or have different defaults.

- `group.protocol=consumer` (KIP-848).
- The Protobuf Schema Registry serializer (`schema_registry.protobuf`); the native
  `ProtobufRegistrySerde` exists. `SerializingProducer` / `DeserializingConsumer`, the
  Schema Registry client object and the Avro / JSON Schema serializers are through, with
  the limits the [migration page](https://docs.pyrula.io/kafka/confluent-compat.md#schema-registry-serializers)
  lists.