Outbox pattern com Kafka: não perca eventos ao gravar

1. Visão geral

Um serviço geralmente precisa fazer duas coisas consecutivas: salvar dados no banco de dados e emitir eventos para o Kafka para outros serviços. Esses dois sistemas não compartilham uma transação. Se um passo tem sucesso e o outro falha, o sistema perderá consistência. O resultado é que o pedido foi salvo, mas ninguém recebeu o evento OrderPlaced.

Este artigo explica o problema da escrita dupla (dual write) e implementa o Outbox pattern com Java 21, Spring Boot 3.x e Spring Kafka. Ao terminar, você terá um message relay que envia eventos da tabela outbox para o Kafka, junto com testes e um consumer idempotente.

2. O problema do dual write

Veja como uma abordagem “ingênua” funciona:

@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;
}

Esse código tem pelo menos dois cenários de falha:

• Kafka envia com sucesso, transaction reverte. Por exemplo, erro ao fazer commit ou violação de constraint. Um evento “fantasma” foi enviado para um pedido que não existe.
• Transaction é confirmada, Kafka falha ao enviar. O broker pode estar inacessível, ou a aplicação cai logo após o commit. O pedido está no banco de dados, mas o evento desaparece.

Trocar a ordem dos dois comandos também não resolve o problema fundamental. Duas operações em dois sistemas diferentes não podem ser atômicas sem uma transação distribuída. A maioria das equipes também não quer usar XA/2PC porque é complexo e lento.

3. A ideia do Outbox pattern

O Outbox pattern transforma “escrever database + enviar Kafka” em “escrever database + escrever database”:

1. Na mesma transação, o serviço salva Order e insere uma linha na tabela outbox_event.
2. Um processo separado chamado message relay lê as linhas não enviadas na tabela outbox, envia-as para o Kafka e depois marca como enviadas.

Como o passo 1 toca apenas um banco de dados, é absolutamente atômico: se há um pedido, há definitivamente um evento. O passo 2 pode ser retentado quantas vezes forem necessárias. Em troca, você obtém a semântica at-least-once: um evento pode ser enviado duplicadamente, nunca será perdido.

4. Implementação com Spring Boot

4.1. Dependências

<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>

Em seguida, vem a configuração do producer seguro:

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 força o broker a confirmar depois que todas as réplicas sincronizadas gravaram. enable.idempotence=true ajuda o producer a não gerar duplicatas ao tentar novamente na camada de rede.

4.2. Tabela outbox e entidade

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;

O índice parcial contém apenas as linhas não enviadas. Graças a isso, a consulta do relay é sempre rápida, mesmo quando a tabela já tem milhões de linhas antigas.

@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 também funciona como um event id único. O consumer vai usá-lo para filtrar duplicatas (seção 6). aggregateId será a chave da mensagem Kafka.

4.3. Gravar Order e evento na mesma transação

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("Não foi possível serializar o evento", e);
        }
    }
}

O ponto crucial está em @Transactional. Os dois comandos save são confirmados ou revertidos juntos. OrderService não sabe mais nada sobre Kafka, então não há chamadas de rede dentro da transação.

4.4. Message relay envia eventos para o Kafka

Primeiro, escreva um repositório com uma consulta que bloqueia linhas:

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+) permite executar múltiplas instâncias simultaneamente. Cada instância bloqueia seu próprio lote e ignora as linhas que outras instâncias estão processando. Assim, duas instâncias não enviam o mesmo evento em paralelo.

A seguir, o 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("Falha ao enviar evento {}, será retentado", event.getId(), e);
                break;
            }
        }
    }

    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;
    }
}

Analisando cada parte em sequência:

• @Scheduled(fixedDelay = 1000) executa o relay um segundo após a execução anterior terminar. Lembre-se de adicionar @EnableScheduling em uma classe de configuração.
• .get(5, TimeUnit.SECONDS) aguarda a confirmação do broker. O relay marca publishedAt apenas quando tem certeza de que o Kafka recebeu a mensagem.
• markPublished modifica a entidade sob gerenciamento. JPA gera automaticamente o comando UPDATE quando a transação é confirmada, você não precisa chamar save.
• Em caso de erro, o relay faz break em vez de pular a linha. Os eventos do mesmo pedido não serão reordenados.
• A chave é aggregateId, então todos os eventos de uma Order caem na mesma partição e mantêm a ordem no Kafka.

