A transactional producer writes records to multiple Kafka topic-partitions atomically: every write succeeds together, or Kafka commits none. This suits duplicate-sensitive operations such as financial records. section 4.4 section 4.4.2
Configure a unique transactional.id for each producer. Create the KafkaProducer, call initTransactions(), begin each transaction before sending, then commit or abort it. Kafka automatically enables idempotence for transactional producers. section 4.4.2 section 4.4.2
A practical transaction lifecycle is:
- Assign a transactional.id that identifies the producer application segment and remains unique among producers. section 4.4.2
- Call initTransactions() so the transaction coordinator registers the producer and completes recovery from an earlier transaction when necessary. section 4.4.2
- Call beginTransaction(), send records to the required topic-partitions, and call commitTransaction() to publish them atomically. section 4.4.2
- Abort on KafkaException when retrying is safe; close the producer for fencing, ordering, or authorization failures. section 4.4.2
Handling a transactional send
A KafkaProducer configured with transactional.id initializes once, begins a transaction, sends a record, and commits. ProducerFencedException, OutOfOrderSequenceException, and AuthorizationException require closing the producer; another KafkaException triggers abortTransaction() and can be safely retried. section 4.4.2
Listing 4.7 KafkaProducer basics for transactions
HashMap<String, Object> producerProps = new HashMap<>();
producerProps.put("transactional.id", "set-a-unique-transactional-id"); #1
Producer<String, String> producer = new KafkaProducer<>(producerProps);
producer.initTransactions(); #2
try {
producer.beginTransaction(); #3
producer.send(topic, "key", "value"); #4
producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException #5
| AuthorizationException e) {
producer.close(); #6
} catch (KafkaException e) {
producer.abortTransaction();
// safe to retry at this point #7
}
Browse the complete Java configuration and transaction lifecycle, including initialization, sending, commit, abort, retry, and terminal error handling. section 4.4.2
