Some Random Notes

7. Tx

3 min read

To set up a producer that safely writes messages inside a transaction block, you must configure a Transactional Producer. When using transactions, you group multiple send() calls together. They either all succeed together (Commit), or they are completely thrown away (Abort).

The Code: Transactional Producer

This configuration assigns a unique transactional.id to the producer. This ID registers the producer with the Kafka cluster so the brokers can track whether the messages are officially completed or canceled.

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.errors.OutOfOrderSequenceException;
import org.apache.kafka.common.errors.AuthorizationException;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
public class TransactionalOrderedProducer {
    public static void main(String[] args) {
        Properties props = new Properties();
        props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class.getName());
        // ─── CRITICAL TRANSACTION CONFIGURATIONS ──────────────────────────────
        // 1. Must give the producer a unique Transactional ID
        props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-tx-producer-v1");
        // 2. Transactions REQUIRE idempotence to be turned on automatically
        props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, "true");
        props.put(ProducerConfig.ACKS_CONFIG, "all");
        try (KafkaProducer<String, String> producer = new KafkaProducer<>(props)) {
            // Step 1: Initialize the transactional states with the Kafka cluster
            producer.initTransactions();
            try {
                // Step 2: Begin the transaction block
                producer.beginTransaction();
                System.out.println("Transaction started...");
                String topic = "order-events";
                String key = "customer_abc";
                // Step 3: Send multiple related records sequentially
                // In a read_committed consumer, NONE of these are visible until commitTransaction() runs
                producer.send(new ProducerRecord<>(topic, key, "Item_1_Added"));
                producer.send(new ProducerRecord<>(topic, key, "Item_2_Added"));
                producer.send(new ProducerRecord<>(topic, key, "Payment_Submitted"));
                // Step 4: Permanently seal the transaction logs
                producer.commitTransaction();
                System.out.println("Transaction successfully committed! Messages are visible.");
            } catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
                // These are unrecoverable errors. We cannot abort or retry; the producer must close.
                System.err.println("Fatal producer error occurred: " + e.getMessage());
            } catch (Exception e) {
                // For any standard runtime errors, clear the conveyor belt and cancel the write
                System.err.println("Error during execution. Aborting transaction: " + e.getMessage());
                producer.abortTransaction();
            }
        } 
    }
}

End-to-End Summary Checklist

You now have the complete blueprint for handling message order under every core Kafka pattern. To ensure you do not experience out-of-order bugs, verify that your production system checks off these rules:

  • Mapping: Messages that must stay in order must share the exact same Message Key.
  • Producer: Set enable.idempotence = true and acks = all to handle transient network retry glitches smoothly.
  • Consumer Threading: Process messages sequentially inside the single-threaded poll loop. Do not hand records over to an asynchronous ExecutorService thread pool.
  • Failures: Do not bypass single-message failures using a blind DLQ if subsequent messages depend on them. Use a retry circuit breaker or a hold-state lookup to park the specific entity key.
  • Transactions: If writing messages inside a transaction block, make sure your reading consumers have isolation.level = read_committed configured. Would you like to analyze a specific log error or architecture layout you are currently building to see if it meets these constraints?

Reference

https://www.conduktor.io/glossary/kafka-transactions-deep-dive https://medium.com/shoutloudz/apache-kafka-transactions-349ad4617856 https://books.japila.pl/kafka-internals/transactions/ https://www.warpstream.com/blog/kafka-transactions-explained-twice https://www.confluent.io/blog/transactions-apache-kafka/

Backlinks