Transactional Messaging with Kafka Producers

Transactional messaging in Apache Kafka lets a producer write to multiple partitions and topics as a single atomic unit, and atomically commit the consumer offsets that drove those writes. For platform and SRE engineers running production clusters, the mechanics, failure modes, and operational overhead of transactions are what make exactly-once stream processing actually hold up under load. This article dissects the producer-side transaction protocol, its place in the broader Producers, Consumers & Delivery Semantics framework, and the configuration and monitoring required to run it reliably at scale.

The Atomic Multi-Partition Write Problem

A standard Kafka producer with idempotency enabled gives you per-partition ordering and de-duplication, but it cannot coordinate writes across partitions. Without transactions, an application that must publish an output record to a processed-events topic and update a state marker in a job-control topic has a window of inconsistency: if the producer crashes between the two sends, one write lands and the other does not. This is the classic dual-write problem. Kafka solves it with a transaction coordinator and a two-phase commit protocol adapted to the log.

The transaction protocol introduces one new entity: the transaction coordinator. Each transactional producer is assigned a coordinator, hosted on the broker that leads the partition of the internal __transaction_state topic to which the producer’s transactional ID maps (abs(hash(transactional.id)) % numPartitions). The coordinator manages the transaction lifecycle, persists state changes to __transaction_state, and coordinates with the consumer group coordinator when offsets are part of the transaction. The availability of that coordinator partition and the replication factor of the transaction state topic directly bound the resilience of your transactional pipelines.

By default, __transaction_state has 50 partitions and a replication factor of 3. Verify its health with:

kafka-topics.sh --bootstrap-server broker1:9092 \
  --describe --topic __transaction_state

Look for under-replicated partitions or offline replicas. A single unavailable transaction-state partition stalls every producer whose transactional ID maps to it, so this topic belongs on your critical-path monitoring.

Configuring the Transactional Producer

Enabling transactions requires a specific configuration set and a deliberate initialization sequence. The producer must run with idempotency (a prerequisite for transactions, and the default since Kafka 3.0) and a stable, unique transactional.id. That identifier ties the producer to its transactional state across restarts and enables fencing of zombie producers through the epoch protocol described in Idempotent Producers and Exactly-Once Semantics.

Properties props = new Properties();
props.put("bootstrap.servers", "broker1:9092,broker2:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("enable.idempotence", true);
props.put("transactional.id", "my-app-txn-01");
props.put("transaction.timeout.ms", 90000);

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

Setting transactional.id implicitly enables idempotence, so enable.idempotence=true is redundant but harmless to state explicitly. The ID must be unique per active producer instance; if two live instances share it, the one that calls initTransactions() last bumps the epoch and fences the other, which then sees a ProducerFencedException on its next transactional call.

transaction.timeout.ms controls how long the coordinator waits before proactively aborting an open transaction. The default is 60000 (60 seconds). Workloads with long processing gaps between sends and the commit should raise it (e.g. 90,000–120,000 ms). The broker caps this value at transaction.max.timeout.ms, which defaults to 900000 (15 minutes); a producer requesting a larger timeout is rejected at initTransactions().

For Python applications using confluent-kafka-python, the configuration mirrors the Java client:

from confluent_kafka import Producer

conf = {
    'bootstrap.servers': 'broker1:9092,broker2:9092',
    'enable.idempotence': True,
    'transactional.id': 'py-app-txn-01',
    'transaction.timeout.ms': 90000,
}
producer = Producer(conf)
producer.init_transactions()

Always call initTransactions() before any transactional work. It blocks until the producer has registered with the coordinator, obtained its producer ID, and fenced any prior instance of the same transactional.id by bumping the epoch. If the coordinator is unavailable it retries until the timeout you pass (or the default), so wrap initialization in logic that handles a TimeoutException rather than assuming it returns immediately.

The Two-Phase Commit Protocol in Kafka

Kafka adapts the classic two-phase commit pattern to a distributed log. A transaction moves through begin, add-partitions, write, and commit-or-abort. The first time a transaction writes to a partition, the producer sends an AddPartitionsToTxn request so the coordinator can track the partition set. Each produced record batch carries the producer ID and epoch. Consumers configured with isolation.level=read_committed do not see these records while the transaction is open; they remain buffered behind the last stable offset (LSO) until the transaction resolves.

When the producer calls commitTransaction(), it sends an EndTxn request with the commit decision to the coordinator. The coordinator writes a PrepareCommit record to the transaction log, then sends WriteTxnMarkers requests so a COMMIT control marker is appended to every partition in the transaction. Only once those markers are written do the records advance the LSO and become visible to read-committed consumers. On abortTransaction() or a timeout, the coordinator writes ABORT markers instead, and read-committed consumers skip the aborted records.

The canonical transactional loop, matching the pattern in the KafkaProducer javadoc:

producer.initTransactions();
try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("processed-events", key1, event)).get();
    producer.send(new ProducerRecord<>("job-control", key2, controlUpdate)).get();
    // Atomically commit consumed offsets alongside produced records
    producer.sendOffsetsToTransaction(offsetMap, consumerGroupMetadata);
    producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException
         | UnsupportedVersionException | AuthorizationException e) {
    // Fatal: the producer is defunct. Close it and rebuild.
    producer.close();
    throw e;
} catch (KafkaException e) {
    // Abortable: roll back and retry the whole transaction from beginTransaction()
    producer.abortTransaction();
}

