# Transactions

> Exactly-once across a consume/transform/produce cycle.

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

A transaction commits the consumed input offsets and the produced output records as one
unit. `read_committed` consumers see all of a committed transaction or none of an aborted
one. Side effects outside Kafka are not covered.

A second producer that initialises the same `transactional_id` bumps the epoch (KIP-98) and
the broker refuses the older one's writes. Use one transactional id per input partition, as
`@kafka_workflow` does, so whoever takes over a partition fences the previous owner.
Group-generation fencing (KIP-447) is not implemented: the broker can't refuse a commit from
a member that lost its partitions in a rebalance.

Set `transactional_id` and call `init_transactions` once at startup. A transactional
producer is always idempotent (`enable_idempotence=False` is refused), and its
`message_timeout_ms` defaults to the 60 s transaction timeout and can't exceed it.

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

producer = KafkaProducer(KafkaConfig(
    bootstrap_servers="localhost:9092",
    transactional_id="etl-1",
))
producer.init_transactions().unwrap_or_raise()
```

## The cycle

```python
while True:
    result = consumer.poll_batch(max_records=500, timeout_ms=1000)
    if result.is_err() or result.value.count == 0:
        continue
    batch = result.value

    producer.begin_transaction().unwrap_or_raise()
    outputs = [transform(r) for r in batch.records.to_list()]
    producer.produce_many_records_txn(outputs)   # a receipt; commit waits for it

    # commit the consumed offsets inside the same transaction
    group = consumer.consumer_group_metadata()   # Nothing until the consumer joins
    producer.send_offsets_to_transaction(batch.offsets, group).unwrap_or_raise()
    committed = producer.commit_transaction()
    if committed.is_err():
        producer.abort_transaction().unwrap_or_raise()
        consumer.rewind(batch).unwrap_or_raise()   # the next poll delivers the batch again
```

After an abort the consumer's position is already past the batch, so call
`consumer.rewind(batch)`. Without it the next transaction commits offsets beyond the
aborted batch and those records are never processed.

A complete program is in [Examples](https://docs.pyrula.io/kafka/examples.md#transactional-consume-transform-produce).

## Producing inside a transaction

Every produce path (`produce()`, `produce_many()`, `produce_many_bytes()`,
`produce_many_records_txn`, ...) writes into the open transaction and returns a
`ProduceReceipt` at once. A failed record reports at its receipt; the commit says to abort.
`produce_many_records_txn` takes records for several topics in one call.

A produce with no open transaction (before `begin_transaction()`, during a commit or
abort, or once the transaction requires abort) raises `ProduceError` with `code()`
`ERR__STATE` and queues nothing.

## Commit and abort

`commit_transaction()` waits for every record produced in the transaction to be answered.
If any failed, the commit is refused with `txn_requires_abort()` true, neither `fatal()`
nor `retriable()`.

`abort_transaction()` purges queued records (receipts fail with `_purge_kind == "queue"`),
waits for requests already on the wire, then ends the transaction. When a record's fate is
unknown, or after `OUT_OF_ORDER_SEQUENCE_NUMBER` or `UNKNOWN_PRODUCER_ID`, the abort also
bumps the producer epoch, as librdkafka does. Both codes are abortable, not fatal.

### Timeouts

Each call takes an optional `timeout_ms`. `init_transactions` defaults to twice the 60 s
transaction timeout; `begin_transaction`, `commit_transaction` and `abort_transaction` to
the transaction timeout. A commit, abort or init that times out fails with a retriable
error but keeps running in the background: call the same method again to resume it, and it
returns the stored answer if it already finished. `timeout_ms=0` returns at once unless
finished. `send_offsets_to_transaction` is capped at the transaction timeout and is not
resumed; call it again.

Each request to a coordinator is bounded by `request_timeout_ms`. A different transactional
call while one is running is refused as `_conflicting_use` (the shim reports `_CONFLICT`);
the same call from another thread is `_prev_in_progress`. Both are retriable.

## Failures

After a failed transactional record, partition registration or
`send_offsets_to_transaction`, the transaction can only be aborted. `commit_transaction`
then fails with the original failure's class and code, and produces and
`send_offsets_to_transaction` are refused (`ERR__STATE`).

```python
result = producer.commit_transaction()
if result.is_err():
    if result.error.txn_requires_abort():
        producer.abort_transaction().unwrap_or_raise()
        consumer.rewind(batch).unwrap_or_raise()  # then reprocess the input
    elif result.error.retriable():
        ...  # timed out: call commit_transaction() again to resume it
    elif result.error.fatal():
        ...  # fenced: close this producer and start a new one
```

Refusals made before a record is queued (oversize record, bad partition) leave the
transaction committable.

A fatal error, such as being fenced by a newer instance with the same `transactional_id`,
ends the producer: every later transactional call and produce fails with it. Close it and
create a new one with the same id.

## Reading committed data

`isolation_level` defaults to `read_committed`, so a consumer sees only committed
transactions' records. Verify exactly-once at the broker, not from process output: aborted
records are invisible to a `read_committed` consumer.