Event sourcing com Kafka: guardar eventos em vez de estado

1. Visão geral

A maioria das aplicações apenas guarda o estado atual. Quando um pedido é cancelado, a linha na tabela orders muda para CANCELLED. Ninguém mais sabe o que tinha no pedido antes e se foi pago ou não.

Event sourcing faz o contrário: você guarda apenas a sequência de eventos que aconteceram, e o estado é o resultado de calcular essa sequência. Neste artigo, você vai usar Kafka como event log para o aggregate Order com Spring Boot 3.x. Você vai reconstruir o estado através de replay de eventos, construir read model com projection e entender as limitações práticas dessa abordagem.

2. Guardar estado vs guardar eventos

Com a abordagem tradicional, um pedido é um registro:

orders: id=A1 | status=CANCELLED | total=250000

Com event sourcing, o mesmo pedido é uma sequência de eventos que apenas cresce, nunca é modificada ou deletada:

OrderCreated   (A1, customer=C7)
ItemAdded      (A1, sku=COFFEE, price=150000)
ItemAdded      (A1, sku=MUG,    price=100000)
OrderCancelled (A1, reason="Cliente mudou de ideia")

O estado CANCELLED, total=250000 é calculado ao “redobrar” (fold) os eventos em ordem. Guardar dessa forma traz três vantagens:

• Auditoria natural: o histórico completo é o próprio dado bruto.
• Novo read model sem migration: quer adicionar um relatório, você faz replay do log em uma nova projection.
• Debug no tempo: você pode reconstruir o estado em qualquer momento.

Kafka é adequado para esse papel de log. Um topic é uma sequência append-only ordenada em cada partition. Se usar orderId como key, todos os eventos de um mesmo pedido caem na mesma partition e mantêm a ordem correta.

3. Configuração do projeto

3.1. Dependências

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-json</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

Spring Boot gerencia a versão do spring-kafka, então você não precisa declarar <version>. spring-boot-starter-json traz Jackson e o módulo jsr310 para serializar Instant.

3.2. Configuração

spring:
  kafka:
    bootstrap-servers: localhost:9092
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.springframework.kafka.support.serializer.JsonSerializer
      acks: all
      properties:
        enable.idempotence: true
        spring.json.add.type.headers: false
    consumer:
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializer
      properties:
        spring.json.value.default.type: com.nonla.orders.OrderEvent
        spring.json.trusted.packages: com.nonla.orders
        spring.json.use.type.headers: false

Pontos a notar:

• acks: all e enable.idempotence: true garantem que o evento foi escrito duramente e retry do producer não cria registros duplicados.
• Desligamos o type header do Spring. O tipo de evento fica no próprio JSON via atributo type, então um consumer em outra linguagem ainda consegue ler.
• auto-offset-reset: earliest permite que um novo consumer group leia desde o início do log. Isso é necessário para replay.

3.3. Topic que guarda eventos para sempre

@Configuration
public class KafkaTopicConfig {

    @Bean
    NewTopic orderEvents() {
        return TopicBuilder.name("order-events")
                .partitions(3)
                .replicas(1)
                .config(TopicConfig.RETENTION_MS_CONFIG, "-1")
                .config(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE)
                .build();
    }
}

Por padrão, Kafka deleta dados após 7 dias. Com event store, isso significa perder o dado bruto. retention.ms=-1 desativa deleção por tempo.

Aviso: Não use cleanup.policy=compact para o topic de eventos. Compaction mantém apenas o último registro de cada key, ou seja, deleta todo o histórico do pedido. Compaction é só para topic de snapshot.

4. Definindo os eventos

Um evento é um dado imutável que descreve algo que já aconteceu e é nomeado no tempo passado. sealed interface combinado com record do Java 21 é perfeito para isso:

package com.nonla.orders;

@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
@JsonSubTypes({
        @JsonSubTypes.Type(value = OrderEvent.Created.class, name = "OrderCreated"),
        @JsonSubTypes.Type(value = OrderEvent.ItemAdded.class, name = "ItemAdded"),
        @JsonSubTypes.Type(value = OrderEvent.Paid.class, name = "OrderPaid"),
        @JsonSubTypes.Type(value = OrderEvent.Cancelled.class, name = "OrderCancelled")
})
public sealed interface OrderEvent {

    String orderId();
    Instant occurredAt();

