Spring Boot 3 + Kafka Outbox: From Transactional Consistency to Production‑Ready HA

Why the Outbox pattern is needed

In distributed systems the classic problem is that a business transaction may be committed while the subsequent Kafka message fails to be delivered, leaving the system in an inconsistent state. Two naïve solutions – writing to the database then sending to Kafka, or sending to Kafka then writing to the database – both break atomicity when a crash occurs between the two steps.

Two‑phase commit (2PC/XA) is theoretically correct but impractical for modern microservices because middleware often does not support XA, the performance overhead is high, operational complexity is large, and it conflicts with cloud‑native autonomy.

Transactional Outbox as the practical solution

The recommended approach is the Transactional Outbox : within a single local transaction the application writes the business data and an outbox_event record. A separate Relay component later reads pending rows, sends them to Kafka, and updates the row status.

The core guarantees are:

Atomic write of business data and event record.

At‑least‑once delivery with idempotent consumers.

Full recoverability: if the app crashes after committing, the Relay will still find the pending row.

Outbox problem scope

Atomicity : order creation and event generation succeed or fail together.

Recoverability : after a crash the event remains in the table and can be retried.

Observability : the event table can be queried, audited, retried, or manually fixed.

When to use the Outbox

Strong business consistency (order, payment, inventory, marketing).

Event‑driven architectures where services communicate via Kafka.

Scenarios tolerating sub‑second latency but not allowing message loss.

Do not use it for ultra‑low latency (< 1 ms) or for low‑value logging where occasional loss is acceptable.

Implementation choices

Polling Publisher : a simple periodic poll that works for most teams.

CDC Outbox : lower latency and tighter integration with change‑data‑capture tools (Debezium, Kafka Connect) but adds operational complexity.

Database schema

CREATE TABLE outbox_event (
    id BIGSERIAL PRIMARY KEY,
    event_id VARCHAR(64) NOT NULL UNIQUE,
    aggregate_type VARCHAR(64) NOT NULL,
    aggregate_id VARCHAR(64) NOT NULL,
    event_type VARCHAR(128) NOT NULL,
    partition_key VARCHAR(128) NOT NULL,
    topic VARCHAR(128) NOT NULL,
    payload JSONB NOT NULL,
    headers JSONB NOT NULL DEFAULT '{}'::jsonb,
    status VARCHAR(32) NOT NULL,
    retry_count INTEGER NOT NULL DEFAULT 0,
    max_retry INTEGER NOT NULL DEFAULT 16,
    next_retry_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    processing_owner VARCHAR(128),
    processing_at TIMESTAMPTZ,
    published_at TIMESTAMPTZ,
    last_error TEXT,
    shard_key INTEGER NOT NULL DEFAULT 0,
    version BIGINT NOT NULL DEFAULT 0,
    created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
    updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW()
);

CREATE INDEX idx_outbox_pending_scan ON outbox_event(status, next_retry_at, created_at, id);
CREATE INDEX idx_outbox_owner_scan ON outbox_event(processing_owner, processing_at);
CREATE INDEX idx_outbox_shard_scan ON outbox_event(shard_key, status, next_retry_at, created_at, id);

Key columns

status

– state machine (PENDING, PROCESSING, PUBLISHED, FAILED). shard_key – logical partition for horizontal scaling. next_retry_at – supports exponential back‑off.

State machine

PENDING → PROCESSING → PUBLISHED
PENDING → PROCESSING → PENDING   (retry)
PENDING → PROCESSING → FAILED    (max retries exceeded)

Domain model examples

OutboxStatus enum

public enum OutboxStatus {
    PENDING,
    PROCESSING,
    PUBLISHED,
    FAILED
}

Order entity (simplified)

@Entity
@Table(name = "orders")
public class Order {
    @Id @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String orderNo;
    private Long userId;
    private BigDecimal amount;
    @Enumerated(EnumType.STRING)
    private OrderStatus status;
    private Instant createdAt;
    private Instant updatedAt;
    @PrePersist
    void prePersist() { Instant now = Instant.now(); createdAt = now; updatedAt = now; if (status == null) status = OrderStatus.CREATED; }
    @PreUpdate
    void preUpdate() { updatedAt = Instant.now(); }
}

OutboxEvent entity

