Transactions

Transactions give you exactly-once semantics for the read-process-write pattern: you consume from one topic, produce to another, and commit the input offsets and the output records as one atomic unit. Either all of it lands or none of it does.

Set a transactional_id on the producer config and call init_transactions once at startup.

from pyrula.kafka import KafkaConfig, KafkaProducer

producer = KafkaProducer(KafkaConfig(
    bootstrap_servers="localhost:9092",
    transactional_id="etl-1",
    enable_idempotence=True,
))
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).unwrap_or_raise()

    # commit the consumed offsets inside the same transaction
    producer.send_offsets_to_transaction(
        batch.offsets,
        consumer.consumer_group_metadata(),
    ).unwrap_or_raise()
    producer.commit_transaction().unwrap_or_raise()

If anything fails mid-cycle, call producer.abort_transaction() and the partial output and offsets are discarded. The next attempt reprocesses the same input.

produce_many_bytes_txn produces a transactional batch of raw values; produce_many_records_txn produces fully-formed records.

Reading committed data

A consumer’s isolation_level defaults to read_committed, so it only sees records from committed transactions and skips aborted ones. That’s what makes the downstream side of an exactly-once pipeline correct without extra work.

Verify EOS at the broker, not from process output. Aborted records are invisible to a read_committed consumer, so a quick print can look right while the transaction state is wrong.