How to Implement Exactly-Once Transactions in Spring Boot with Kafka

Core Concepts

Kafka transaction : Treats producing, consuming, and offset committing as a single atomic operation.

Transactional producer : Opens, commits or aborts a transaction; messages become visible to consumers only after a commit.

Transaction coordinator : Broker component that records transaction state and logs.

Transactional ID : Unique identifier for a producer instance; enables ordering and recovery after failures.

Isolation level : read_committed reads only messages from committed transactions; read_uncommitted (default) reads all messages.

Project Dependencies and Configuration

Add the Spring for Kafka starter to pom.xml:

<dependency>
  <groupId>org.springframework.kafka</groupId>
  <artifactId>spring-kafka</artifactId>
  <version>2.8.0</version>
</dependency>

Configure application.yml (replace your-kafka-server with the actual broker address):

spring:
  kafka:
    bootstrap-servers: your-kafka-server:9092
    producer:
      retries: 3
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
      transaction-id-prefix: tx-
      properties:
        enable.idempotence: true
    consumer:
      group-id: my-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      properties:
        isolation.level: read_committed

Practical Code Example

1. Event POJO

public class UserRegisteredEvent {
    private String userId;
    private String email;
    private java.time.LocalDateTime timestamp;
    // getters, setters, constructors omitted for brevity
}

2. Service Layer – combine DB and Kafka transaction

import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

@Service
public class UserService {
    private final UserRepository userRepository;
    private final KafkaTemplate<String, Object> kafkaTemplate;
    private static final String TOPIC_USER_REGISTERED = "topic.user.registered";

    public UserService(UserRepository userRepository, KafkaTemplate<String, Object> kafkaTemplate) {
        this.userRepository = userRepository;
        this.kafkaTemplate = kafkaTemplate;
    }

    @Transactional(rollbackFor = Exception.class)
    public User registerUser(User user) {
        // 1. Persist user in the database
        User savedUser = userRepository.save(user);
        // 2. Build the event object
        UserRegisteredEvent event = new UserRegisteredEvent();
        event.setUserId(savedUser.getId());
        event.setEmail(savedUser.getEmail());
        event.setTimestamp(java.time.LocalDateTime.now());
        // 3. Send the event to Kafka (participates in the same transaction)
        kafkaTemplate.send(TOPIC_USER_REGISTERED, savedUser.getId(), event);
        return savedUser;
    }
}

3. Consumer Listener

import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

@Component
public class UserEventListener {
    @KafkaListener(topics = "topic.user.registered")
    public void handleUserRegistered(UserRegisteredEvent event) {
        System.out.println("Received user registered event: " + event);
        // business logic such as sending a welcome email
    }
}

Execution Flow

Spring starts a combined database + Kafka transaction.

The service saves the user record to the database.

The Kafka producer writes the event to the transaction log (pre‑commit).

If the method returns normally, Spring commits both the database transaction and the Kafka transaction, making the message visible to consumers.

If an exception is thrown, Spring rolls back the database transaction and aborts the Kafka transaction, discarding the message.

Advanced Usage and Tips

Programmatic Kafka Transaction

@Autowired
private KafkaTransactionManager<String, Object> transactionManager;

public void someMethod() {
    TransactionTemplate template = new TransactionTemplate(transactionManager);
    template.execute(status -> {
        kafkaTemplate.send("someTopic", "key", "value");
        return null;
    });
}

Multiple Transaction Managers

Spring Boot can auto‑configure ChainedTransactionManager to coordinate a relational database transaction and a Kafka transaction in a single @Transactional boundary.

Transactional ID and Performance

Do not reuse the same transactional.id across many producer instances; each instance should have a unique ID.

The transaction-id-prefix property automatically appends a numeric suffix (e.g., tx-0, tx-1) to guarantee uniqueness.

Error Handling and Retries

Configure retries and delivery.timeout.ms in the producer settings; the transactional producer will retry internally before aborting.

When a transaction aborts, the surrounding Spring transaction is rolled back, preserving atomicity.

Summary

Enable a transactional producer by setting transaction-id-prefix and enable.idempotence=true in the producer configuration.

Wrap database and Kafka operations inside a single @Transactional method (or a programmatic TransactionTemplate) to achieve exactly‑once semantics.

Configure consumers with isolation.level=read_committed so they only receive messages from committed transactions.