@Entity
@Table(name = "outbox_event")
public class OutboxEvent {
    @Id @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    private String eventId;
    private String aggregateType;
    private String aggregateId;
    private String eventType;
    private String partitionKey;
    private String topic;
    @Column(columnDefinition = "jsonb")
    private String payload;
    @Column(columnDefinition = "jsonb")
    private String headers;
    @Enumerated(EnumType.STRING)
    private OutboxStatus status;
    private Integer retryCount;
    private Integer maxRetry;
    private Instant nextRetryAt;
    private String processingOwner;
    private Instant processingAt;
    private Instant publishedAt;
    private String lastError;
    private Integer shardKey;
    @Version
    private Long version;
    private Instant createdAt;
    private Instant updatedAt;
    @PrePersist
    void prePersist() { Instant now = Instant.now(); createdAt = now; updatedAt = now; if (status == null) status = OutboxStatus.PENDING; if (retryCount == null) retryCount = 0; if (maxRetry == null) maxRetry = 16; if (nextRetryAt == null) nextRetryAt = now; if (headers == null) headers = "{}"; }
    @PreUpdate
    void preUpdate() { updatedAt = Instant.now(); }
}

Writing to the outbox

The OutboxWriter persists an OutboxEvent inside the same transaction that saves the business data, guaranteeing atomicity. Do not use a separate REQUIRES_NEW transaction because that would break the atomic guarantee and produce “dirty” messages.

@Service
@RequiredArgsConstructor
public class OutboxWriter {
    private final OutboxEventRepository outboxEventRepository;
    private final ObjectMapper objectMapper;
    public void append(String eventId, String aggregateType, String aggregateId,
                       String eventType, String topic, String partitionKey,
                       Object payload, Map<String, Object> headers, int shardKey) {
        try {
            OutboxEvent event = OutboxEvent.builder()
                .eventId(eventId)
                .aggregateType(aggregateType)
                .aggregateId(aggregateId)
                .eventType(eventType)
                .topic(topic)
                .partitionKey(partitionKey)
                .payload(objectMapper.writeValueAsString(payload))
                .headers(objectMapper.writeValueAsString(headers == null ? Map.of() : headers))
                .status(OutboxStatus.PENDING)
                .retryCount(0)
                .maxRetry(16)
                .nextRetryAt(Instant.now())
                .shardKey(shardKey)
                .build();
            outboxEventRepository.save(event);
        } catch (Exception e) {
            throw new IllegalStateException("Failed to persist outbox event", e);
        }
    }
}

Application service example (order creation)

@Service
@RequiredArgsConstructor
public class OrderApplicationService {
    private static final String ORDER_CREATED_TOPIC = "order.created";
    private static final int TOTAL_SHARDS = 8;
    private final OrderRepository orderRepository;
    private final InventoryRepository inventoryRepository;
    private final OutboxWriter outboxWriter;

    @Transactional
    public String createOrder(CreateOrderCommand cmd) {
        Inventory inventory = inventoryRepository.findByProductId(cmd.productId())
            .orElseThrow(() -> new IllegalArgumentException("Inventory not found"));
        inventory.decrease(cmd.quantity());
        String orderNo = "ORD-" + System.currentTimeMillis() + "-" + cmd.userId();
        Order order = Order.builder()
            .orderNo(orderNo)
            .userId(cmd.userId())
            .amount(cmd.amount())
            .status(OrderStatus.CREATED)
            .build();
        orderRepository.save(order);
        String eventId = UUID.randomUUID().toString().replace("-", "");
        OrderCreatedEvent event = OrderCreatedEvent.builder()
            .eventId(eventId)
            .orderNo(orderNo)
            .userId(cmd.userId())
            .productId(cmd.productId())
            .quantity(cmd.quantity())
            .amount(cmd.amount())
            .occurredAt(Instant.now())
            .schemaVersion(1)
            .build();
        int shardKey = Math.floorMod(orderNo.hashCode(), TOTAL_SHARDS);
        outboxWriter.append(eventId, "ORDER", orderNo, "OrderCreated",
            ORDER_CREATED_TOPIC, orderNo, event,
            Map.of("eventId", eventId, "aggregateId", orderNo, "aggregateType", "ORDER",
                   "eventType", "OrderCreated", "schemaVersion", 1), shardKey);
        return orderNo;
    }
}

Kafka publishing component

@Component
@RequiredArgsConstructor
public class KafkaOutboxPublisher {
    private final KafkaTemplate<String, String> kafkaTemplate;
    public RecordMetadata publish(OutboxEvent event) {
        CompletableFuture<SendResult<String, String>> future =
            kafkaTemplate.send(event.getTopic(), event.getPartitionKey(), event.getPayload());
        try {
            SendResult<String, String> result = future.get();
            return result.getRecordMetadata();
        } catch (Exception e) {
            throw new IllegalStateException("Kafka publish failed, outboxId=" + event.getId(), e);
        }
    }
}

