Demo: Transactional Kafka Producer¶

This demo shows the internals of transactional KafkaProducer that is a Kafka producer with transaction.id defined.

KafkaProducer¶

Start Up¶

Use sbt console for interactive environment (or IntelliJ IDEA).

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
val props = new Properties()
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, classOf[StringSerializer].getName)
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, ":9092")
props.put(ProducerConfig.CLIENT_ID_CONFIG, "txn-demo")
// Define transaction.id
val transactionalId = "my-custom-txnId"
props.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, transactionalId)
val producer = new KafkaProducer[String, String](props)

Initialize Transactions¶

producer.initTransactions

Once initialized, the transactional producer must not be initialized again.

scala> producer.initTransactions
org.apache.kafka.common.KafkaException: TransactionalId my-custom-txnId: Invalid transition attempted from state READY to state INITIALIZING
  at org.apache.kafka.clients.producer.internals.TransactionManager.transitionTo(TransactionManager.java:1078)
  at org.apache.kafka.clients.producer.internals.TransactionManager.transitionTo(TransactionManager.java:1071)
  at org.apache.kafka.clients.producer.internals.TransactionManager.lambda$initializeTransactions$1(TransactionManager.java:337)
  at org.apache.kafka.clients.producer.internals.TransactionManager.handleCachedTransactionRequestResult(TransactionManager.java:1200)
  at org.apache.kafka.clients.producer.internals.TransactionManager.initializeTransactions(TransactionManager.java:334)
  at org.apache.kafka.clients.producer.internals.TransactionManager.initializeTransactions(TransactionManager.java:329)
  at org.apache.kafka.clients.producer.KafkaProducer.initTransactions(KafkaProducer.java:596)
  ... 31 elided

Start Transaction¶

Next up is starting a transaction using KafkaProducer.beginTransaction

producer.beginTransaction

Transactional sends¶

import org.apache.kafka.clients.producer.ProducerRecord
val topic = "txn-demo"
val record = new ProducerRecord[String, String](topic, "Hello from transactional producer")
producer.send(record)

producer.send(
  new ProducerRecord[String, String](topic, "Another hello from txn producer"))

Logs¶

Kafka Producer¶

You should see the following INFO messages in the logs of the Kafka producer:

INFO [Producer clientId=producer-my-custom-txnId, transactionalId=my-custom-txnId] Invoking InitProducerId for the first time in order to acquire a producer ID (org.apache.kafka.clients.producer.internals.TransactionManager)
INFO [Producer clientId=producer-my-custom-txnId, transactionalId=my-custom-txnId] Discovered transaction coordinator localhost:9092 (id: 0 rack: null) (org.apache.kafka.clients.producer.internals.TransactionManager)
INFO [Producer clientId=producer-my-custom-txnId, transactionalId=my-custom-txnId] ProducerId set to 0 with epoch 0 (org.apache.kafka.clients.producer.internals.TransactionManager)

Kafka Cluster¶

You should see the following INFO message in the logs of a Kafka cluster:

INFO [TransactionCoordinator id=0] Initialized transactionalId my-custom-txnId with producerId 0 and producer epoch 0 on partition __transaction_state-20 (kafka.coordinator.transaction.TransactionCoordinator)

The calculation to determine the transactional partition (__transaction_state-20) is as follows:

Math.abs(transactionalId.hashCode) % 50

Start Up Consumer¶

kcat¶

kcat -C -b localhost -t txn-demo

You should see no records produced yet (since the transaction has not been committed yet).

kafka-console-consumer¶

./bin/kafka-console-consumer.sh \
  --bootstrap-server :9092 \
  --topic txn-demo \
  --from-beginning

Unlike kcat, kafka-console-consumer uses read_uncommitted isolation level and so there should be records printed out to the console.

Use --isolation-level option to set isolation.level configuration property.

Commit Transaction¶

Let's commit the transaction using KafkaProducer.commitTransaction.

producer.commitTransaction

Immediately after committing the transaction you should see the record printed out by the Kafka consumer.