Using Transactions to Atomically Produce to Multiple Topics

Apache Kafka’s transactions let a producer write to multiple topic-partitions as a single all-or-nothing unit: either every record in the batch becomes visible to consumers, or none of it does. This is the primitive behind exactly-once stream processing, where a consume-transform-produce loop must atomically commit its output records and the source-offset commit. For platform and SRE engineers running production clusters, the details that matter are the transaction lifecycle, the broker-side coordinator, the configuration knobs, and the failure modes — because that is where the guarantees actually hold or break. This builds directly on Idempotent Producers and Exactly-Once Semantics.

The Problem: Why Atomic Multi-Topic Writes Matter

Consider a producer writing to two topics — an orders topic and a payments topic. Without transactions, if the application crashes or a network partition occurs after the orders record is acknowledged but before the payments record is sent, the system is left inconsistent: a downstream consumer sees an order with no matching payment. Workarounds — packing both events into one composite record, or implementing compensating logic in every consumer — push complexity and latency onto everyone downstream.

Kafka transactions solve this by wrapping multiple send() calls (and, in stream processing, the consumer-offset commit) into one atomic commit. The guarantee is enforced by a two-phase commit protocol between the producer and a broker-side transaction coordinator. This is a natural extension of the broader Producers, Consumers & Delivery Semantics model.

How Kafka Transactions Work: A Two-Phase Commit

A transactional producer operates within a session identified by a unique transactional.id. Setting this property (which also implicitly enables idempotence) is what allows guarantees to span producer restarts: it lets the coordinator fence stale instances and recover or abort pending transactions from a previous session. The lifecycle is a strict sequence:

  1. Initialization (initTransactions()) — The producer registers its transactional.id with the coordinator. The coordinator bumps the producer epoch, fencing any older producer using the same ID, and aborts any transaction left dangling by a prior session.
  2. Begin (beginTransaction()) — Starts a new atomic unit, locally on the producer.
  3. Produce — Records are sent to multiple topic-partitions as usual. On the first send to a new partition, the producer registers that partition with the coordinator (an AddPartitionsToTxn request). The records are written to the partition logs immediately but remain invisible to consumers running with isolation.level=read_committed.
  4. Send offsets (optional) — In a consume-transform-produce loop, sendOffsetsToTransaction() adds the consumer-group offset commit to the same transaction, so input progress and output records commit together.
  5. Commit or abort (commitTransaction() / abortTransaction()) — The coordinator writes a PREPARE_COMMIT (or PREPARE_ABORT) record to the internal __transaction_state log, then fans out transaction markers (control records) to every involved partition leader. Once the markers are written, the coordinator writes the final COMPLETE_COMMIT/COMPLETE_ABORT record. The markers are what flip the records from invisible to visible — a read_committed consumer only delivers records up to the last stable offset (LSO) and filters out aborted batches.

The transaction state itself lives in the internal __transaction_state topic, which is log-compacted and replicated like any other topic.

Broker and Producer Configuration

The coordinator state topic is governed by broker configs, with these defaults (Kafka 4.1):

  • transaction.state.log.num.partitions — partitions of __transaction_state. Default 50. Each transactional.id is hashed to a partition, and the leader of that partition is the coordinator for that ID, so this is what spreads coordinator load across brokers. It should not be changed after deployment.
  • transaction.state.log.replication.factor — default 3. The cluster must have at least this many brokers before the topic is created.
  • transaction.state.log.min.isr — default 2.
  • transaction.max.timeout.ms — the broker-side ceiling on a producer’s requested transaction timeout. Default 900000 (15 min). A producer asking for more is rejected.
  • transactional.id.expiration.ms — how long the coordinator retains a transactional.id’s state with no transaction activity before expiring it. Default 604800000 (7 days). (This config is real and broker-side; do not confuse it with a producer setting.)

On the producer:

  • transaction.timeout.ms — the maximum time the coordinator allows a transaction to remain open before it proactively aborts it. Default 60000 (1 min), and it must not exceed transaction.max.timeout.ms. Set it too low and long batches get aborted mid-flight; set it too high and zombie transactions linger, holding the last stable offset back and stalling read_committed consumers. For multi-minute batch jobs, raise it above the worst-case batch duration (and raise the broker ceiling to match).

Failure Modes and Recovery Patterns

Producer crash before commit. If a producer dies after beginTransaction() but before commitTransaction(), no commit marker is written. Once transaction.timeout.ms elapses, the coordinator aborts the transaction and writes abort markers; read_committed consumers never see those records. When the producer restarts, initTransactions() aborts any still-pending transaction from the previous session before proceeding.

Coordinator failover. The coordinator is the leader broker for the relevant __transaction_state partition. If it fails, a new leader is elected and rebuilds transaction state by replaying that partition’s log. During the handoff, producers may get NOT_COORDINATOR (or COORDINATOR_NOT_AVAILABLE) errors; the client transparently re-discovers the coordinator and retries.

