In this article, you will learn how to use Kafka transactions with the Spring Kafka project in your Spring Boot app. In order to run the Kafka cluster we will use
Upstash
. This article provides a basic introduction to Kafka transactions. If you are looking for more advanced usage and scenarios you may refer to that
article
, about distributed transactions in microservices. You can also read more about Kafka Streams the Spring Cloud Stream project in this
article
.
public static void main(String[] args) {
SpringApplication.run(TransactionsService.class, args);
@Bean
public NewTopic transactionsTopic() {
return TopicBuilder.name("transactions")
.partitions(3)
.replicas(1)
.build();
In Kafka, a producer initiates a transaction by making a request to the transaction coordinator. You can find a detailed description of that process in the following
article
on the Confluent blog.
With Spring Boot, we just need to set the
spring.kafka.producer.transaction-id-prefix
property to enable transactions. Spring Boot will do the rest by automatically configuring a
KafkaTransactionManager
bean and wiring it into the listener container. Here’s a part of the configuration responsible for the message producer. We use JsonSerializer to serialize data from objects into JSON. Transactions prefix is
tx-
.
@Transactional
public void generateAndSendPackage()
throws InterruptedException, TransactionException {
for (long i = 0; i < 10; i++) {
Order t = new Order(id++, i+1, i+2, 1000, "NEW");
ListenableFuture<SendResult<Long, Order>> result =
kafkaTemplate.send("transactions", t.getId(), t);
result.addCallback(callback);
Thread.sleep(1000);
consumer:
key-deserializer: org.apache.kafka.common.serialization.LongDeserializer
value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
properties:
spring.json.trusted.packages: "*"
isolation.level: read_committed
@Transactional
public void generateAndSendPackage(boolean error)
throws InterruptedException {
for (long i = 0; i < 10; i++) {
Order t = new Order(id++, i+1, i+2, 1000, "NEW");
ListenableFuture<SendResult<Long, Order>> result =
kafkaTemplate.send("transactions", t.getId(), t);
result.addCallback(callback);
if (error && i > 5)
throw new RuntimeException();
Thread.sleep(1000);
It is important to know that Spring rollbacks are only on unchecked exceptions by default. To rollback checked exceptions, we need to specify the rollbackFor on the @Transactional annotation.
The transactional producer sends messages to the Kafka cluster even before committing the transaction. You could see it in the previous section, where the listener was continuously receiving messages if the isolation level was read_uncommited. Consequently, if we roll back a transaction on the producer side the message sent before rollback occurs come to the Kafka broker. We can see it e.g. in the Upstash live message view for the transactions topic.
Here’s the current value of offsets for all partitions in the transactions topic for the a consumer group. We made a successful commit after sending the first package of 10 messages and we rollbacked the transaction with the second package. The sum of offsets is 10 in that case. But in fact, it is different that the current latest offset on those partitions.
Of course, the messages have been rollbacked. But the important thing to understand here is that these operations happen on the Kafka broker side. The transaction coordinator changes the values of Kafka offsets. We can easily verify that consumer won’t receive messages after rollback even if we the initial offset to the earliest with the spring.kafka.consumer.auto-offset-reset property.
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
private String status;
private int totalNoOfOrders;
private int processedNoOfOrders;
// GETTERS/SETTERS ...
After including Spring Data JPA there are two registered TransactionManager beans with names transactionManager and kafkaTransactionManager. Therefore we need to choose the name of the transaction manager inside the @Transactional annotation. In the first step, we add a new entity to the database. The primary key id is auto-generated in the database and then returned to the object. After that, we get groupId and generate the sequence of orders within that group. Of course, both operations (save to database, sent to Kafka) are part of the same transaction.
@Transactional("kafkaTransactionManager")
public void sendOrderGroup(boolean error) throws InterruptedException {
OrderGroup og = repository.save(new OrderGroup("SENT", 10, 0));
generateAndSendPackage(error, og.getId());
private void generateAndSendPackage(boolean error, Long groupId)
throws InterruptedException {
for (long i = 0; i < 10; i++) {
Order o = new Order(id++, i+1, i+2, 1000, "NEW", groupId);
ListenableFuture<SendResult<Long, Order>> result =
kafkaTemplate.send("transactions", o.getId(), o);
result.addCallback(callback);
if (error && i > 5)
throw new RuntimeException();
Thread.sleep(1000);
The accounts-service app listens for incoming orders. It is processing every single order in a separate transaction. It checks if sufficient funds are in the customer account to make a transfer. If there is enough money, it performs a transfer. Finally, it sends the response to transactions-service with the transaction status. The message is sent to the orders topic.
The transactions-service listens for order confirmations on the orders topic. Once it receives the message it increases the number of processed orders within an order group and stores the current result in the database. We should use a default Spring transaction manager since we don’t send any messages to Kafka.
@KafkaListener(
id = "orders",
topics = "orders",
groupId = "a",
concurrency = "3")
@Transactional("transactionManager")
public void listen(Order order) {
LOG.info("{}", order);
OrderGroup og = repository
.findById(order.getGroupId())
.orElseThrow();
if (order.getStatus().equals("PROCESSED")) {
og.setProcessedNoOfOrders(og.getProcessedNoOfOrders() + 1);
og = repository.save(og);
LOG.info("Current: {}", og);
You can easily handle Kafka transactions with Spring Boot using the Spring Kafka project. You can integrate your app with a database and handle transactions across multiple resources. However, one thing needs to be clarified – Kafka does not support XA transactions. It may result in data inconsistency. Spring does not solve that case, it just performs two transactions in the background. When the @Transactional method exits, Spring Boot will commit the database transactions first and then the Kafka transactions. You can just change that order to enable Kafka transaction commit first by configuring the outer method configured to use the DataSourceTransactionManager, and the inner method to use the KafkaTransactionManager.
Can we solve that case somehow? Of course. There is, for example, project Debezium that allows you to stream database changes into Kafka topics. With that approach, you can just commit changes in the database, and then configure Debezium to send events with changes to Kafka. For more details about that tool and outbox pattern please refer to the article available here.
Or you could use JMS, which does support XA transactions.
Very nice article.
Loading...
nhan
Failed to start bean ‘org.springframework.kafka.config.internalKafkaListenerEndpointRegistry’; nested exception is org.apache.kafka.common.KafkaException: Failed to construct kafka consumer
please help me fix this error. Thanks.
Loading...
piotr.minkowski
Can you paste the whole stack trace? Did you create an instance of Kafka and provide connection settings in the app?
Loading...
Gaurav Khanna
The information on XA transactions and the way to get around it is very helpful
