Outbox pattern con Kafka: no pierdas eventos al escribir

1. Visión general

Un servicio típicamente debe hacer dos cosas seguidas: guardar datos en la base de datos y enviar un evento a Kafka para que otros servicios lo lean. Estos dos sistemas no comparten transacción. Si un paso tiene éxito pero el otro falla, el sistema perderá coherencia. El resultado es que el pedido se guardó pero nadie recibió el evento OrderPlaced.

Este artículo explica el problema del dual write e implementa el Outbox pattern con Java 21, Spring Boot 3.x y Spring Kafka. Después de leerlo, tendrás un message relay que empuja eventos de la tabla outbox a Kafka, junto con pruebas e idempotencia del consumer.

2. El problema dual write

Mira la siguiente forma “ingenua” de escribir:

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

Este código tiene al menos dos escenarios de fallo:

• Kafka envía con éxito, transaction revierte. Por ejemplo, error al hacer commit o violación de restricción. El evento “fantasma” ya salió para un pedido que no existe.
• Transaction se confirma, Kafka falla al enviar. El broker podría no ser accesible o la aplicación podría crashear justo después del commit. El pedido está en la base de datos pero el evento desapareció.

Cambiar el orden de los dos comandos tampoco resuelve el problema de raíz. Dos operaciones en dos sistemas diferentes no pueden ser atómicas sin una transacción distribuida. La mayoría de los equipos tampoco quieren usar XA/2PC porque es complejo y lento.

3. La idea del Outbox pattern

El Outbox pattern convierte “escribir en BD + enviar a Kafka” en “escribir en BD + escribir en BD”:

1. En la misma transacción, el servicio guarda la Order e inserta una fila en la tabla outbox_event.
2. Un proceso separado llamado message relay lee las filas sin enviar en la tabla outbox, las envía a Kafka y luego las marca como enviadas.

Como el paso 1 solo toca una base de datos, es absolutamente atómico: si hay pedido, definitivamente hay evento. El paso 2 se puede reintentar cuantas veces sea necesario. A cambio, recibes la semántica at-least-once: un evento puede enviarse duplicado, pero nunca se pierde.

4. Implementación con Spring Boot

4.1. Dependencias

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

A continuación, la configuración del productor 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 obliga al broker a confirmar después de que todas las réplicas hayan escrito. enable.idempotence=true ayuda al productor a no crear duplicados cuando reintenta en la capa de red.

4.2. Tabla outbox y entidad

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;

El índice parcial solo contiene filas sin enviar. Gracias a eso, la consulta del relay siempre es rápida, incluso cuando la tabla ya tiene millones de filas antiguas.

@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 también juega el rol de event id único. El consumer lo usará para filtrar duplicados (sección 6). aggregateId será la clave del mensaje Kafka.

4.3. Escribir Order y evento en la misma transacción

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("No se puede serializar el evento", e);
        }
    }
}

El punto crucial está en @Transactional. Ambas órdenes save se confirman o revierten juntas. OrderService ya no conoce nada sobre Kafka, así que no hay llamadas de red dentro de la transacción.

4.4. Message relay que empuja eventos a Kafka

Primero, escribe un repositorio con una consulta de bloqueo:

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 ejecutar múltiples instancias de la aplicación al mismo tiempo. Cada instancia bloquea su propio lote e ignora las filas que otras instancias están procesando. Gracias a eso, dos instancias no envían el mismo evento en paralelo.

A continuación está el 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("Envío del evento {} falló, se reintentará", event.getId(), e);
                break; // detente para mantener el orden, el siguiente ciclo lo intenta de nuevo
            }
        }
    }

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

Cada parte, en orden:

• @Scheduled(fixedDelay = 1000) ejecuta el relay un segundo después de que termina la ejecución anterior. Recuerda añadir @EnableScheduling a una clase de configuración.
• .get(5, TimeUnit.SECONDS) espera la confirmación del broker. El relay solo marca publishedAt cuando está seguro de que Kafka recibió el mensaje.
• markPublished cambia la entidad siendo gestionada. JPA genera automáticamente la sentencia UPDATE cuando la transacción se confirma, no necesitas llamar a save.
• Cuando hay error, relay hace break en lugar de saltar esa fila. Los eventos del mismo pedido por lo tanto no cambian de orden.
• La clave es aggregateId, así que todos los eventos de un Order caen en la misma partición y mantienen el orden en Kafka.

