Transactional producer in Kafka - Mini Project: Build & Apply

Transactional producer

📖 Scenario: You are building a simple Kafka producer that sends messages to a topic in a safe way using transactions. This means either all messages in a batch are sent successfully, or none are sent at all. This helps keep data consistent.

🎯 Goal: Create a Kafka transactional producer that sends two messages to a topic orders inside a transaction and commits the transaction.

📋 What You'll Learn

Create a Kafka producer with transactional id order-producer-1

Initialize transactions before sending messages

Begin a transaction before sending messages

Send two messages with keys order1 and order2 and values apple and banana to topic orders

Commit the transaction after sending messages

Close the producer at the end

💡 Why This Matters

🌍 Real World

Transactional producers are used in financial systems, order processing, and other applications where data consistency is critical.

💼 Career

Understanding Kafka transactions is important for backend developers, data engineers, and anyone working with reliable event streaming.

1

Create Kafka producer with transactional id

Create a Kafka producer called producer with the following properties: bootstrap.servers set to localhost:9092, key.serializer and value.serializer set to org.apache.kafka.common.serialization.StringSerializer, and transactional.id set to order-producer-1.

Kafka

// Your code here to create the Kafka producer with transactional id

Need a hint?

Use Properties to set configuration and pass it to KafkaProducer. Don't forget to set transactional.id.

2

Initialize transactions

Call producer.initTransactions() to initialize transactions before sending any messages.

Kafka

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;

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());
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-producer-1");

KafkaProducer producer = new KafkaProducer(props);
// Initialize transactions

Need a hint?

Call initTransactions() on the producer before starting any transaction.

3

Begin transaction and send messages

Begin a transaction by calling producer.beginTransaction(). Then send two messages to topic orders: one with key order1 and value apple, and another with key order2 and value banana. Use producer.send() with new ProducerRecord<>(topic, key, value).

Kafka

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import org.apache.kafka.clients.producer.ProducerRecord;

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());
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-producer-1");

KafkaProducer producer = new KafkaProducer(props);
producer.initTransactions();
// Begin transaction and send two messages

Need a hint?

Use beginTransaction() before sending messages. Use ProducerRecord to create messages with topic, key, and value.

4

Commit transaction and close producer

Commit the transaction by calling producer.commitTransaction(). Then close the producer by calling producer.close(). Finally, print "Transaction committed successfully".

Kafka

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import java.util.Properties;
import org.apache.kafka.clients.producer.ProducerRecord;

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());
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, "order-producer-1");

KafkaProducer producer = new KafkaProducer(props);
producer.initTransactions();

producer.beginTransaction();
producer.send(new ProducerRecord("orders", "order1", "apple"));
producer.send(new ProducerRecord("orders", "order2", "banana"));
// Commit transaction and close producer

Need a hint?

Use commitTransaction() to finish the transaction and close() to release resources. Use System.out.println() to print the success message.