    record Created(String orderId, String customerId, Instant occurredAt) implements OrderEvent {}
    record ItemAdded(String orderId, String sku, BigDecimal price, Instant occurredAt) implements OrderEvent {}
    record Paid(String orderId, BigDecimal amount, Instant occurredAt) implements OrderEvent {}
    record Cancelled(String orderId, String reason, Instant occurredAt) implements OrderEvent {}
}

Ao enviar um ItemAdded, o payload no Kafka terá esta forma:

{"type":"ItemAdded","orderId":"A1","sku":"COFFEE","price":150000,"occurredAt":"2026-10-11T03:15:00Z"}

O nome em @JsonSubTypes é um contrato de longa duração. Você pode mudar o nome da classe Java, mas não pode mudar a string "ItemAdded", senão eventos antigos não conseguirão ser lidos novamente.

5. Aggregate: construindo estado a partir de eventos

Order tem dois tipos de método:

• Command (addItem, pay, cancel): verificam regras de negócio e retornam um evento. Commands não alteram estado por conta própria.
• apply: recebe um evento e atualiza o estado. Esta função não deve lançar exceções de negócio, já que o evento já aconteceu.

public final class Order {

    public enum Status { NEW, PAID, CANCELLED }

    private String id;
    private String customerId;
    private Status status;
    private BigDecimal total = BigDecimal.ZERO;
    private final List<String> items = new ArrayList<>();
    private long version;

    public static Order replay(List<OrderEvent> history) {
        Order order = new Order();
        history.forEach(order::apply);
        return order;
    }

    public void apply(OrderEvent event) {
        switch (event) {
            case OrderEvent.Created e -> { id = e.orderId(); customerId = e.customerId(); status = Status.NEW; }
            case OrderEvent.ItemAdded e -> { items.add(e.sku()); total = total.add(e.price()); }
            case OrderEvent.Paid e -> status = Status.PAID;
            case OrderEvent.Cancelled e -> status = Status.CANCELLED;
        }
        version++;
    }

    public OrderEvent addItem(String sku, BigDecimal price) {
        requireStatus(Status.NEW, "Só adicione itens a um pedido novo");
        return new OrderEvent.ItemAdded(id, sku, price, Instant.now());
    }

    public OrderEvent pay() {
        requireStatus(Status.NEW, "O pedido não está em estado de aguardando pagamento");
        if (total.signum() == 0) {
            throw new IllegalStateException("Não é possível pagar um pedido vazio");
        }
        return new OrderEvent.Paid(id, total, Instant.now());
    }

    public OrderEvent cancel(String reason) {
        if (status == Status.CANCELLED) {
            throw new IllegalStateException("O pedido já foi cancelado");
        }
        return new OrderEvent.Cancelled(id, reason, Instant.now());
    }

    private void requireStatus(Status expected, String message) {
        if (status != expected) throw new IllegalStateException(message);
    }

    public String id() { return id; }
    public Status status() { return status; }
    public BigDecimal total() { return total; }
    public List<String> items() { return List.copyOf(items); }
    public long version() { return version; }
}

Como OrderEvent é sealed, o switch dentro de apply é verificado completamente pelo compilador. Quando você adiciona um novo tipo de evento mas esquece de tratá-lo, o código não compila. Essa é uma rede de segurança valiosa.

5.1. Testando o aggregate

Aggregate não depende de Kafka, então testes são muito rápidos:

class OrderTest {

    private final Instant now = Instant.parse("2026-10-11T03:00:00Z");

    @Test
    void replayRebuildsStateFromHistory() {
        Order order = Order.replay(List.of(
                new OrderEvent.Created("A1", "C7", now),
                new OrderEvent.ItemAdded("A1", "COFFEE", new BigDecimal("150000"), now),
                new OrderEvent.ItemAdded("A1", "MUG", new BigDecimal("100000"), now),
                new OrderEvent.Cancelled("A1", "Cliente mudou de ideia", now)));

        assertThat(order.status()).isEqualTo(Order.Status.CANCELLED);
        assertThat(order.total()).isEqualByComparingTo("250000");
        assertThat(order.items()).containsExactly("COFFEE", "MUG");
        assertThat(order.version()).isEqualTo(4);
    }

    @Test
    void cannotPayCancelledOrder() {
        Order order = Order.replay(List.of(
                new OrderEvent.Created("A1", "C7", now),
                new OrderEvent.Cancelled("A1", "Sem estoque", now)));

        assertThatThrownBy(order::pay)
                .isInstanceOf(IllegalStateException.class)
                .hasMessageContaining("aguardando pagamento");
    }
}