Retry back‑off policy (exponential)

@Component
public class RetryBackoffPolicy {
    public Duration nextDelay(int retryCount) {
        long seconds = Math.min(300, Math.pow(2, Math.min(retryCount, 8)));
        return Duration.ofSeconds(seconds);
    }
}

Relay service – claim, send, update status

@Service
@RequiredArgsConstructor
public class OutboxRelayService {
    private final OutboxRelayRepository relayRepository;
    private final KafkaOutboxPublisher publisher;
    private final RetryBackoffPolicy retryBackoffPolicy;

    public int relay(String owner, int shardKey, int batchSize, int maxRetry) {
        List<OutboxEvent> events = relayRepository.claimBatch(owner, shardKey, batchSize);
        if (events.isEmpty()) return 0;
        for (OutboxEvent event : events) {
            try {
                publisher.publish(event);
                relayRepository.markPublished(event.getId());
            } catch (Exception ex) {
                int nextRetry = event.getRetryCount() + 1;
                if (nextRetry >= Math.min(event.getMaxRetry(), maxRetry)) {
                    relayRepository.markFailed(event.getId(), nextRetry, ex.getMessage());
                    // log permanently failed
                } else {
                    Duration delay = retryBackoffPolicy.nextDelay(nextRetry);
                    relayRepository.markRetry(event.getId(), nextRetry, delay, ex.getMessage());
                    // log retry scheduled
                }
            }
        }
        return events.size();
    }

    public int recover(Duration processingTimeout) {
        return relayRepository.recoverTimedOutProcessing(processingTimeout);
    }
}

Repository for claim and state updates (native SQL with SKIP LOCKED)

@Repository
@RequiredArgsConstructor
public class JpaOutboxRelayRepository implements OutboxRelayRepository {
    @PersistenceContext
    private final EntityManager em;

    @Transactional
    public List<OutboxEvent> claimBatch(String owner, int shardKey, int batchSize) {
        List<Long> ids = em.createNativeQuery(
            "SELECT id FROM outbox_event " +
            "WHERE status = 'PENDING' AND next_retry_at <= NOW() AND shard_key = :shardKey " +
            "ORDER BY created_at, id LIMIT :batchSize FOR UPDATE SKIP LOCKED"
        )
        .setParameter("shardKey", shardKey)
        .setParameter("batchSize", batchSize)
        .getResultList();
        if (ids.isEmpty()) return List.of();
        em.createNativeQuery(
            "UPDATE outbox_event SET status = 'PROCESSING', processing_owner = :owner, processing_at = NOW(), updated_at = NOW() " +
            "WHERE id = ANY(:ids)"
        )
        .setParameter("owner", owner)
        .setParameter("ids", ids.toArray(Long[]::new))
        .executeUpdate();
        return em.createQuery(
            "SELECT o FROM OutboxEvent o WHERE o.id IN :ids ORDER BY o.createdAt, o.id", OutboxEvent.class)
            .setParameter("ids", ids)
            .getResultList();
    }

    @Transactional
    public void markPublished(Long id) {
        em.createNativeQuery(
            "UPDATE outbox_event SET status = 'PUBLISHED', processing_owner = NULL, processing_at = NULL, " +
            "published_at = NOW(), updated_at = NOW(), last_error = NULL WHERE id = :id"
        ).setParameter("id", id).executeUpdate();
    }

    @Transactional
    public void markRetry(Long id, int retryCount, Duration delay, String errorMessage) {
        em.createNativeQuery(
            "UPDATE outbox_event SET status = 'PENDING', processing_owner = NULL, processing_at = NULL, " +
            "retry_count = :retryCount, next_retry_at = NOW() + CAST(:delaySeconds || ' seconds' AS interval), " +
            "last_error = :errorMessage, updated_at = NOW() WHERE id = :id"
        )
        .setParameter("id", id)
        .setParameter("retryCount", retryCount)
        .setParameter("delaySeconds", delay.toSeconds())
        .setParameter("errorMessage", truncate(errorMessage))
        .executeUpdate();
    }

    @Transactional
    public void markFailed(Long id, int retryCount, String errorMessage) {
        em.createNativeQuery(
            "UPDATE outbox_event SET status = 'FAILED', processing_owner = NULL, processing_at = NULL, " +
            "retry_count = :retryCount, last_error = :errorMessage, updated_at = NOW() WHERE id = :id"
        )
        .setParameter("id", id)
        .setParameter("retryCount", retryCount)
        .setParameter("errorMessage", truncate(errorMessage))
        .executeUpdate();
    }

