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.

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

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.

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

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.