A Kafka transaction makes a group of writes atomic across topic partitions: every write commits, or none does. It supports exactly-once delivery and is useful where duplicate records are unacceptable, such as financial processing. section 4.2.2 section 4.4.2
Enable transactions with a unique transactional.id. After initTransactions() registers that ID with the transaction coordinator, beginTransaction() starts work; commitTransaction() publishes it, while abortTransaction() discards it. Transactional producers are automatically idempotent, and transaction recovery uses the ID across producer sessions. 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
}
Use this producer pattern: initialize once, bracket sends with beginTransaction() and commitTransaction(), close on fencing or authorization failures, and abort retryable Kafka failures. section 4.4.2
Consumers should set isolation.level to read_committed to retrieve only committed transactional records. They stop at the last stable offset, because records in an open transaction are not yet decided; read_uncommitted can return aborted records too. section 4.4.3
Consume, transform, produce, and commit together
For a stock compliance-reporting flow, consume stock transactions, create broker summaries, send the summaries, then pass processed TopicPartition offsets to sendOffsetsToTransaction() before committing. If the transaction aborts, those offsets are not committed, so a restart does not skip the input records. section 4.4.4
Listing 4.9 Example of the consume–transform–produce with transactions
Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>(); #1
producer.beginTransaction(); #2
consumerRecords.partitions().forEach(topicPartition -> {
consumerRecords.records(topicPartition).forEach(record -> {
lastOffset.set(record.offset());
StockTransaction stockTransaction = record.value();
BrokerSummary brokerSummary = BrokerSummary.newBuilder() #3
producer.send(new ProducerRecord<>(outputTopic, brokerSummary));
});
offsets.put(topicPartition,
new OffsetAndMetadata(lastOffset.get() + 1L)); #4
});
try {
producer.sendOffsetsToTransaction(offsets,
consumer.groupMetadata()); #5
producer.commitTransaction(); #6
}
The listing tracks each partition’s next offset, sends output records, adds offsets to the producer transaction, and commits the whole consume-transform-produce unit. section 4.4.4
