You need to save an order to the database and publish an event to Kafka. That sounds simple until you ask: what happens if the database write succeeds and the Kafka publish fails? Or the other way around? You end up with a phantom event in Kafka and no order in the database, or an order that silently disappears from the event stream.
The transactional outbox pattern solves this. Instead of writing to the database and Kafka separately, you write to the database twice in one transaction. A separate relay process then handles the Kafka part. All code for this tutorial is in the companion repo: github.com/umur/transactional-outbox. Run mvn verify and all 7 tests pass.
The Problem
The naive approach looks like this:
@Transactional
public OrderResponse place(PlaceOrderRequest request) {
Order order = orderRepository.save(...);
kafkaTemplate.send("orders", buildPayload(order)); // what if this fails?
return new OrderResponse(...);
}This has two failure modes. If Kafka is down, the database write succeeds but the event never goes out. If the service crashes after Kafka publish but before the database commits, the event is published for a transaction that was rolled back. Either way, your downstream consumers get a different version of reality than your database.
You cannot make a database transaction and a Kafka send atomic. They are different systems. But you can make two database writes atomic, which is exactly what the outbox pattern does.
The Outbox Table
The outbox table sits next to your business tables in the same database. It holds messages that need to be published to Kafka.
@Entity
@Table(name = "outbox")
public class OutboxMessage {
@Id
@GeneratedValue(strategy = GenerationType.IDENTITY)
private Long id;
@Column(nullable = false)
private String topic;
@Column(nullable = false)
private String aggregateId; // used as the Kafka message key
@Column(nullable = false, columnDefinition = "TEXT")
private String payload; // JSON string
@Column(nullable = false)
@Builder.Default
private boolean published = false;
@Column(nullable = false)
@Builder.Default
private Instant createdAt = Instant.now();
}The aggregateId becomes the Kafka message key. Using the order ID as the key means all events for the same order land on the same partition and stay ordered.
The Atomic Write
OrderService writes both rows inside a single @Transactional method. Either both rows land in the database or neither does.
@Service
@RequiredArgsConstructor
@Transactional
public class OrderService {
private final OrderRepository orderRepository;
private final OutboxMessageRepository outboxMessageRepository;
public OrderResponse place(PlaceOrderRequest request) {
Order order = orderRepository.save(
Order.builder()
.userId(request.userId())
.total(request.total())
.build()
);
outboxMessageRepository.save(
OutboxMessage.builder()
.topic("orders")
.aggregateId(String.valueOf(order.getId()))
.payload(buildPayload(order))
.build()
);
return new OrderResponse(order.getId(), order.getUserId(), order.getTotal(), order.getStatus());
}
private String buildPayload(Order order) {
return """
{"event":"OrderPlaced","orderId":%d,"userId":"%s","total":%s,"status":"%s"}
""".formatted(order.getId(), order.getUserId(), order.getTotal(), order.getStatus()).strip();
}
}Kafka is not touched here at all. The service only cares about the database. If anything goes wrong, Spring rolls back the whole transaction and the outbox row never exists.
The Relay
OutboxRelay runs on a fixed schedule. It reads every unpublished row in order, sends each one to Kafka, and marks it published.
@Component
@RequiredArgsConstructor
@Slf4j
public class OutboxRelay {
private final OutboxMessageRepository outboxMessageRepository;
private final KafkaTemplate<String, String> kafkaTemplate;
@Scheduled(fixedDelayString = "${outbox.relay.interval-ms:500}")
public void scheduledProcess() {
process();
}
@Transactional
public void process() {
List<OutboxMessage> pending =
outboxMessageRepository.findByPublishedFalseOrderByCreatedAtAsc();
if (pending.isEmpty()) {
return;
}
for (OutboxMessage message : pending) {
kafkaTemplate.send(message.getTopic(), message.getAggregateId(), message.getPayload())
.whenComplete((result, ex) -> {
if (ex != null) {
log.error("Failed to publish outbox message id={}", message.getId(), ex);
}
});
message.setPublished(true);
outboxMessageRepository.save(message);
}
}
}The @Scheduled method delegates to process() so integration tests can trigger it directly without waiting for the scheduler. The interval is configurable via outbox.relay.interval-ms.
At-Least-Once Delivery
Notice that message.setPublished(true) happens after the Kafka send, not inside a Kafka callback. If the relay sends to Kafka successfully and then crashes before the database update commits, the message will be sent again on the next poll. That is at-least-once delivery.
This is the right tradeoff for most systems. Exactly-once requires Kafka transactions. That adds a producer-side transaction protocol, broker-side transaction state, and consumer-side read_committed semantics. At-least-once with idempotent consumers is simpler and covers the majority of real use cases.
Making a consumer idempotent usually means tracking which event IDs it has already processed. Deduplicate on aggregateId or on a stable event identifier included in the payload.
Kafka Configuration
Spring Boot 4 does not auto-configure a KafkaTemplate bean by default, so the config is explicit:
@Configuration
public class KafkaConfig {
private final String bootstrapServers;
public KafkaConfig(@Value("${spring.kafka.bootstrap-servers}") String bootstrapServers) {
this.bootstrapServers = bootstrapServers;
}
@Bean
public ProducerFactory<String, String> producerFactory() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.ACKS_CONFIG, "all");
props.put(ProducerConfig.RETRIES_CONFIG, 3);
return new DefaultKafkaProducerFactory<>(props);
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<>(producerFactory());
}
}ACKS_CONFIG = "all" means the producer waits for acknowledgment from all in-sync replicas before considering a send successful. That is the right setting for a reliability-focused pattern like this one.
Testing the Atomic Write
OrderServiceIT uses Testcontainers with a real PostgreSQL instance. The Kafka template is mocked out because this test is only verifying the database side of the pattern.
@SpringBootTest
@Testcontainers
class OrderServiceIT {
@Container
static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16");
@DynamicPropertySource
static void configureProperties(DynamicPropertyRegistry registry) {
registry.add("spring.datasource.url", postgres::getJdbcUrl);
registry.add("spring.datasource.username", postgres::getUsername);
registry.add("spring.datasource.password", postgres::getPassword);
}
@MockitoBean
private KafkaTemplate<String, String> kafkaTemplate;
@Autowired
private OrderService orderService;
@Test
@DisplayName("placing an order persists both the order and the outbox message")
void placeOrder_persistsOrderAndOutboxMessage() {
orderService.place(new PlaceOrderRequest("user-123", new BigDecimal("59.99")));
assertThat(orderRepository.findAll()).hasSize(1);
List<OutboxMessage> messages = outboxMessageRepository.findAll();
assertThat(messages).hasSize(1);
assertThat(messages.get(0).getTopic()).isEqualTo("orders");
assertThat(messages.get(0).getPayload()).contains("OrderPlaced");
assertThat(messages.get(0).isPublished()).isFalse();
}
}Note: Spring Boot 4 moved @MockBean to @MockitoBean from the org.springframework.test.context.bean.override.mockito package.
Testing the Relay End-to-End
OutboxRelayIT spins up both PostgreSQL and Kafka containers. The scheduling interval is set to a huge number so only manual process() calls run during the test. A Kafka consumer subscribes before each test places any orders so it only sees messages from that specific test run.
@SpringBootTest(properties = "outbox.relay.interval-ms=2147483647")
@Testcontainers
class OutboxRelayIT {
@Container
static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:16");
@Container
static KafkaContainer kafka = new KafkaContainer(
DockerImageName.parse("confluentinc/cp-kafka:7.6.0"));
@Autowired
private OutboxRelay outboxRelay;
private KafkaConsumer<String, String> consumer;
@BeforeEach
void setUp() {
outboxMessageRepository.deleteAll();
orderRepository.deleteAll();
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-consumer-" + System.nanoTime());
props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("orders"));
consumer.poll(Duration.ofMillis(500)); // trigger partition assignment and seek to current end
}
@Test
@DisplayName("relay publishes unpublished outbox messages to Kafka")
void relay_publishesOutboxMessages() {
orderService.place(new PlaceOrderRequest("user-1", new BigDecimal("49.99")));
outboxRelay.process();
List<String> received = poll(1);
assertThat(received).hasSize(1);
assertThat(received.get(0)).contains("OrderPlaced");
}
@Test
@DisplayName("relay does not re-publish already published messages")
void relay_skipsAlreadyPublishedMessages() {
orderService.place(new PlaceOrderRequest("user-3", new BigDecimal("9.99")));
outboxRelay.process(); // first run - publishes
outboxRelay.process(); // second run - should skip
List<String> received = poll(1);
assertThat(received).hasSize(1); // still only 1 message in Kafka
}
}The consumer subscribes with AUTO_OFFSET_RESET_CONFIG = "latest" and does an empty poll to trigger partition assignment. This positions it at the current end of the topic, so the test only reads messages produced during its own run. If you use "earliest" instead, the consumer reads every message that has ever been written to that topic across all tests and the assertions fail.
Project Structure
src/main/java/com/umurinan/outbox/
├── entity/
│ ├── Order.java
│ └── OutboxMessage.java
├── repository/
│ ├── OrderRepository.java
│ └── OutboxMessageRepository.java
├── service/
│ ├── OrderService.java # atomic write: Order + OutboxMessage
│ ├── PlaceOrderRequest.java
│ └── OrderResponse.java
├── relay/
│ └── OutboxRelay.java # polls outbox, publishes to Kafka
├── controller/
│ └── OrderController.java
└── config/
└── KafkaConfig.javaWhen to Use This Pattern
The outbox pattern makes sense when:
- You need events to reflect database state exactly, with no phantom events and no missed events
- Your downstream consumers need reliable delivery and can handle duplicates
- You control the producer side and can add the outbox table
It is overkill when you are publishing fire-and-forget notifications where occasional misses are acceptable, or when you are already using a framework that handles transactional messaging for you.
The polling relay introduced here is the simplest implementation. A production system might replace it with change data capture using Debezium, which reads directly from the Postgres write-ahead log and eliminates the polling delay entirely. Polling, though, works well for most use cases and is far easier to operate.
Clone the repo, run mvn verify, and the full test suite runs against real PostgreSQL and Kafka containers. The relay and atomicity guarantees are all verified.