    @Transactional
    public int recoverTimedOutProcessing(Duration timeout) {
        return em.createNativeQuery(
            "UPDATE outbox_event SET status = 'PENDING', processing_owner = NULL, processing_at = NULL, updated_at = NOW() " +
            "WHERE status = 'PROCESSING' AND processing_at < NOW() - CAST(:timeoutSeconds || ' seconds' AS interval)"
        )
        .setParameter("timeoutSeconds", timeout.toSeconds())
        .executeUpdate();
    }

    private String truncate(String text) {
        if (text == null) return null;
        return text.length() <= 2000 ? text : text.substring(0, 2000);
    }
}

Scheduler with shard workers

@Component
@RequiredArgsConstructor
public class OutboxRelayScheduler {
    private final OutboxRelayService relayService;
    private final OutboxRelayProperties props;
    private ScheduledExecutorService scheduler;
    private ExecutorService workers;
    private String owner;

    @PostConstruct
    public void start() throws Exception {
        if (!props.isEnabled()) return;
        owner = InetAddress.getLocalHost().getHostName() + "-" + UUID.randomUUID();
        scheduler = Executors.newScheduledThreadPool(1);
        workers = Executors.newFixedThreadPool(props.getConcurrency());
        for (Integer shard : props.getWorkerShards()) {
            scheduler.scheduleWithFixedDelay(() ->
                workers.submit(() -> runShard(shard)),
                1_000L, props.getPollInterval().toMillis(), TimeUnit.MILLISECONDS);
        }
        scheduler.scheduleWithFixedDelay(() ->
            workers.submit(this::recoverTimeoutRecords), 30_000L, 30_000L, TimeUnit.MILLISECONDS);
    }

    private void runShard(int shard) {
        try {
            int processed = relayService.relay(owner, shard, props.getBatchSize(), props.getMaxRetry());
            if (processed > 0) {
                // log processed count
            }
        } catch (Exception e) {
            // log worker failure
        }
    }

    private void recoverTimeoutRecords() {
        try {
            int recovered = relayService.recover(props.getProcessingTimeout());
            if (recovered > 0) {
                // log recovered count
            }
        } catch (Exception e) {
            // log recovery failure
        }
    }

    @PreDestroy
    public void shutdown() {
        scheduler.shutdown();
        workers.shutdown();
    }
}

Idempotent consumption guard (persistent)

@Service
@RequiredArgsConstructor
public class IdempotentConsumeGuard {
    private final ConsumedMessageRepository repo;
    @Transactional
    public boolean tryMarkProcessed(String consumerGroup, String eventId, String topic,
                                   int partition, long offset) {
        try {
            repo.save(ConsumedMessage.builder()
                .consumerGroup(consumerGroup)
                .eventId(eventId)
                .topic(topic)
                .partitionNo(partition)
                .offsetNo(offset)
                .processedAt(Instant.now())
                .build());
            return true;
        } catch (DataIntegrityViolationException ex) {
            return false; // already processed
        }
    }
}

Example consumer (loyalty points)

@Component
@RequiredArgsConstructor
@Slf4j
public class LoyaltyPointsConsumer {
    private static final String CONSUMER_GROUP = "loyalty-points-group";
    private final ObjectMapper mapper;
    private final IdempotentConsumeGuard guard;
    private final LoyaltyPointsService service;

    @KafkaListener(topics = "order.created", groupId = CONSUMER_GROUP)
    @Transactional
    public void onMessage(ConsumerRecord<String, String> record, Acknowledgment ack) throws Exception {
        String eventId = Optional.ofNullable(record.headers().lastHeader("eventId"))
            .map(h -> new String(h.value()))
            .orElseThrow(() -> new IllegalStateException("eventId header missing"));
        boolean first = guard.tryMarkProcessed(CONSUMER_GROUP, eventId, record.topic(),
            record.partition(), record.offset());
        if (!first) {
            log.info("Skip duplicate event {}", eventId);
            ack.acknowledge();
            return;
        }
        OrderCreatedEvent ev = mapper.readValue(record.value(), OrderCreatedEvent.class);
        service.grantPoints(ev.userId(), ev.amount());
        ack.acknowledge();
    }
}

REST endpoint to create an order

@RestController
@RequestMapping("/api/orders")
@RequiredArgsConstructor
public class OrderController {
    private final OrderApplicationService service;
    @PostMapping
    @ResponseStatus(HttpStatus.CREATED)
    public Map<String, Object> create(@Valid @RequestBody CreateOrderCommand cmd) {
        String orderNo = service.createOrder(cmd);
        return Map.of("success", true, "orderNo", orderNo);
    }
}