The .get() calls make each send synchronous so a send-time error surfaces before the commit. This is not strictly required—commitTransaction() flushes and will itself fail if any buffered send failed—but it gives you a precise point to catch and classify per-record errors. ProducerFencedException, OutOfOrderSequenceException, UnsupportedVersionException, and AuthorizationException are fatal: the producer enters a defunct state and the only option is close(). Other KafkaExceptions are abortable—call abortTransaction() and retry the transaction.

sendOffsetsToTransaction is the bridge that makes consume-process-produce exactly-once: it folds the consumed offsets into the same transaction as the produced records, so a consumer restart neither duplicates nor drops work. This is the core primitive Kafka Streams uses under the hood.

Operational Tuning and Failure Modes

Broker-Side Settings

The most impactful broker settings for the transaction state topic are below. Note the config prefix is transaction.state.log.* (not transactional.state.log.*—the latter does not exist).

Note: transaction.state.log.num.partitions is read only when __transaction_state is first created. Changing it on an existing cluster means deleting and recreating the topic, which is disruptive and aborts in-flight transactions; treat it as a deployment-time decision. The replication factor and min-ISR are likewise applied at topic creation.

Commit Latency and Partition Fan-Out

A transaction that spans many partitions pays commit latency proportional to the partition count, because the coordinator writes a marker to each one. To see the partition set recorded for transactions, decode the transaction log. On Kafka 4.0+, the formatter lives in the tools module:

kafka-console-consumer.sh --bootstrap-server broker1:9092 \
  --topic __transaction_state \
  --formatter "org.apache.kafka.tools.consumer.TransactionLogMessageFormatter" \
  --from-beginning --max-messages 10

On Kafka 3.x the class is kafka.coordinator.transaction.TransactionLog$TransactionLogMessageFormatter (deprecated in 3.9 and removed in 4.0). If a producer cannot finish a commit before transaction.timeout.ms elapses, the coordinator aborts the transaction. Mitigate by raising the timeout or writing fewer partitions per transaction.

Zombie Producers and Timed-Out Transactions

A zombie producer is an instance that stalls (often a network partition or long GC pause), then resumes and tries to continue a transaction the cluster has moved past. Two mechanisms protect you:

• Fencing. If a newer instance registered the same transactional.id, it bumped the epoch. The zombie’s next request carries the stale epoch and is rejected with ProducerFencedException—fatal; the instance must close().
• Timeout recovery. If the zombie’s transaction simply timed out and was aborted by the coordinator (and no newer instance exists), the producer is not permanently dead. Since KIP-588 the timeout surfaces as an abortable error: the client bumps its own epoch internally, and the application recovers by calling abortTransaction() and starting a new transaction—no full initTransactions() rebuild required.

