Transactional producer in Kafka - Commands & Configuration

Introduction

Sometimes you want to send multiple messages to Kafka as a single group that either all succeed or all fail. A transactional producer helps you do this by making sure messages are sent together or not at all, avoiding partial updates.

When you need to update multiple Kafka topics atomically to keep data consistent.

When you want to ensure that a batch of messages is fully processed or fully discarded.

When your application requires exactly-once message delivery to avoid duplicates.

When you are implementing financial transactions or inventory updates that must not be partial.

When you want to combine producing messages with consuming and processing them in one atomic step.

Config File - producer.properties

producer.properties

bootstrap.servers=localhost:9092
key.serializer=org.apache.kafka.common.serialization.StringSerializer
value.serializer=org.apache.kafka.common.serialization.StringSerializer
transactional.id=my-transactional-producer-1
enable.idempotence=true
acks=all

This configuration file sets up the Kafka producer for transactions.

• bootstrap.servers: Kafka server address.
• key.serializer and value.serializer: How keys and values are converted to bytes.
• transactional.id: Unique ID to enable transactions for this producer.
• enable.idempotence: Ensures messages are not duplicated.
• acks=all: Waits for all replicas to confirm message receipt for safety.

Commands

Starts a Kafka producer with transactions enabled using the transactional ID. This allows sending messages as part of a transaction.

Terminal

kafka-console-producer --broker-list localhost:9092 --topic example-topic --producer-property transactional.id=my-transactional-producer-1

Expected Output

No output (command runs silently)

→

--broker-list - Specifies Kafka server addresses to connect.

→

--topic - Specifies the topic to send messages to.

→

--producer-property transactional.id=my-transactional-producer-1 - Enables transactional support with a unique ID.

Sends 5 messages of 10 bytes each to the topic using a transactional producer to test transactional message sending.

Terminal

kafka-producer-perf-test --topic example-topic --num-records 5 --record-size 10 --throughput 1 --producer-props bootstrap.servers=localhost:9092 transactional.id=my-transactional-producer-1 enable.idempotence=true acks=all

Expected Output

5 records sent, 5 records acknowledged, 0 records failed

1 records/sec (approximate throughput)

→

--topic - Topic to send messages to.

→

--num-records - Number of messages to send.

→

--producer-props - Properties to configure the producer including transactional settings.

Checks the consumer group status to verify if messages sent transactionally are available for consumption.

Terminal

kafka-consumer-groups --bootstrap-server localhost:9092 --describe --group example-group

Expected Output

TOPIC           PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG  CONSUMER-ID  HOST  CLIENT-ID

example-topic   0          5               5               0    consumer-1   /127.0.0.1  consumer-1

→

--bootstrap-server - Kafka server address.

→

--describe - Shows detailed info about the consumer group.

→

--group - Specifies the consumer group to check.

Key Concept

If you remember nothing else from this pattern, remember: transactional producers let you send multiple messages as one all-or-nothing group to keep data consistent.

Common Mistakes

Not setting a unique transactional.id in the producer configuration.

Without a transactional ID, Kafka cannot track transactions, so messages won't be sent atomically.

Always set a unique transactional.id property for each transactional producer instance.

Not enabling idempotence (enable.idempotence=true) when using transactions.

Idempotence prevents duplicate messages; without it, exactly-once delivery is not guaranteed.

Set enable.idempotence=true in the producer configuration to ensure safe retries.

Starting to send messages without initializing transactions in the producer code.

Messages sent outside a transaction are not part of the atomic group, defeating the purpose.

Always call initTransactions() and beginTransaction() before sending messages in transactional mode.

Summary

Configure the Kafka producer with a unique transactional.id and enable idempotence.

Start the producer with transactional support to send messages as a single atomic group.

Verify message delivery and consumer group status to ensure transactional messages are processed.

Practice

1.

What is the main purpose of using a transactional producer in Kafka?

easy

A. To compress messages before sending

B. To increase the speed of message delivery

C. To send multiple messages as a single atomic unit

D. To create multiple topics automatically

Solution

1. Step 1: Understand transactional producer role

   Transactional producers ensure that a group of messages are sent together as one unit, so either all succeed or none do.

2. Step 2: Compare options

   Options A, B, and D describe unrelated features like compression, speed, or topic creation, which are not the main purpose of transactions.

