Outbox pattern with Kafka: never lose events on database write

1. Overview

A service often needs to do two things in sequence: save data to database and emit events to Kafka for other services. These two systems don’t share a transaction. If one step succeeds while the other fails, the system loses consistency. The result is an order gets saved but no one receives the OrderPlaced event.

This article explains the dual write problem and implements the Outbox pattern with Java 21, Spring Boot 3.x, and Spring Kafka. By the end, you’ll have a message relay pushing events from the outbox table to Kafka, complete with tests and idempotent consumers.

2. The dual write problem

Look at this naive approach:

@Transactional
public Order placeOrder(String customerId, BigDecimal amount) {
    Order order = orderRepository.save(new Order(customerId, amount));
    kafkaTemplate.send("orders", order.getId().toString(), toJson(order));
    return order;
}

This code has at least two failure scenarios:

• Kafka send succeeds, transaction rolls back. For example, a commit error or constraint violation. A “ghost” event has leaked for an order that doesn’t exist.
• Transaction commits, Kafka send fails. The broker might be unreachable, or the app crashes right after commit. The order exists in the database but the event vanishes.

Swapping the order of the two operations doesn’t fix the root cause. Two operations on two different systems can’t be atomic without a distributed transaction. Most teams don’t want to use XA/2PC because it’s complex and slow.

3. The Outbox pattern idea

The Outbox pattern transforms “write database + send Kafka” into “write database + write database”:

1. In the same transaction, the service saves an Order and inserts a row into the outbox_event table.
2. A separate process called the message relay reads unpublished rows from the outbox table, sends them to Kafka, and marks them as published.

Since step 1 only touches one database, it’s absolutely atomic: if there’s an order, there’s guaranteed to be an event. Step 2 can be retried any number of times. In exchange, you get at-least-once semantics: an event might be sent as a duplicate, but never lost.

4. Implementation with Spring Boot

4.1. Dependency

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-data-jpa</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>postgresql</artifactId>
        <scope>runtime</scope>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    <dependency>
        <groupId>com.h2database</groupId>
        <artifactId>h2</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

Next, configure a safe producer:

spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer
spring.kafka.producer.acks=all
spring.kafka.producer.properties.enable.idempotence=true

acks=all forces the broker to acknowledge only after all in-sync replicas have written. enable.idempotence=true prevents the producer from creating duplicates when retrying at the network layer.

4.2. Outbox table and entity

CREATE TABLE outbox_event (
    id             UUID PRIMARY KEY,
    aggregate_type VARCHAR(100) NOT NULL,
    aggregate_id   VARCHAR(100) NOT NULL,
    event_type     VARCHAR(100) NOT NULL,
    payload        TEXT         NOT NULL,
    created_at     TIMESTAMP    NOT NULL,
    published_at   TIMESTAMP
);
CREATE INDEX idx_outbox_unpublished ON outbox_event (created_at) WHERE published_at IS NULL;

Partial index contains only unpublished rows. This keeps relay queries fast, even when the table has millions of old rows.

@Entity
@Table(name = "outbox_event")
public class OutboxEvent {

    @Id
    private UUID id;
    private String aggregateType;
    private String aggregateId;
    private String eventType;
    @Column(columnDefinition = "text")
    private String payload;
    private Instant createdAt;
    private Instant publishedAt;

    protected OutboxEvent() {
    }

    public OutboxEvent(String aggregateType, String aggregateId, String eventType, String payload) {
        this.id = UUID.randomUUID();
        this.aggregateType = aggregateType;
        this.aggregateId = aggregateId;
        this.eventType = eventType;
        this.payload = payload;
        this.createdAt = Instant.now();
    }

    public void markPublished(Instant at) {
        this.publishedAt = at;
    }

    public UUID getId() { return id; }
    public String getAggregateId() { return aggregateId; }
    public String getEventType() { return eventType; }
    public String getPayload() { return payload; }
    public Instant getPublishedAt() { return publishedAt; }
}

id also serves as the unique event id. Consumers use it to deduplicate (section 6). aggregateId becomes the key for the Kafka message.

4.3. Writing Order and event in the same transaction

public record OrderPlacedEvent(Long orderId, String customerId, BigDecimal amount) {
}

@Service
public class OrderService {

    private final OrderRepository orderRepository;
    private final OutboxEventRepository outboxRepository;
    private final ObjectMapper objectMapper;

    public OrderService(OrderRepository orderRepository,
                        OutboxEventRepository outboxRepository,
                        ObjectMapper objectMapper) {
        this.orderRepository = orderRepository;
        this.outboxRepository = outboxRepository;
        this.objectMapper = objectMapper;
    }