Treat the fatal/abortable split exactly as the loop above does, and you handle both cases correctly.

Interaction with Consumer Group Rebalancing

If a consumer group rebalance revokes partitions mid-transaction, the offsets you are about to commit may no longer belong to this member. sendOffsetsToTransaction (or the commit) then fails with CommitFailedException, an abortable error. The application must abort and retry after the rebalance settles. Kafka Streams handles this internally; custom consume-process-produce loops must implement the retry. See Consumer Group Rebalancing: Protocols and Tuning.

Monitoring and Observability

The Java producer exposes transaction timing under the kafka.producer:type=producer-metrics,client-id=<id> MBean as cumulative nanosecond totals (there are no per-transaction rate or average attributes built in—derive rates from the deltas of these totals):

To track fencing, instrument it where it is observable: count ProducerFencedException occurrences in your application’s error handling (the producer does not publish a dedicated fenced-rate metric). Any non-zero count means a zombie was fenced—investigate.

A Prometheus JMX-exporter rule set for the transaction totals:

lowercaseOutputName: true
rules:
  - pattern: 'kafka.producer<type=producer-metrics, client-id=(.+)><>(txn-init-time-ns-total|txn-begin-time-ns-total|txn-commit-time-ns-total|txn-abort-time-ns-total|txn-send-offsets-time-ns-total):'
    name: kafka_producer_$2
    labels:
      client_id: "$1"

Because these are counters, graph rate(kafka_producer_txn_commit_time_ns_total[5m]) to see commit-time growth, and pair it with the broker side. On the broker, the transaction coordinator emits metrics under kafka.coordinator.transaction and kafka.server, including request rates for AddPartitionsToTxn and EndTxn (via the standard kafka.network:type=RequestMetrics request-rate MBeans). Watch these alongside producer commit times: when broker EndTxn handling slows but producer commit totals keep climbing, the coordinator—not the client—is the bottleneck.

For the full metric catalog, see the official Kafka Monitoring Guide and the Confluent producer metrics reference.

Performance Overhead and Capacity Planning

Transactions add real overhead on both sides. Each transaction needs at least an AddPartitionsToTxn round trip per new partition plus an EndTxn to commit, the coordinator persists every state transition to __transaction_state, and a control marker is written to every data partition in the transaction. A transaction spanning ~100 partitions can push commit latency into the hundreds of milliseconds, versus single-digit milliseconds for a non-transactional write, because all of those markers must be written and acknowledged.

The coordinator processes a given transactional ID through a single transaction-log partition, so transaction-state partition count is your horizontal-scaling lever for high transactional-ID cardinality and throughput. If a single application’s commit volume is the constraint, the more effective levers are fewer partitions per transaction and larger batches per transaction (amortizing the fixed RPC cost), rather than more transaction-state partitions.

For a coarse, offline look at how many transaction markers exist in a data partition, dump its log segment:

kafka-run-class.sh kafka.tools.DumpLogSegments \
  --files /var/lib/kafka/data/processed-events-0/00000000000000000000.log \
  --print-data-log | grep -c "endTxnMarker"

The --print-data-log flag prints record contents, and committed/aborted transactions appear as control records (isControl: true) carrying endTxnMarker: COMMIT or ABORT. This is a rough, point-in-time heuristic on one segment, not a live counter—use the JMX totals for ongoing monitoring. (kafka-dump-log.sh is the supported wrapper for the same tool.)

Finally, transactions are not a substitute for idempotent consumers. Even with exactly-once on the produce path, any consumer that performs external side effects must still de-duplicate or be idempotent; the guarantee Kafka gives you is bounded by isolation.level and the consumer’s offset-management strategy. For the complete delivery picture, revisit Producers, Consumers & Delivery Semantics and Idempotent Producers and Exactly-Once Semantics.