Transactional producer in Kafka - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Kafka transactional producer code snippet?

Consider the following Java Kafka producer code using transactions. What will be printed to the console?

Kafka

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("enable.idempotence", "true");
props.put("transactional.id", "txn-1");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("my-topic", "key1", "value1"));
    producer.commitTransaction();
    System.out.println("Transaction committed");
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    producer.close();
    System.out.println("Fatal error, producer closed");
} catch (KafkaException e) {
    producer.abortTransaction();
    System.out.println("Transaction aborted");
}

A Transaction committed

B Transaction aborted

C Fatal error, producer closed

D No output, program hangs

💡 Hint

Think about what happens when no exceptions are thrown during the transaction.

The code begins a transaction, sends a message, and commits the transaction without errors. So it prints "Transaction committed".

❓ Predict Output

intermediate

What error does this Kafka transactional producer code raise?

Examine this Kafka producer code snippet. What error will it raise when run?

Kafka

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
// Missing transactional.id property

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

A No error, runs fine

B IllegalStateException

C InvalidConfigurationException

D TimeoutException

💡 Hint

Transactional producers require a specific property to be set before calling initTransactions().

Without the "transactional.id" property, calling initTransactions() throws IllegalStateException because the producer cannot initialize transactions.

🔧 Debug

advanced

Why does this Kafka transactional producer code abort the transaction?

Look at this code snippet. Why does it print "Transaction aborted"?

Kafka

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("enable.idempotence", "true");
props.put("transactional.id", "txn-2");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

try {
    producer.beginTransaction();
    producer.send(new ProducerRecord<>("my-topic", "key2", "value2"));
    throw new KafkaException("Simulated failure");
    // producer.commitTransaction();
} catch (ProducerFencedException | OutOfOrderSequenceException | AuthorizationException e) {
    producer.close();
    System.out.println("Fatal error, producer closed");
} catch (KafkaException e) {
    producer.abortTransaction();
    System.out.println("Transaction aborted");
}

A Because a KafkaException was thrown before commitTransaction()

B Because the producer was closed before commitTransaction()

C Because the transactional.id was invalid

D Because enable.idempotence was set to false

💡 Hint

Check what happens when an exception occurs after beginTransaction() but before commitTransaction().

The code throws a KafkaException before commitTransaction(), so the catch block aborts the transaction and prints "Transaction aborted".

📝 Syntax

advanced

Which option contains the correct syntax to begin and commit a Kafka transaction?

Choose the code snippet that correctly begins and commits a Kafka transaction.

A

producer.beginTransaction();
producer.send(record)
producer.commitTransaction

B

producer.beginTransaction;
producer.send(record);
producer.commitTransaction();

C

producer.beginTransaction()
producer.send(record)
producer.commitTransaction()

D

producer.beginTransaction();
producer.send(record);
producer.commitTransaction();

💡 Hint

Remember Java method calls require parentheses and semicolons.

Option D correctly uses parentheses and semicolons for method calls. Others have syntax errors like missing parentheses or semicolons.

🚀 Application

expert

How many messages are guaranteed to be written atomically in this transactional producer code?

Given the following code, how many messages will be committed atomically to Kafka?

Kafka

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
props.put("enable.idempotence", "true");
props.put("transactional.id", "txn-3");

KafkaProducer<String, String> producer = new KafkaProducer<>(props);
producer.initTransactions();

producer.beginTransaction();
producer.send(new ProducerRecord<>("topic1", "key1", "value1"));
producer.send(new ProducerRecord<>("topic2", "key2", "value2"));
producer.send(new ProducerRecord<>("topic3", "key3", "value3"));
producer.commitTransaction();

A 0 messages

B 1 message

C 3 messages

D Depends on the number of partitions

💡 Hint

Think about what commitTransaction() does with all messages sent after beginTransaction().

All messages sent between beginTransaction() and commitTransaction() are committed atomically as a single transaction, so 3 messages are committed together.

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