    @Transactional
    public Order placeOrder(String customerId, BigDecimal amount) {
        Order order = orderRepository.save(new Order(customerId, amount));

        var event = new OrderPlacedEvent(order.getId(), customerId, amount);
        outboxRepository.save(new OutboxEvent(
                "Order", order.getId().toString(), "OrderPlaced", toJson(event)));

        return order;
    }

    private String toJson(Object value) {
        try {
            return objectMapper.writeValueAsString(value);
        } catch (JsonProcessingException e) {
            throw new IllegalStateException("Could not serialize event", e);
        }
    }
}

The key point is @Transactional. Both save calls commit or roll back together. OrderService knows nothing about Kafka, so no network calls happen inside the transaction.

4.4. Message relay pushing events to Kafka

First, write a repository with row-locking queries:

public interface OutboxEventRepository extends JpaRepository<OutboxEvent, UUID> {

    @Query(value = """
            SELECT * FROM outbox_event
            WHERE published_at IS NULL
            ORDER BY created_at
            LIMIT :batchSize
            FOR UPDATE SKIP LOCKED
            """, nativeQuery = true)
    List<OutboxEvent> findBatchToPublish(@Param("batchSize") int batchSize);
}

FOR UPDATE SKIP LOCKED (PostgreSQL, MySQL 8+) allows running multiple app instances simultaneously. Each instance locks its own batch and skips rows other instances are processing. This ensures two instances never send the same event in parallel.

Next is the relay:

@Component
public class OutboxRelay {

    private static final Logger log = LoggerFactory.getLogger(OutboxRelay.class);

    private final OutboxEventRepository outboxRepository;
    private final KafkaTemplate<String, String> kafkaTemplate;

    public OutboxRelay(OutboxEventRepository outboxRepository,
                       KafkaTemplate<String, String> kafkaTemplate) {
        this.outboxRepository = outboxRepository;
        this.kafkaTemplate = kafkaTemplate;
    }

    @Scheduled(fixedDelay = 1000)
    @Transactional
    public void publishPending() {
        for (OutboxEvent event : outboxRepository.findBatchToPublish(100)) {
            try {
                kafkaTemplate.send(toRecord(event)).get(5, TimeUnit.SECONDS);
                event.markPublished(Instant.now());
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                return;
            } catch (Exception e) {
                log.warn("Sending event {} failed, will retry", event.getId(), e);
                break; // stop to preserve order; next run will retry from here
            }
        }
    }

    private ProducerRecord<String, String> toRecord(OutboxEvent event) {
        var record = new ProducerRecord<>("orders", event.getAggregateId(), event.getPayload());
        record.headers()
              .add("eventId", event.getId().toString().getBytes(StandardCharsets.UTF_8))
              .add("eventType", event.getEventType().getBytes(StandardCharsets.UTF_8));
        return record;
    }
}

Let’s break it down:

• @Scheduled(fixedDelay = 1000) runs the relay one second after the previous run finishes. Remember to add @EnableScheduling to a configuration class.
• .get(5, TimeUnit.SECONDS) waits for broker acknowledgment. The relay only marks publishedAt when certain Kafka has received the message.
• markPublished modifies the managed entity. JPA automatically generates the UPDATE on transaction commit; you don’t need to call save.
• On error, the relay breaks instead of skipping. Events from the same order thus never get reordered.
• The key is aggregateId, so all events for an Order land in the same partition and stay ordered on Kafka.

Log when the broker is temporarily down looks like:

WARN  OutboxRelay : Sending event 3f6c1e0a-... failed, will retry
org.apache.kafka.common.errors.TimeoutException: Topic orders not present in metadata after 5000 ms.

When the broker comes back, the next run continues from exactly that event.

Note: If the app crashes after Kafka receives a message but before the transaction commits, the outbox row still shows “unpublished.” The next run will send it again. This is at-least-once semantics, and consumers must handle duplicates.

5. Testing

The first test verifies that each order creates exactly one outbox row:

@DataJpaTest
@Import({OrderService.class, JacksonAutoConfiguration.class})
class OrderServiceTest {

    @Autowired
    OrderService orderService;

    @Autowired
    OutboxEventRepository outboxRepository;