3. Final Answer:

   To send multiple messages as a single atomic unit -> Option C

4. Quick Check:

   Transactional producer = atomic message group [OK]

Hint: Transactional producer = all-or-nothing message send [OK]

Common Mistakes:

• Confusing transactions with message compression
• Thinking transactions speed up delivery
• Believing transactions create topics automatically

2.

Which configuration is required to enable transactions in a Kafka producer?

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
// Which property below is needed for transactions?

easy

A. props.put("max.in.flight.requests.per.connection", "5");

B. props.put("acks", "0");

C. props.put("compression.type", "gzip");

D. props.put("transactional.id", "my-transactional-id");

Solution

1. Step 1: Identify transaction enabling property

   The property "transactional.id" uniquely identifies the producer's transaction and is required to enable transactions.

2. Step 2: Check other options

   Options B, C, and D configure acknowledgments, compression, and request limits, which are unrelated to enabling transactions.

3. Final Answer:

   props.put("transactional.id", "my-transactional-id"); -> Option D

4. Quick Check:

   Enable transactions = set transactional.id [OK]

Hint: Set transactional.id to enable transactions [OK]

Common Mistakes:

• Using acks=0 instead of transactional.id
• Confusing compression with transactions
• Missing transactional.id causes errors

3.

What will be the output of the following Kafka producer code snippet?

producer.initTransactions();
producer.beginTransaction();
producer.send(new ProducerRecord<>("topic", "key1", "value1"));
producer.commitTransaction();
System.out.println("Transaction committed");

Assuming no errors occur, what is printed?

medium

A. Transaction aborted

B. Transaction committed

C. Exception thrown

D. No output

Solution

1. Step 1: Follow transaction flow

   The code initializes transactions, begins one, sends a message, then commits the transaction.

2. Step 2: Check output statement

   After commitTransaction(), it prints "Transaction committed". No errors occur, so this line executes.

3. Final Answer:

   Transaction committed -> Option B

4. Quick Check:

   Commit success prints message [OK]

Hint: commitTransaction() followed by print means success message [OK]

Common Mistakes:

• Assuming commitTransaction() prints automatically
• Confusing abortTransaction() with commitTransaction()
• Expecting no output when print is present

4.

Identify the error in this Kafka transactional producer code snippet:

producer.beginTransaction();
producer.send(new ProducerRecord<>("topic", "key", "value"));
// Missing commit or abort here
producer.close();

medium

A. Missing call to commitTransaction() or abortTransaction() before close()

B. beginTransaction() should be called after send()

C. producer.close() should be called before beginTransaction()

D. ProducerRecord is missing a partition number

Solution

1. Step 1: Check transaction lifecycle

   After beginTransaction() and send(), the code must call commitTransaction() or abortTransaction() to complete the transaction.

2. Step 2: Identify missing step

   The code closes the producer without committing or aborting, which is incorrect and causes errors.

3. Final Answer:

   Missing call to commitTransaction() or abortTransaction() before close() -> Option A

4. Quick Check:

   Transaction must end before close() [OK]

Hint: Always commit or abort before closing producer [OK]

Common Mistakes:

• Skipping commitTransaction() or abortTransaction()
• Calling beginTransaction() after send()
• Ignoring transaction lifecycle rules

5.

You want to send two messages atomically to two different topics using a Kafka transactional producer. Which sequence of calls is correct?

// Assume producer is configured with transactional.id
producer.initTransactions();
// What is the correct order?

hard

A. beginTransaction(); send(topic1); send(topic2); commitTransaction();

B. send(topic1); beginTransaction(); send(topic2); commitTransaction();

C. beginTransaction(); send(topic1); commitTransaction(); send(topic2);

D. send(topic1); send(topic2); beginTransaction(); commitTransaction();

Solution

1. Step 1: Understand transaction boundaries

   beginTransaction() must be called before sending any messages to start the transaction.

2. Step 2: Send all messages within the transaction and then commit

   Both messages must be sent after beginTransaction() and before commitTransaction() to be atomic.

3. Final Answer:

   beginTransaction(); send(topic1); send(topic2); commitTransaction(); -> Option A

4. Quick Check:

   Start transaction, send all, then commit [OK]

Hint: Begin transaction before sends, commit after all sends [OK]

Common Mistakes:

• Sending messages before beginTransaction()
• Committing before all messages sent
• Starting transaction after sending messages