El log cuando el broker se detiene se verá así:

WARN  OutboxRelay : Envío del evento 3f6c1e0a-... falló, se reintentará
org.apache.kafka.common.errors.TimeoutException: Topic orders not present in metadata after 5000 ms.

Cuando el broker vuelve a la vida, el siguiente ciclo continuará desde exactamente ese evento.

Nota: Si la aplicación se cuelga después de que Kafka recibió el mensaje pero antes de que la transacción se confirme, la fila outbox sigue siendo “sin enviar”. El siguiente ciclo lo enviará de nuevo. Esto es exactamente at-least-once, y el consumer debe ser capaz de manejar duplicados.

5. Pruebas

La primera prueba verifica que cada pedido genera exactamente una fila en 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();
    }
}

La consulta nativa del repositorio no se valida cuando se inicia. Por lo tanto, H2 todavía puede ejecutar esta prueba aunque no soporte completamente SKIP LOCKED. Para probar el relay con consultas reales, deberías usar PostgreSQL a través de Testcontainers.

La segunda prueba garantiza que relay no marque como enviado cuando Kafka falla:

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

Estas dos pruebas aseguran el comportamiento central: solo marcar como enviado cuando el broker confirma.

6. Del lado del consumer: manejo idempotente

Como puede haber duplicados, el consumer debe recordar qué eventId ya procesó. Este registro debe estar en la misma transacción que el negocio:

@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; // ya procesado, ignora el duplicado
        }
        inventoryService.reserveStock(payload);
        processedEvents.save(new ProcessedEvent(eventId));
    }
}

La tabla processed_event debe tener event_id como clave primaria. Si dos threads procesan el mismo duplicado, la restricción bloqueará uno de ellos.

7. Errores comunes y cuándo no usar esto

• Llamar Kafka dentro de OrderService “para ir rápido”. Eso es volver al dual write. Todo evento debe pasar por la tabla outbox.
• Olvidar limpiar la tabla outbox. La tabla crecerá muy rápido. Debes tener un job que elimine filas con published_at más antiguas que unos días.
• Saltar la fila con error y continuar. Esto arruina el orden de los eventos. Si un payload está corrupto para siempre, cámbialo a estado FAILED para investigación, no reintentar infinitamente.
• Polling demasiado frecuente. Ciclos de 100 ms con muchas instancias crean carga significativa en la BD. Ciclos de 0,5–2 segundos suelen ser suficientes.
• Necesitas latencia muy baja o throughput muy grande. En ese caso, reemplaza polling con CDC (Change Data Capture). Por ejemplo, Debezium lee WAL/binlog y envía directamente las filas outbox a Kafka. La tabla outbox permanece igual, solo el relay cambia.

Nota: No uses el Outbox pattern si el evento no es crítico, por ejemplo, métricas o logs de comportamiento del usuario que pueden perder algo. En ese caso, enviar directamente a Kafka es mucho más simple. Por el contrario, Kafka transactions (transactional.id) no reemplaza al Outbox: solo garantizan atomicidad entre topics de Kafka, no incluyen tu base de datos.

8. Conclusión

El dual write entre la base de datos y Kafka siempre tiene brechas que pierden o generan eventos extra. El Outbox pattern cierra esa brecha en tres pasos. Primero, escribe el evento en la tabla outbox en la misma transacción que los datos del negocio. Segundo, usa un relay separado que empuje eventos a Kafka y solo marque como enviado cuando el broker confirma. Tercero, maneja idempotencia en el consumer para absorber duplicados de la semántica at-least-once.

Ya tienes todas las piezas: schema con índice parcial, OrderService sin dependencia de Kafka, relay usando FOR UPDATE SKIP LOCKED para correr múltiples instancias, pruebas del comportamiento central y consumer a prueba de duplicados. Cuando el sistema crezca más, puedes reemplazar el relay polling con CDC sin modificar el código del negocio.