Monitoring and metrics (Prometheus compatible)

outbox_pending_count

– number of rows waiting to be sent. outbox_processing_count – rows currently claimed. outbox_failed_count – permanently failed rows. outbox_publish_success_total and outbox_publish_failure_total. outbox_publish_latency – time from row creation to successful Kafka ack. outbox_recover_total – number of timed‑out PROCESSING rows recovered.

Failure handling strategy

Transient – network glitch, broker timeout – retry with exponential back‑off.

Configuration – missing topic, ACL error – immediate alert, no retry.

Data – invalid payload, serialization error – move to FAILED state and trigger manual repair.

After the maximum retry count is exceeded the row moves to FAILED. An operational UI can list failed rows, allow payload edits, and trigger a manual re‑publish.

Scaling considerations

Shard the outbox table (e.g., shard_key = hash(aggregate_id) % N) and let each worker process a distinct shard to avoid lock contention.

Adjust batchSize (100‑1000) and pollInterval (100 ms‑1 s) based on DB capacity.

Configure Kafka producer settings (compression, linger, batch size) to match throughput.

Separate the Relay into its own service when write load grows, allowing independent horizontal scaling.

High‑concurrency bottlenecks and mitigations

Database scanning – use the composite index (status, next_retry_at, created_at, id) and FOR UPDATE SKIP LOCKED to keep lock time short.

Outbox table growth – periodic archiving of PUBLISHED rows (e.g., older than 7 days) to an archive table or external storage.

Kafka broker saturation – enable compression (zstd), tune linger.ms and batch.size, monitor ISR health.

When to switch to CDC

If the outbox write volume reaches hundreds of thousands of rows per second, or sub‑millisecond latency is required, consider moving the polling logic to Debezium/Kafka Connect. This removes the polling load from the application but adds platform complexity.

Common pitfalls

Assuming Kafka transactions replace the Outbox – they do not provide cross‑system atomicity.

Relying only on an in‑memory cache for idempotence – persistent storage is required for true guarantees.

Believing the Outbox eliminates duplicates – consumers must still be idempotent.

Polling too aggressively – adds unnecessary DB load.

Infinite retries for configuration or data errors – they never succeed and should surface as alerts.

Integration testing with Testcontainers

@Testcontainers
@SpringBootTest
public class OutboxIntegrationTest {
    @Container static PostgreSQLContainer<?> postgres = new PostgreSQLContainer<>("postgres:15");
    @Container static KafkaContainer kafka = new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.6.1"));
    @DynamicPropertySource
    static void configure(DynamicPropertyRegistry r) {
        r.add("spring.datasource.url", postgres::getJdbcUrl);
        r.add("spring.datasource.username", postgres::getUsername);
        r.add("spring.datasource.password", postgres::getPassword);
        r.add("spring.kafka.bootstrap-servers", kafka::getBootstrapServers);
    }
    @Autowired OrderApplicationService service;
    @Autowired OutboxEventRepository repo;
    @Test
    void shouldPersistOrderAndOutboxAtomically() {
        String orderNo = service.createOrder(new CreateOrderCommand(10001L, 20001L, 1, new BigDecimal("199.00")));
        assertThat(orderNo).isNotBlank();
        assertThat(repo.count()).isEqualTo(1);
    }
}

Load‑testing recommendations

Measure API QPS and order‑creation latency.

Measure outbox write TPS.

Measure Relay send rate and end‑to‑end latency (order creation → Kafka ack).

Monitor DB CPU, connection pool saturation, slow queries.

Monitor Kafka broker write latency and ISR health.

Production deployment checklist

Business data and outbox are persisted in the same @Transactional method.

Relay uses FOR UPDATE SKIP LOCKED and respects PROCESSING timeout recovery.

Consumers implement the idempotent guard (unique constraint on consumer_group + event_id).

Failure handling path (retry, back‑off, FAILED state, manual UI) is in place.

Monitoring alerts for pending backlog, failure count growth, and latency spikes.

Archiving job removes old PUBLISHED rows.

Chaos tests for Kafka outage and application crash verify that messages are not lost.

Conclusion

The Outbox pattern bridges the gap between relational transactions and asynchronous event delivery. By combining a well‑designed table schema, a lock‑free polling Relay, exponential back‑off, idempotent consumers, comprehensive metrics, and operational tooling, Spring Boot 3 applications can achieve reliable, recoverable, and observable event publishing at production scale.