O log quando o broker é temporariamente interrompido será assim:

WARN  OutboxRelay : Falha ao enviar evento 3f6c1e0a-... será retentado
org.apache.kafka.common.errors.TimeoutException: Topic orders not present in metadata after 5000 ms.

Quando o broker volta a funcionar, a próxima execução continuará a partir do evento correto.

Nota: Se a aplicação falhar depois que o Kafka recebeu a mensagem mas antes da transação ser confirmada, a linha outbox ainda estará como “não enviada”. A próxima execução reenviará. Essa é exatamente a semântica at-least-once, e o consumer obrigatoriamente deve lidar com duplicatas.

5. Testes

O primeiro teste verifica que cada pedido gera exatamente uma linha outbox:

@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();
    }
}

A consulta nativa do repositório não é validada na inicialização. Por isso, H2 ainda consegue executar esse teste mesmo sem suporte completo a SKIP LOCKED. Para testar o relay com consultas reais, você deve usar PostgreSQL via Testcontainers.

O segundo teste garante que o relay não marca como enviado quando o Kafka falha:

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();
    }
}

Esses dois testes fixam o comportamento central: marca como enviado apenas quando o broker confirma.

6. No lado do consumer: processamento idempotente

Como pode receber duplicatas, o consumer precisa lembrar quais eventId já foram processados. Essa memorização deve estar na mesma transação que a lógica de negócio:

@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; // já processado, ignorar duplicata
        }
        inventoryService.reserveStock(payload);
        processedEvents.save(new ProcessedEvent(eventId));
    }
}

A tabela processed_event deve ter event_id como chave primária. Se duas threads processarem a mesma duplicata, a constraint impedirá uma delas.

7. Erros comuns e quando não usar

• Chamar Kafka em OrderService “por conveniência”. Fazer isso é voltar ao dual write. Todos os eventos devem passar pela tabela outbox.
• Esquecer de limpar a tabela outbox. A tabela crescerá muito rapidamente. Tenha um job que delete as linhas com published_at mais antigo que alguns dias.
• Pular a linha com erro e continuar. Isso prejudica a ordem dos eventos. Se um payload está permanentemente corrompido, coloque-o em um estado FAILED para investigação, não faça retry infinito.
• Polling muito frequente. Um período de 100 ms com múltiplas instâncias cria carga considerável no banco de dados. Um período de 0,5–2 segundos geralmente é suficiente.
• Precisar de latência muito baixa ou throughput muito alto. Nesse caso, substitua polling por CDC (Change Data Capture). Por exemplo, Debezium lê WAL/binlog e envia as linhas outbox diretamente para o Kafka. A tabela outbox permanece a mesma, apenas o relay muda.

Nota: Não use Outbox pattern se os eventos não forem importantes, por exemplo, métricas ou logs de comportamento do usuário que podem ser perdidos. Nesse caso, enviar diretamente para o Kafka é muito mais simples. Por outro lado, Kafka transactions (transactional.id) não substituem o Outbox: elas apenas garantem atomicidade entre tópicos Kafka, não incluem seu banco de dados.

8. Conclusão

Dual write entre banco de dados e Kafka sempre tem brechas que causam perda ou duplicação de eventos. O Outbox pattern fecha essas brechas em três passos. Primeiro, escreva eventos na tabela outbox na mesma transação que os dados de negócio. Segundo, use um relay separado que envia eventos para o Kafka e marca como enviado apenas quando o broker confirma. Terceiro, processe idempotência no consumer para absorver as duplicatas da semântica at-least-once.

Você tem todas as peças: o schema com índice parcial, OrderService que não depende de Kafka, relay usando FOR UPDATE SKIP LOCKED para rodas múltiplas instâncias, testes para o comportamento central e consumer à prova de duplicatas. Quando o sistema cresce, você pode substituir o relay de polling por CDC sem precisar modificar o código de negócio.