O primeiro teste mostra quatro eventos sendo redobrados em um estado consistente, com version igual ao número de eventos aplicados. O segundo teste mostra a regra de negócio sendo verificada com base no estado reconstruído do histórico.

6. Escrevendo eventos no Kafka

6.1. Event store

Para processar um command, você precisa do histórico de um pedido. Kafka não suporta query “me dê todos os registros com key A1”, então mantemos uma cópia em memória. Essa cópia é carregada lendo o topic inteiro cada vez que a aplicação inicia:

@Component
public class OrderEventStore {

    private static final String TOPIC = "order-events";

    private final KafkaTemplate<String, OrderEvent> kafkaTemplate;
    private final Map<String, List<OrderEvent>> streams = new ConcurrentHashMap<>();

    public OrderEventStore(KafkaTemplate<String, OrderEvent> kafkaTemplate) {
        this.kafkaTemplate = kafkaTemplate;
    }

    public void append(OrderEvent event) {
        try {
            kafkaTemplate.send(TOPIC, event.orderId(), event).get(10, TimeUnit.SECONDS);
        } catch (Exception e) {
            throw new IllegalStateException("Não foi possível escrever o evento " + event, e);
        }
    }

    public List<OrderEvent> load(String orderId) {
        return List.copyOf(streams.getOrDefault(orderId, List.of()));
    }

    @KafkaListener(topics = TOPIC, groupId = "order-store-#{T(java.util.UUID).randomUUID()}")
    void on(OrderEvent event) {
        streams.computeIfAbsent(event.orderId(), id -> new CopyOnWriteArrayList<>()).add(event);
    }
}

Explicação de cada parte:

• send(TOPIC, event.orderId(), event) usa orderId como key. Por isso todos os eventos de um pedido ficam na mesma partition e mantêm a ordem correta.
• .get(...) espera a confirmação do broker. Um command só é considerado bem-sucedido quando o evento está no log.
• O listener usa groupId aleatório. A cada inicialização é um novo group, combinado com earliest então todo o histórico é lido de volta para o mapa.

6.2. Command service

@Service
public class OrderCommandService {

    private final OrderEventStore store;

    public OrderCommandService(OrderEventStore store) {
        this.store = store;
    }

    public String createOrder(String customerId) {
        String orderId = UUID.randomUUID().toString();
        store.append(new OrderEvent.Created(orderId, customerId, Instant.now()));
        return orderId;
    }

    public void addItem(String orderId, String sku, BigDecimal price) {
        Order order = loadExisting(orderId);
        store.append(order.addItem(sku, price));
    }

    public void pay(String orderId) {
        store.append(loadExisting(orderId).pay());
    }

    private Order loadExisting(String orderId) {
        List<OrderEvent> history = store.load(orderId);
        if (history.isEmpty()) {
            throw new NoSuchElementException("Pedido não encontrado: " + orderId);
        }
        return Order.replay(history);
    }
}

O fluxo de processamento é sempre o mesmo: carrega histórico → faz replay → chama command → adiciona evento. Nenhum comando UPDATE de verdade.

Aviso: O mapa em memória é atualizado de forma assíncrona através do listener. Se você chamar addItem logo após createOrder, é possível que o mapa ainda não tenha recebido Created. Além disso, Kafka não tem concorrência otimista tipo “escreva só se a versão atual for 4”. Duas instâncias processando o mesmo pedido podem escrever dois eventos conflitantes. Em produção, você deveria rotear todo command de um orderId para um consumer único (command topic com key por orderId), ou usar Kafka Streams com state store. Outra opção é usar um event store dedicado ou database como fonte primária de escrita, e depois publicar para Kafka.

7. Projection: construindo read model

O log de eventos é ótimo para escrita, mas não é conveniente para queries tipo “liste pedidos pagos do cliente C7”. Projection resolve isso. É um consumer que lê eventos e mantém um modelo otimizado para leitura. Essa é a parte “Q” no CQRS.

public record OrderSummary(String orderId, String customerId, String status, BigDecimal total) {}

@Component
public class OrderSummaryProjection {

    private final Map<String, OrderSummary> summaries = new ConcurrentHashMap<>();