Zombie producer fencing. If a producer is presumed dead but is merely stalled (e.g., a long GC pause) and a replacement starts with the same transactional.id, the replacement’s initTransactions() bumps the producer epoch. Any later send(), sendOffsetsToTransaction(), or commitTransaction() from the old instance fails with ProducerFencedException, which is fatal and requires closing the producer. This is what prevents split-brain duplicate writes.

To inspect transaction state on disk, dump the relevant __transaction_state partition:

# See the transactional offset commits tracked for a consumer group
kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group my-transactional-group --describe

# Decode the transaction state log to inspect pending transactions
kafka-dump-log.sh \
  --files /var/lib/kafka/data/__transaction_state-0/00000000000000000000.log \
  --transaction-log-decoder --deep-iteration

The --transaction-log-decoder option parses records from __transaction_state as transaction metadata. Look for entries still in the Ongoing state whose txnStartTimestamp is older than the configured timeout — those are stuck transactions holding back the last stable offset.

Monitoring and Observability

Expose both producer-side and broker-side metrics via JMX (and scrape with the Prometheus JMX exporter).

Producer metrics — MBean kafka.producer:type=producer-metrics,client-id=*. These are cumulative nanosecond counters, so alert on their rate of change:

  • txn-init-time-ns-total — time spent in initTransactions().
  • txn-begin-time-ns-total — time spent in beginTransaction().
  • txn-send-offsets-time-ns-total — time spent in sendOffsetsToTransaction().
  • txn-commit-time-ns-total — time spent committing.
  • txn-abort-time-ns-total — time spent aborting. A rising abort time/rate is the clearest sign of timeouts or fencing.

A spike in record-error-rate together with fatal ProducerFencedExceptions in the application logs indicates a fencing event (usually a duplicated transactional.id).

Broker coordinator metrics — the transaction marker channel manager exposes two gauges under MBean kafka.coordinator.transaction:type=TransactionMarkerChannelManager:

  • UnknownDestinationQueueSize — markers queued for brokers the coordinator can’t yet route to. A persistently nonzero value points to broker-discovery or connectivity problems delaying transaction completion.
  • LogAppendRetryQueueSize — transaction-log appends queued for retry. Growth here means the coordinator is struggling to persist state updates to __transaction_state.

Treat the __transaction_state topic like any other critical topic: alert on under-replicated partitions, since losing quorum there stalls all transactions whose IDs map to the affected partitions. A Prometheus alert on a stuck marker queue:

groups:
- name: kafka_transactions
  rules:
  - alert: TxnMarkerQueueStuck
    expr: avg_over_time(kafka_coordinator_transaction_transactionmarkerchannelmanager_unknowndestinationqueuesize[5m]) > 0
    for: 10m
    labels:
      severity: warning
    annotations:
      summary: "Transaction markers stuck for unknown destinations"
      description: "Coordinator on {{ $labels.instance }} cannot route transaction markers; transactions may hang."

(The exact scraped metric name depends on your JMX-exporter naming rules; verify the rendered name against /metrics before wiring the alert.)

Performance Considerations and Anti-Patterns

A commit is not free: it is a coordinator round-trip plus a marker fan-out to every partition the transaction touched, so commit latency grows with both commit frequency and partition fan-out.

  • Amortize commits over a batch. Group many send() calls into one transaction instead of committing per record pair. The transaction size trades off how much work you replay on abort against per-commit overhead.
  • Bound the partition fan-out. A single transaction spanning hundreds of partitions multiplies the marker fan-out. Design partitioning so a transactional unit touches a bounded set of partitions.
  • Keep transactions short. Open transactions pin the last stable offset on their partitions, blocking read_committed consumers from advancing and delaying log cleaning of the affected segments. Set transaction.timeout.ms to the shortest value your workload tolerates.

Two anti-patterns to avoid:

  • Sharing a transactional.id across instances for load balancing. Each initTransactions() fences the other instance, so they fence each other in a loop and throughput collapses. Every instance needs a stable, unique transactional.id (for Kafka Streams this is derived automatically from the application ID and task). Derive it from a stable identity — not a random UUID per restart, which would orphan the prior session’s state until transactional.id.expiration.ms cleans it up.
  • Mixing transactional and non-transactional producers on the same topic. Non-transactional writes have no markers, so a read_committed consumer sees them interleaved among committed transactional records, defeating the isolation you were paying for. Standardize on transactional producers for any topic read with read_committed.

The original design is specified in KIP-98: Exactly Once Delivery and Transactional Messaging, and the configuration defaults above are from the Apache Kafka broker and producer configuration references.

Used correctly, Kafka transactions give you atomic multi-topic writes and the foundation for exactly-once pipelines. The operational discipline — unique transactional IDs, bounded transaction scope, a healthy __transaction_state topic, and alerting on the real coordinator and producer metrics — is what keeps that guarantee intact when brokers and producers fail.

Back to Idempotent Producers And Exactly Once Semantics