Transactional Kafka with Spring — Consume-Process-Produce Atomically
Fast recall
Quick revision
Core answer
Understanding of exactly-once vs at-least-once and transaction overhead
Using Kafka transactions with a relational database — you need a database XA transaction or the Outbox pattern instead
Explain that Kafka transactions only work between Kafka topics, not across Kafka + DB
Interview-ready guide
Interview answer
The failure scenario: consumer reads a message, processes it, produces a downstream message, then crashes before committing the offset. On restart, everything happens again — duplicate processing, duplicate downstream messages.
Kafka transactions make consumed offset commit and produced messages atomic. Both commit or both roll back. Configure with spring.kafka.producer.transaction-id-prefix.
@Transactional on a listener starts a Kafka transaction. Downstream consumers must set `isolation.level=read_committed` to avoid reading uncommitted messages.
The hard problem is Kafka + database coordination. They're separate systems. If the database commits but the Kafka transaction fails, you have inconsistency.
The Outbox Pattern is the recommended solution: write business data and the outbound message to the database in one ACID transaction. A CDC tool (Debezium) publishes to Kafka separately. No Kafka transactions needed.
In practice, I use Kafka transactions for pure consume-transform-produce pipelines and the Outbox Pattern when a database is involved.
Transactional Kafka with Spring — Consume-Process-Produce Atomically
Direct Answer
Kafka transactions (exactly-once semantics) ensure a consume-process-produce cycle is atomic. Spring Kafka's @Transactional or KafkaTransactionManager makes this transparent. Requires a transactional producer with a unique transactional.id.
Code Example
1) Producer factory with transactional.id prefix:
@Configuration
public class TransactionalKafkaConfig {
@Bean
public ProducerFactory<String, OrderEvent> producerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
"kafka1:9092,kafka2:9092,kafka3:9092");
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
// Required for transactions:
props.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
props.put(ProducerConfig.ACKS_CONFIG, "all");
DefaultKafkaProducerFactory<String, OrderEvent> pf =
new DefaultKafkaProducerFactory<>(props);
// Spring appends a per-instance suffix to make transactional.id unique
// across replicas while keeping the prefix stable for fencing.
pf.setTransactionIdPrefix("order-svc-tx-");
return pf;
}
@Bean
public KafkaTransactionManager<String, OrderEvent> kafkaTxManager(
ProducerFactory<String, OrderEvent> pf) {
return new KafkaTransactionManager<>(pf);
}
@Bean
public KafkaTemplate<String, OrderEvent> kafkaTemplate(
ProducerFactory<String, OrderEvent> pf) {
return new KafkaTemplate<>(pf);
}
// Consumer side MUST use read_committed to skip uncommitted/aborted records.
@Bean
public ConcurrentKafkaListenerContainerFactory<String, OrderEvent>
kafkaListenerContainerFactory(ConsumerFactory<String, OrderEvent> cf,
KafkaTransactionManager<String, OrderEvent> tm) {
ConcurrentKafkaListenerContainerFactory<String, OrderEvent> f =
new ConcurrentKafkaListenerContainerFactory<>();
f.setConsumerFactory(cf);
f.getContainerProperties().setKafkaAwareTransactionManager(tm);
return f;
}
}
2) application.yml:
spring:
kafka:
consumer:
isolation-level: read_committed
enable-auto-commit: false
properties:
spring.json.trusted.packages: com.example.events
3) Atomic consume-process-produce — @Transactional ties offset commit + sends together:
@Service
public class OrderProcessor {
private final KafkaTemplate<String, OrderEvent> kafka;
public OrderProcessor(KafkaTemplate<String, OrderEvent> kafka) {
this.kafka = kafka;
}
@KafkaListener(topics = "orders", groupId = "order-svc")
@Transactional("kafkaTxManager")
public void process(OrderEvent in) {
// Both sends + the consumer offset commit are part of ONE Kafka transaction.
// If this method throws, ALL THREE roll back atomically.
kafka.send("inventory-updates", in.orderId(), new InventoryAdjust(in));
kafka.send("email-queue", in.orderId(), new EmailRequest(in));
}
}
4) Programmatic alternative — executeInTransaction for one-off sends:
@Service
public class BatchOrderPublisher {
private final KafkaTemplate<String, OrderEvent> kafka;
public void publishBatch(List<OrderEvent> batch) {
kafka.executeInTransaction(t -> {
for (OrderEvent e : batch) {
t.send("orders", e.orderId(), e);
}
return null; // commits if no exception; aborts on any throw
});
}
}
Critical points:
transactional.idMUST be stable per logical producer instance — Spring's prefix + per-instance suffix handles this correctly. A changingtransactional.iddefeats the broker's zombie-fencing protection.- The consumer container needs the
KafkaAwareTransactionManagerso the offset commit becomes part of the transaction. - All downstream consumers MUST set
isolation.level=read_committed, otherwise they will see records from transactions that are still in-flight or get aborted.
Common Mistakes & Pitfalls
Using Kafka transactions with a relational database — you need a database XA transaction or the Outbox pattern instead
Check your understanding
Answer these without looking back at the article.
- Why can't Kafka transactions span both Kafka and a relational database?
- How does the Outbox pattern solve Kafka + database atomicity?
- What is isolation.level=read_committed and why is it required for consumers?
Related interview prep categories
2Want a guided curriculum order?
These modules are also sequenced in our flagship backend interview track — open it when you prefer an ordered path instead of browsing by topic.