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.