    @Test
    void placeOrder_shouldWriteOutboxEventInSameTransaction() {
        Order order = orderService.placeOrder("customer-42", new BigDecimal("150000"));

        List<OutboxEvent> events = outboxRepository.findAll();

        assertThat(events).hasSize(1);
        OutboxEvent event = events.getFirst();
        assertThat(event.getAggregateId()).isEqualTo(order.getId().toString());
        assertThat(event.getEventType()).isEqualTo("OrderPlaced");
        assertThat(event.getPayload()).contains("\"customerId\":\"customer-42\"");
        assertThat(event.getPublishedAt()).isNull();
    }
}

Native queries aren’t validated on startup. So H2 can still run this test even though it doesn’t fully support SKIP LOCKED. To test the relay with real queries, use PostgreSQL via Testcontainers.

The second test ensures the relay doesn’t mark as sent when Kafka fails:

class OutboxRelayTest {

    OutboxEventRepository repository = mock(OutboxEventRepository.class);
    KafkaTemplate<String, String> kafkaTemplate = mock(KafkaTemplate.class);
    OutboxRelay relay = new OutboxRelay(repository, kafkaTemplate);

    @Test
    void shouldKeepEventUnpublished_whenKafkaFails() {
        var event = new OutboxEvent("Order", "1", "OrderPlaced", "{}");
        when(repository.findBatchToPublish(100)).thenReturn(List.of(event));
        when(kafkaTemplate.send(any(ProducerRecord.class)))
                .thenReturn(CompletableFuture.failedFuture(new RuntimeException("broker down")));

        relay.publishPending();

        assertThat(event.getPublishedAt()).isNull();
    }

    @Test
    void shouldMarkPublished_whenKafkaAcknowledges() {
        var event = new OutboxEvent("Order", "1", "OrderPlaced", "{}");
        when(repository.findBatchToPublish(100)).thenReturn(List.of(event));
        when(kafkaTemplate.send(any(ProducerRecord.class)))
                .thenReturn(CompletableFuture.completedFuture(null));

        relay.publishPending();

        assertThat(event.getPublishedAt()).isNotNull();
    }
}

These two tests lock in the core behavior: only mark as sent when the broker has acknowledged.

6. Consumer side: idempotent handling

Since duplicates are possible, consumers must remember which eventIds they’ve processed. This memory must live in the same transaction as the business logic:

@Component
public class InventoryListener {

    private final ProcessedEventRepository processedEvents;
    private final InventoryService inventoryService;

    public InventoryListener(ProcessedEventRepository processedEvents,
                             InventoryService inventoryService) {
        this.processedEvents = processedEvents;
        this.inventoryService = inventoryService;
    }

    @KafkaListener(topics = "orders", groupId = "inventory-service")
    @Transactional
    public void onOrderPlaced(@Header("eventId") String eventId, String payload) {
        if (processedEvents.existsById(eventId)) {
            return; // already processed, skip duplicate
        }
        inventoryService.reserveStock(payload);
        processedEvents.save(new ProcessedEvent(eventId));
    }
}

The processed_event table should have event_id as primary key. If two threads try to process the same duplicate, the constraint prevents one of them.

7. Common mistakes and when not to use

• Calling Kafka from OrderService ‘for speed’. That’s dual write again. Every event must go through the outbox table.
• Forgetting to clean the outbox table. It’ll grow huge fast. Have a job delete rows with published_at older than a few days.
• Skipping failed rows and continuing. This breaks event order. If a payload is permanently corrupt, transition it to FAILED status for investigation; don’t retry forever.
• Polling too frequently. A 100 ms interval with many instances puts significant load on the database. Intervals of 0.5–2 seconds are usually enough.
• Need very low latency or high throughput. Replace polling with CDC (Change Data Capture). For example, Debezium reads the WAL/binlog and pushes outbox rows straight to Kafka. The table stays the same; only the relay changes.

Note: Don’t use the Outbox pattern if events don’t matter, like metrics or user behavior logs that can tolerate loss. Then sending straight to Kafka is much simpler. Conversely, Kafka transactions (transactional.id) can’t replace Outbox: they only guarantee atomicity between Kafka topics, not your database.

8. Conclusion

Dual writes between database and Kafka always have gaps that lose or duplicate events. The Outbox pattern closes those gaps in three steps. First, write events to the outbox table in the same transaction as business data. Second, use a dedicated relay to push events to Kafka and only mark sent when the broker acknowledges. Third, implement idempotent consumers to absorb duplicates from at-least-once semantics.

You now have all the pieces: a schema with partial index, OrderService independent of Kafka, relay using FOR UPDATE SKIP LOCKED for multi-instance support, tests for core behavior, and deduplicating consumers. When your system grows, you can replace relay polling with CDC without changing business code.