Outbox pattern avec Kafka : ne jamais perdre d'événements lors de l'écriture en base

1. Vue d’ensemble

Un service doit généralement accomplir deux tâches consécutives : sauvegarder les données en base de données et publier un événement sur Kafka pour les autres services. Ces deux systèmes ne partagent pas une transaction commune. Si une étape réussit et l’autre échoue, le système perd sa cohérence. Résultat : la commande est sauvegardée mais personne ne reçoit l’événement OrderPlaced.

Cet article explique le problème de la dual write et implémente le Outbox pattern avec Java 21, Spring Boot 3.x et Spring Kafka. À la fin, vous disposerez d’un relais de messages qui pousse les événements de la table outbox vers Kafka, avec tests et consumer idempotent inclus.

2. Le problème de la dual write

Regardons l’approche “naïve” suivante :

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

Ce code présente au moins deux scénarios d’erreur :

• Kafka envoie avec succès, la transaction se termine en rollback. Par exemple une erreur lors du commit ou une violation de contrainte. L’événement “fantôme” est parti pour une commande qui n’existe pas.
• Transaction commit, l’envoi vers Kafka échoue. Le broker peut être inaccessible, ou l’application crash juste après le commit. La commande est en base mais l’événement disparaît.

Inverser l’ordre des deux commandes ne résout pas le problème fondamental. Deux opérations sur deux systèmes différents ne peuvent pas être atomiques sans transaction distribuée. La plupart des équipes évitent aussi XA/2PC car c’est complexe et lent.

3. L’idée du Outbox pattern

Le Outbox pattern transforme “écrire en base + envoyer à Kafka” en “écrire en base + écrire en base” :

1. Dans la même transaction, le service sauvegarde Order et insère une ligne dans la table outbox_event.
2. Un processus séparé appelé message relay lit les lignes non envoyées de la table outbox, les envoie à Kafka puis les marque comme envoyées.

Puisque l’étape 1 ne touche qu’une seule base de données, elle est absolument atomique : s’il y a une commande, il y a forcément un événement. L’étape 2 peut être retentée autant de fois qu’on veut. En contrepartie, vous obtenez la sémantique at-least-once : un événement peut être envoyé deux fois, jamais perdu.

4. Implémentation avec Spring Boot

4.1. Dépendances

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

Ensuite, la configuration d’un producer sécurisé :

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 force le broker à confirmer après que tous les replicas aient écrit. enable.idempotence=true empêche le producer de créer des doublons lors des retries au niveau réseau.

4.2. Table outbox et 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;

L’index partiel ne contient que les lignes non envoyées. Ainsi la requête du relay reste toujours rapide, même quand la table contient des millions de lignes anciennes.

@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 joue aussi le rôle d’event id unique. Le consumer l’utilisera pour filtrer les doublons (section 6). aggregateId sera la clé du message Kafka.

4.3. Écrire Order et événement dans la même 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("Impossible de sérialiser l'événement", e);
        }
    }
}

Le point crucial se trouve dans @Transactional. Les deux appels save sont validés ensemble ou annulés ensemble. OrderService n’a plus aucune connaissance de Kafka, donc aucun appel réseau ne se trouve dans la transaction.

4.4. Message relay pour pousser les événements vers Kafka

D’abord, écrivez un repository avec une requête verrouillante :

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+) permet de lancer plusieurs instances de l’application simultanément. Chaque instance verrouille son propre lot et ignore les lignes que les autres instances traitent. Cela empêche deux instances d’envoyer le même événement en parallèle.

Ensuite vient le 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("L'envoi de l'événement {} a échoué, nouvelle tentative au prochain passage", event.getId(), e);
                break; // arrêter pour préserver l'ordre, réessayer au prochain cycle
            }
        }
    }

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

Détaillons chaque partie :

• @Scheduled(fixedDelay = 1000) lance le relay une seconde après la fin du précédent cycle. N’oubliez pas @EnableScheduling sur une classe de configuration.
• .get(5, TimeUnit.SECONDS) attend la confirmation du broker. Le relay ne marque publishedAt que quand il est certain que Kafka a reçu le message.
• markPublished modifie l’entity gérée. JPA génère automatiquement une requête UPDATE lors du commit, vous n’avez pas besoin d’appeler save.
• En cas d’erreur, le relay break au lieu de passer au suivant. Les événements d’une même Order ne seront donc jamais dés-ordonnés.
• La clé est aggregateId, donc tous les événements d’une même Order tombent dans la même partition et conservent leur ordre sur Kafka.

