The Transactional Outbox Pattern
Store outgoing events in the same database transaction as your data, then relay them to the broker, so a crash can never save one without the other.
@Transactional business method: save(entity) + save(outboxRow)
@Scheduled relay: claim batch in a short statement (SKIP LOCKED, lease) -> send and wait, no transaction open -> mark sent rows published, release the restA service that saves an order and then publishes OrderPlaced is doing two writes to two systems, and no transaction spans both. If the broker is down, the order exists but nobody hears about it. If you publish first and the database commit fails, consumers react to an order that does not exist. Publishing inside a @Transactional method does not help: the send happens immediately, while the commit may still roll back.
The pattern
- In the same transaction that changes your data, insert a row into an
outbox_eventtable holding the event type, the key and the JSON payload. - A separate relay reads unpublished rows, sends each to the broker, waits for the broker to confirm, and marks the row published.
The database commit is now the single point of truth: either the order and its event both exist, or neither does. The relay turns that into at-least-once delivery. If it crashes after sending but before marking the row, the event is sent again on the next run, so consumers still need to be idempotent, keyed on the event id.
@Transactional
public Order place(NewOrder cmd) {
Order order = orders.save(Order.from(cmd));
outbox.save(OutboxEvent.of("orders", order.id(), "order.placed", json(order)));
return order;
}Running the relay
A @Scheduled method that polls every few hundred milliseconds is the simplest relay. Claim a batch with FOR UPDATE SKIP LOCKED so several application instances can poll at once without sending the same row twice. Wait for each send to finish before marking the row; a fire-and-forget send would mark rows whose records never arrived.
Do not hold a database transaction open while waiting on Kafka. A slow or unreachable broker then keeps row locks and a pooled connection for the whole timeout, the pool runs dry and ordinary requests start failing. Use short transactions instead: one statement claims the batch by setting a lease column, the sends happen with no transaction open, and a second statement marks the sent rows published. If a send fails, the relay stops and clears the lease on every row it did not send, so the next poll retries them first, in order. A lease that simply runs out, because an instance died mid-batch, frees its rows the same way.
Order matters. Several relays running in parallel can deliver two events for one order out of sequence. If consumers depend on per-aggregate order, order the batch by creation time and run a single relay, or have consumers compare a version number and ignore stale events.
Housekeeping and alternatives
The table grows forever unless you delete published rows after a few days. Index the unpublished rows, or the poll becomes a full table scan. Alert when the oldest unpublished row is more than a minute old: that means the relay is stuck.
Change data capture tools such as Debezium read the outbox table from the database's write-ahead log instead of polling, which lowers latency at the cost of running another component. Spring Modulith offers an event publication registry that records events in the database for the same reason; it is a separate project you add yourself.
Example
import jakarta.persistence.Column;
import jakarta.persistence.Entity;
import jakarta.persistence.GeneratedValue;
import jakarta.persistence.GenerationType;
import jakarta.persistence.Id;
import jakarta.persistence.Table;
import java.math.BigDecimal;
import java.time.Instant;
import java.util.UUID;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import tools.jackson.databind.json.JsonMapper;
(name = "orders")
class PurchaseOrder {
(strategy = GenerationType.IDENTITY)
private Long id;
private long customerId;
private BigDecimal total;
protected PurchaseOrder() {
}
PurchaseOrder(long customerId, BigDecimal total) {
this.customerId = customerId;
this.total = total;
}
Long getId() { return id; }
long getCustomerId() { return customerId; }
BigDecimal getTotal() { return total; }
}
(name = "outbox_event")
class OutboxEvent {
private UUID id;
private String topic;
private String messageKey;
private String type;
(columnDefinition = "text")
private String payload;
private Instant createdAt;
private Instant publishedAt;
protected OutboxEvent() {
}
OutboxEvent(String topic, String messageKey, String type, String payload) {
this.id = UUID.randomUUID();
this.topic = topic;
this.messageKey = messageKey;
this.type = type;
this.payload = payload;
this.createdAt = Instant.now();
}
UUID getId() { return id; }
}
record OrderPlaced(UUID eventId, long orderId, long customerId, BigDecimal total) {
}
interface PurchaseOrderRepository extends JpaRepository<PurchaseOrder, Long> {
}
interface OutboxEventRepository extends JpaRepository<OutboxEvent, UUID> {
}
public class OrderService {
private final PurchaseOrderRepository orders;
private final OutboxEventRepository outbox;
private final JsonMapper json;
OrderService(PurchaseOrderRepository orders, OutboxEventRepository outbox, JsonMapper json) {
this.orders = orders;
this.outbox = outbox;
this.json = json;
}
public long place(long customerId, BigDecimal total) {
PurchaseOrder order = orders.save(new PurchaseOrder(customerId, total));
OrderPlaced event = new OrderPlaced(UUID.randomUUID(), order.getId(), customerId, total);
outbox.save(new OutboxEvent("orders", String.valueOf(order.getId()), "order.placed",
json.writeValueAsString(event)));
return order.getId();
}
}
When to use it
- Guarantee that every committed payment produces exactly one PaymentCaptured event, even when Kafka is briefly unavailable.
- Move from publishing inside service methods to an outbox when audits find orders that downstream systems never heard about.
- Feed a search index from database changes without dual writes that drift apart after a failed request.
More examples
The outbox table with a partial index
Ship this as a Flyway migration next to the business tables. The partial index covers only unpublished rows, so it stays small however many published rows the table holds, and the relay's <code>ORDER BY created_at</code> query reads it directly. <code>claimed_until</code> is the relay's lease: a claimed row is skipped by other instances until it is published or the lease runs out.
CREATE TABLE outbox_event (
id UUID PRIMARY KEY,
topic VARCHAR(200) NOT NULL,
message_key VARCHAR(200) NOT NULL,
type VARCHAR(100) NOT NULL,
payload TEXT NOT NULL,
created_at TIMESTAMP WITH TIME ZONE NOT NULL,
claimed_until TIMESTAMP WITH TIME ZONE,
published_at TIMESTAMP WITH TIME ZONE
);
CREATE INDEX outbox_event_unpublished_idx
ON outbox_event (created_at)
WHERE published_at IS NULL;
A polling relay that keeps transactions short
No transaction is open while the relay waits on Kafka. The claim is one auto-committed statement: it locks the next rows with <code>SKIP LOCKED</code>, sets a 30-second lease so other instances skip them, and returns them. <code>RETURNING</code> does not keep the subquery's order, so the batch is sorted again in Java. Each send is awaited with <code>join()</code>; the first failure stops the batch, only the acknowledged rows are marked published, and the lease on the rest is cleared, all in short statements of their own. Because the unsent rows are released rather than left leased, the next poll claims them before anything newer, so a later event for the same key cannot overtake a failed one while a single relay runs. The 20-second deadline keeps a slow batch inside its lease. <code>send()</code> itself can block for the producer's <code>max.block.ms</code> (60 seconds by default) when the broker is unreachable, so lower that setting below the lease as well. The payload is already JSON, so this producer uses <code>StringSerializer</code> for values. Remember <code>@EnableScheduling</code> on a configuration class, or the method never runs.
import java.time.OffsetDateTime;
import java.util.ArrayList;
import java.util.Comparator;
import java.util.List;
import java.util.Set;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
record PendingEvent(UUID id, String topic, String messageKey, String payload, OffsetDateTime createdAt) {
}
@Component
public class OutboxRelay {
private static final Logger log = LoggerFactory.getLogger(OutboxRelay.class);
private final JdbcClient jdbc;
private final KafkaTemplate<String, String> kafka;
OutboxRelay(JdbcClient jdbc, KafkaTemplate<String, String> kafka) {
this.jdbc = jdbc;
this.kafka = kafka;
}
@Scheduled(fixedDelay = 500)
public void publishPending() {
List<PendingEvent> batch = claimNextBatch(50);
List<UUID> sent = new ArrayList<>();
long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(20);
try {
for (PendingEvent event : batch) {
if (System.nanoTime() > deadline) {
break;
}
kafka.send(event.topic(), event.messageKey(), event.payload())
.orTimeout(5, TimeUnit.SECONDS)
.join();
sent.add(event.id());
}
} catch (RuntimeException ex) {
log.warn("Outbox relay stopped after {} of {} events", sent.size(), batch.size(), ex);
} finally {
markPublished(sent);
Set<UUID> done = Set.copyOf(sent);
releaseClaim(batch.stream().map(PendingEvent::id).filter(id -> !done.contains(id)).toList());
}
}
private List<PendingEvent> claimNextBatch(int limit) {
return jdbc.sql("""
UPDATE outbox_event
SET claimed_until = now() + interval '30 seconds'
WHERE id IN (
SELECT id FROM outbox_event
WHERE published_at IS NULL
AND (claimed_until IS NULL OR claimed_until < now())
ORDER BY created_at
LIMIT :limit
FOR UPDATE SKIP LOCKED)
RETURNING id, topic, message_key, payload, created_at
""")
.param("limit", limit)
.query((rs, rowNum) -> new PendingEvent(
rs.getObject("id", UUID.class),
rs.getString("topic"),
rs.getString("message_key"),
rs.getString("payload"),
rs.getObject("created_at", OffsetDateTime.class)))
.list()
.stream()
.sorted(Comparator.comparing(PendingEvent::createdAt))
.toList();
}
private void markPublished(List<UUID> ids) {
if (ids.isEmpty()) {
return;
}
jdbc.sql("UPDATE outbox_event SET published_at = now() WHERE id IN (:ids)")
.param("ids", ids)
.update();
}
private void releaseClaim(List<UUID> ids) {
if (ids.isEmpty()) {
return;
}
jdbc.sql("UPDATE outbox_event SET claimed_until = NULL WHERE id IN (:ids)")
.param("ids", ids)
.update();
}
}
Deleting published rows
Keeping a week of published rows lets you answer "was this event ever sent?" during an incident, and the nightly delete stops the table from growing without bound. The cutoff is an <code>OffsetDateTime</code> because JDBC drivers such as PostgreSQL's bind it directly, while a raw <code>Instant</code> parameter is rejected. On very busy tables delete in batches, so one statement does not hold locks for minutes.
import java.time.OffsetDateTime;
import java.time.ZoneOffset;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.jdbc.core.simple.JdbcClient;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Component;
@Component
public class OutboxCleanup {
private static final Logger log = LoggerFactory.getLogger(OutboxCleanup.class);
private final JdbcClient jdbc;
OutboxCleanup(JdbcClient jdbc) {
this.jdbc = jdbc;
}
@Scheduled(cron = "0 15 3 * * *")
public void deleteOldPublishedEvents() {
OffsetDateTime cutoff = OffsetDateTime.now(ZoneOffset.UTC).minusDays(7);
int deleted = jdbc.sql("DELETE FROM outbox_event WHERE published_at < :cutoff")
.param("cutoff", cutoff)
.update();
log.info("Deleted {} published outbox events older than {}", deleted, cutoff);
}
}
Discussion