skip to content

Walk through the transactional producer API lifecycle: transactional.id, initTransactions, beginTransaction, commitTransaction, abortTransaction.

level: middleimportance: must knowfreq 65%

answer

  1. initTransactions once at startup
  2. begin → send → commit/abort per unit
  3. sendOffsetsToTransaction for EOS
  4. begin is client-side; commit writes markers
  5. fatal errors → close, don't continue

basics

~10 s

Configure transactional.id, call initTransactions() once at startup. Then per unit of work: beginTransaction(), send records, and either commitTransaction() on success or abortTransaction() on failure. Repeat begin/commit for each transaction.

solid answer

~40 s

You set transactional.id (a stable unique string) in producer config; this forces idempotence. At startup call initTransactions() exactly once — it contacts the transaction coordinator, fetches a producer epoch, and aborts any unfinished transaction left by a prior instance with the same id. For each transaction: beginTransaction() marks the start (local, no broker round-trip), then send() calls accumulate; in consume-process-produce you also call sendOffsetsToTransaction(offsets, groupMetadata) to include consumer offsets. Finish with commitTransaction() (flushes, writes commit markers, returns when durable) or abortTransaction() (writes abort markers; read_committed consumers discard those records). On most send/commit exceptions you abort; on fatal ones like ProducerFencedException, OutOfOrderSequenceException, or AuthorizationException you must close the producer rather than continue. A single producer runs transactions serially — one open at a time.

go deeper

for a junior

Recall the call order: configure id, initTransactions once, then begin/send/commit-or-abort.

for a middle

Explain what each call does and where sendOffsetsToTransaction fits for exactly-once.

for a senior

Distinguish client-side begin vs coordinator markers, epoch bumping in init, and fatal vs abortable error handling.

for a principal

Discuss serial-transaction constraints, recovery semantics on restart, and how this maps onto Streams' commit loop.

## The five pieces **1. `transactional.id` (config)** — A user-chosen, stable, unique string set before constructing the producer (e.g. `props.put("transactional.id", "orders-processor-0")`). It must survive restarts and be unique per logical producer instance. Its purpose: let the broker recognize the same logical producer across restarts so it can fence zombies and recover transactions. Setting it implicitly forces `enable.idempotence=true`, `acks=all`, `retries>0`, and `max.in.flight.requests.per.connection<=5`. **2. `initTransactions()`** — Called **once**, after construction, before any send. It: - Registers the `transactional.id` with the **transaction coordinator** (the broker that owns the relevant `__transaction_state` partition). - Bumps and retrieves the **producer epoch** — a monotonically increasing number that fences older instances. - **Recovers**: aborts any transaction the previous incarnation left open, guaranteeing a clean slate. Blocks up to `max.block.ms`; throws on auth/timeout problems. **3. `beginTransaction()`** — Marks the logical start of a transaction in the client. It is purely client-side bookkeeping (no broker call yet); the coordinator learns of the transaction lazily when the first partition is written (an `AddPartitionsToTxn` request). Calling it twice without committing/aborting throws `IllegalStateException`. **4. `commitTransaction()`** — Flushes all buffered records, asks the coordinator to write **commit markers** (transaction control records) to every partition touched, and only returns once the transaction is durably committed. After this, `read_committed` consumers can see the records. **5. `abortTransaction()`** — Asks the coordinator to write **abort markers** to all touched partitions. `read_committed` consumers will discard those records. Use it in your `catch` block for retriable/abortable errors. ## Typical loop ``` producer.initTransactions(); while (running) { records = consumer.poll(); producer.beginTransaction(); try { for (r : records) producer.send(transform(r)); producer.sendOffsetsToTransaction(offsetsFor(records), consumer.groupMetadata()); producer.commitTransaction(); } catch (KafkaException e) { producer.abortTransaction(); } } ``` ## Edge cases & rules - **One at a time**: a producer cannot interleave two open transactions. - **Fatal vs abortable**: `ProducerFencedException`, `OutOfOrderSequenceException`, `UnsupportedVersionException`, and authorization errors are fatal — close the producer; don't try to abort and continue. - **commit throws**: if commitTransaction throws a non-fatal error you may retry; if fatal, close. - **Idle timeout**: if you begin but never commit within `transaction.timeout.ms`, the coordinator aborts for you and may fence the producer.

  • Does beginTransaction() make a broker round-trip?
    No. It is client-side bookkeeping. The coordinator first learns of the transaction when the first record is sent, via an AddPartitionsToTxn request to register that partition.
  • What does initTransactions() do about a transaction left open by a crashed previous instance?
    It bumps the producer epoch and aborts the prior incarnation's in-flight transaction, giving the new instance a clean state and fencing the old one.

saying these in an interview costs you the question

  • Calling initTransactions() per transaction instead of once at startup.
  • Trying to abort and keep using the producer after a ProducerFencedException (it is fatal — close it).
  • Believing beginTransaction contacts the broker.
  • Running two overlapping transactions on one producer instance.

context

resolving links…