    @KafkaListener(topics = "order-events", groupId = "order-summary-projection")
    void on(OrderEvent event) {
        summaries.compute(event.orderId(), (id, current) -> switch (event) {
            case OrderEvent.Created e -> new OrderSummary(id, e.customerId(), "NEW", BigDecimal.ZERO);
            case OrderEvent.ItemAdded e -> new OrderSummary(id, current.customerId(), current.status(),
                    current.total().add(e.price()));
            case OrderEvent.Paid e -> new OrderSummary(id, current.customerId(), "PAID", current.total());
            case OrderEvent.Cancelled e -> new OrderSummary(id, current.customerId(), "CANCELLED", current.total());
        });
    }

    public List<OrderSummary> paidOrdersOf(String customerId) {
        return summaries.values().stream()
                .filter(s -> s.customerId().equals(customerId) && s.status().equals("PAID"))
                .toList();
    }
}

Este exemplo usa ConcurrentHashMap para simplificar. Na prática, você escreveria em PostgreSQL, Elasticsearch ou Redis. O ponto importante é que a projection tem groupId fixo, então Kafka lembra do offset e processa apenas eventos novos após cada restart.

Projection deve ser idempotente. Kafka garante at-least-once, então um evento pode ser entregue novamente após rebalance. Com database, você pode guardar junto o offset processado por último em cada partition e ignorar registros mais antigos.

7.1. Replay para reconstruir read model

Suponha que projection tem um bug que calcula total errado. Você corrige o código, limpa o read model, para a aplicação e reseta o offset do group para o início:

kafka-consumer-groups.sh --bootstrap-server localhost:9092 \
  --group order-summary-projection \
  --topic order-events \
  --reset-offsets --to-earliest --execute

GROUP                     TOPIC         PARTITION  NEW-OFFSET
order-summary-projection  order-events  0          0
order-summary-projection  order-events  1          0
order-summary-projection  order-events  2          0

Ao reiniciar, a projection lê todo o histórico e cria o read model correto. Quer adicionar um relatório completamente novo, basta escrever um novo listener com groupId novo. Esse relatório automaticamente tem todos os dados desde o primeiro dia. Essa é a maior vantagem de guardar eventos em vez de estado.

8. Erros comuns e quando não usar

Mudar schema do evento. Eventos antigos existem para sempre. Adicione apenas novos campos com valores padrão, e nunca delete ou mude o significado de campos antigos. Quando precisa de grandes mudanças, crie um novo tipo de evento (ItemAddedV2) ou escreva um “upcaster” para converter eventos antigos ao ler. Em sistemas maiores, use Schema Registry com Avro ou Protobuf.

Colocar estado dentro do evento. Um evento tipo OrderUpdated { pedido todo } é só CRUD disfarçado de event sourcing. Eventos descrevem intenção de negócio, como ItemAdded ou OrderCancelled.

Replay muito lento. Um aggregate com milhares de eventos leva muito tempo para carregar. A solução é snapshot: periodicamente guarde o estado junto com version em um topic compacted, ao carregar leia o snapshot e só faça replay dos eventos posteriores.

Chamar side effect dentro de apply. Se apply enviar email, cada replay envia email novamente. apply só pode alterar estado em memória.

Dados pessoais e direito ao esquecimento. Log imutável entra em conflito com pedidos de deletar dados do usuário. A técnica comum é crypto-shredding: criptografe dados pessoais com uma chave privada por cliente, ao deletar, destrua a chave.

Você não deveria usar event sourcing quando:

• O domain é principalmente CRUD simples e ninguém precisa do histórico.
• Seu time não está familiarizado com eventual consistency entre lado de escrita e read model.
• Você precisa de queries consistentes imediatamente após escrita em múltiplos aggregates.

Aviso: Kafka é um event log excelente, mas não é um event store completo. Faltam duas coisas: ler rápido por aggregate individual e escrita com condição baseada em version. Muitos sistemas usam database ou event store dedicado como fonte primária de verdade, depois distribuem eventos para Kafka (frequentemente via outbox pattern) para projeções e outros serviços.

9. Conclusão

Neste artigo, você guardou pedidos como uma sequência imutável de eventos no Kafka em vez de um registro sobrescrito. Você modelou eventos com sealed interface e record, reconstruiu estado com Order.replay e verificou regras de negócio antes de adicionar. Você também construiu read model com projection e reconstruiu via reset de offset.

Três pontos para lembrar:

• Use orderId como key para manter a ordem dos eventos de cada pedido.
• Desative retention por tempo e não use compaction para o topic de eventos.
• Reconheça as limitações de Kafka quanto a concorrência antes de tratá-lo como única fonte de verdade.

Quando aplicado no lugar certo, event sourcing oferece histórico completo e capacidade de criar novos read models sem migration.