Voici un exemple de log quand le broker est temporairement hors ligne :

WARN  OutboxRelay : L'envoi de l'événement 3f6c1e0a-... a échoué, nouvelle tentative au prochain passage
org.apache.kafka.common.errors.TimeoutException: Topic orders not present in metadata after 5000 ms.

Quand le broker revient, le prochain cycle du relay reprendra exactement depuis cet événement.

Note importante : Si l’application crash après que Kafka a reçu le message mais avant le commit de la transaction, la ligne outbox reste marquée comme “non envoyée”. Le cycle suivant renverra le message. C’est précisément la sémantique at-least-once, et le consumer doit absolument gérer les doublons.

5. Tests

Le premier test vérifie que chaque commande génère exactement une ligne 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();
    }
}

Les requêtes natives du repository ne sont pas validées au démarrage. C’est pourquoi H2 peut quand même exécuter ce test même s’il ne supporte pas complètement SKIP LOCKED. Pour tester le relay avec la vraie requête, préférez PostgreSQL via Testcontainers.

Le deuxième test garantit que le relay ne marque pas comme envoyé quand Kafka échoue :

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

Ces deux tests cristallisent le comportement essentiel : ne marquer comme envoyé que si le broker a confirmé.

6. Côté consumer : gestion idempotente

Puisque les doublons sont possibles, le consumer doit mémoriser les eventId déjà traités. Cette mémorisation doit se faire dans la même transaction que la logique métier :

@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; // déjà traité, ignorer le doublon
        }
        inventoryService.reserveStock(payload);
        processedEvents.save(new ProcessedEvent(eventId));
    }
}

La table processed_event doit avoir event_id comme clé primaire. Si deux threads traitent le même doublon, la contrainte en bloquera un.

7. Pièges courants et quand ne pas l’utiliser

• Appeler Kafka directement dans OrderService “pour aller plus vite”. Vous retombez dans la dual write. Chaque événement doit passer par la table outbox.
• Oublier de nettoyer la table outbox. La table gonflera très rapidement. Mettez en place un job pour supprimer les lignes avec published_at plus vieux que quelques jours.
• Ignorer une ligne en erreur et continuer. Cela désordonne les événements. Si un payload est corrompu définitivement, basculez-le en statut FAILED pour investigation, n’essayez pas indéfiniment.
• Sonder trop fréquemment. Un cycle de 100 ms avec plusieurs instances crée une charge significative sur la base. Un cycle de 0,5 à 2 secondes suffit généralement.
• Besoin de très faible latence ou très gros débit. Remplacez le polling par du CDC (Change Data Capture). Debezium par exemple lit le WAL/binlog et pousse directement les lignes outbox vers Kafka. La table outbox reste inchangée, seul le relay change.

Note : N’utilisez le Outbox pattern que si les événements sont importants, par exemple des commandes confirmées. Pour les métriques ou logs de comportement où on peut perdre quelques lignes, envoyer directement à Kafka est bien plus simple. Inversement, les transactions Kafka (transactional.id) ne remplacent pas le Outbox : elles n’assurent l’atomicité qu’entre topics Kafka, pas avec votre base de données.

8. Conclusion

La dual write entre la base de données et Kafka laisse toujours des failles où des événements se perdent ou se dupliquent. Le Outbox pattern colmate ces failles en trois étapes. Premièrement, écrivez l’événement dans la table outbox au sein de la même transaction que les données métier. Deuxièmement, utilisez un relay séparé qui pousse les événements vers Kafka et ne marque “envoyé” que si le broker a confirmé. Troisièmement, traitez les événements de manière idempotente côté consumer pour absorber les doublons de la sémantique at-least-once.

Vous disposez maintenant de toutes les pièces : un schéma avec index partiel, une OrderService indépendante de Kafka, un relay utilisant FOR UPDATE SKIP LOCKED pour fonctionner sur plusieurs instances, des tests pour le comportement clé, et un consumer anti-doublon. Quand votre système grandira, vous pourrez remplacer le polling du relay par du CDC sans modifier le code métier.