ES
FrançaisPortuguêsEspañolDeutschTiếng ViệtEnglish한국어日本語简体中文
Inicio / Arquitectura y diseño / Kafka

Event sourcing con Kafka: guardar eventos en lugar de estado

Cris The Dev · 12/10/2026 · 11 min de lectura · 158

1. Descripción general

La mayoría de las aplicaciones solo guardan el estado actual. Cuando se cancela un pedido, la fila en la tabla orders cambia a CANCELLED. Nadie sabe qué artículos tenía antes el pedido ni si ya se había pagado.

El event sourcing hace lo opuesto: guardamos solo la secuencia de eventos que ocurrieron, y el estado es el resultado de computar esa secuencia. En este artículo, usarás Kafka como event log para el aggregate Order con Spring Boot 3.x. Reconstruirás el estado por medio de replay de eventos, construirás read models con projection y comprenderás las limitaciones prácticas de este enfoque.

2. Guardar estado vs. guardar eventos

Con el enfoque tradicional, un pedido es un registro:

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

Con event sourcing, ese mismo pedido es una secuencia de eventos que solo se añaden, nunca se modifican ni se eliminan:

OrderCreated   (A1, customer=C7)
ItemAdded      (A1, sku=CAFE, price=150000)
ItemAdded      (A1, sku=TAZA,    price=100000)
OrderCancelled (A1, reason="El cliente cambió de opinión")

El estado CANCELLED, total=250000 se calcula ‘plegando’ (fold) los eventos en orden. Esta forma de almacenar aporta tres beneficios:

  • Auditoría natural: el historial completo es el dato original.
  • Los nuevos read models no necesitan migración: si quieres añadir un reporte, haces replay del log en una nueva projection.
  • Depuración en el tiempo: puedes reconstruir el estado en cualquier punto en el tiempo.

Kafka es ideal para este papel de log. Un topic es una secuencia append-only con orden dentro de cada partición. Si usas orderId como clave, todos los eventos de un mismo pedido caerán en la misma partición y mantendrán el orden correcto.

3. Configuración del proyecto

3.1. Dependencia

<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 gestiona la versión de spring-kafka, así que no necesitas declarar <version>. spring-boot-starter-json incluye Jackson y el módulo jsr310 para serializar Instant.

3.2. Configuración

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

Puntos destacados:

  • acks: all y enable.idempotence: true garantizan que los eventos se escriben de forma durabilística y que los reintentos del producer no crean registros duplicados.
  • Desactivamos los type headers de Spring. El tipo de evento estará directamente en el JSON a través de la propiedad type, así que consumers escritos en otros lenguajes también pueden leerlo.
  • auto-offset-reset: earliest permite que un nuevo consumer group lea desde el principio del log. Este es un requisito para replay.

3.3. Topic que mantiene eventos eternamente

@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 defecto, Kafka elimina datos después de 7 días. Con un event store, eso equivale a perder el dato original. retention.ms=-1 desactiva la eliminación por tiempo.

Nota: No uses cleanup.policy=compact para el topic de eventos. La compactación solo mantiene el último registro de cada clave, es decir, elimina todo el historial del pedido. La compactación solo es apropiada para topics de snapshot.

4. Definición de eventos

Los eventos son datos inmutables, describen lo que ya ocurrió y se nombran en pasado. sealed interface combinada con record de Java 21 es perfecta para esto:

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

Cuando envías un ItemAdded, el payload en Kafka tendrá este aspecto:

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

El nombre en @JsonSubTypes es un contrato a largo plazo. Puedes cambiar el nombre de la clase Java, pero no la cadena "ItemAdded", de lo contrario los eventos antiguos no se podrán leer de nuevo.

5. Aggregate: reconstruir el estado a partir de eventos

Order tiene dos tipos de métodos:

  • Command (addItem, pay, cancel): verifica reglas de negocio y devuelve un evento. El comando no modifica el estado por sí mismo.
  • apply: recibe un evento y actualiza el estado. Esta función no debe lanzar errores de negocio, porque el evento ya ocurrió.
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, "Solo añade artículos a pedidos nuevos");
        return new OrderEvent.ItemAdded(id, sku, price, Instant.now());
    }

    public OrderEvent pay() {
        requireStatus(Status.NEW, "El pedido no está en estado de espera de pago");
        if (total.signum() == 0) {
            throw new IllegalStateException("No se puede pagar un pedido vacío");
        }
        return new OrderEvent.Paid(id, total, Instant.now());
    }

    public OrderEvent cancel(String reason) {
        if (status == Status.CANCELLED) {
            throw new IllegalStateException("El pedido ya fue 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 es sealed, el switch en apply es verificado completamente por el compilador. Si añades un nuevo tipo de evento y olvidas manejarlo, el código no compilará. Esta es una red de seguridad valiosa.

5.1. Prueba del aggregate

El aggregate no depende de Kafka, así que las pruebas son muy rápidas:

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", "CAFE", new BigDecimal("150000"), now),
                new OrderEvent.ItemAdded("A1", "TAZA", new BigDecimal("100000"), now),
                new OrderEvent.Cancelled("A1", "El cliente cambió de opinión", now)));

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

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

        assertThatThrownBy(order::pay)
                .isInstanceOf(IllegalStateException.class)
                .hasMessageContaining("espera de pago");
    }
}

La primera prueba muestra que cuatro eventos se pliegan en un estado consistente, con version igual al número de eventos aplicados. La segunda prueba muestra que las reglas de negocio se verifican basándose en el estado reconstruido del historial.

6. Escribir eventos en Kafka

6.1. Event store

Para procesar un comando, necesitas el historial de un pedido. Kafka no soporta consultas como ‘dame todos los registros con clave A1’, así que mantenemos una copia en memoria. Esta copia se carga leyendo todo el topic cada vez que la aplicación se 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("No se pudo escribir el 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);
    }
}

Explicación de cada parte:

  • send(TOPIC, event.orderId(), event) usa orderId como clave. Gracias a esto, todos los eventos de un pedido se encuentran en la misma partición y mantienen el orden correcto.
  • .get(...) espera la confirmación del broker. Un comando solo se considera exitoso cuando el evento ya está en el log.
  • El listener usa un groupId aleatorio. Cada inicio es un grupo nuevo, que combinado con earliest hace que todo el historial se lea de nuevo en el 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 no encontrado " + orderId);
        }
        return Order.replay(history);
    }
}

El flujo de procesamiento siempre es igual: cargar historial → hacer replay → llamar comando → append evento. No hay ninguna sentencia UPDATE.

Nota: El mapa en memoria se actualiza asincronía a través del listener. Si llamas a addItem justo después de createOrder, es posible que el mapa aún no haya recibido Created. Además, Kafka no tiene concurrencia optimista del tipo ‘escribir solo si la versión actual es 4’. Dos instancias procesando el mismo pedido pueden escribir dos eventos contradictorios. En producción, deberías encaminar todos los comandos de un orderId a un único consumer (topic de comandos con clave según orderId), o usar Kafka Streams con state store. Otra opción es usar un event store dedicado o una base de datos como fuente de escritura principal, luego publicar en Kafka.

7. Projection: construir read models

El event log es excelente para escribir, pero no es conveniente para consultas como ‘listar pedidos pagados del cliente C7’. La projection resuelve esto. Es un consumer que lee eventos y mantiene un modelo optimizado para lectura. Esta es la parte ‘Q’ en 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 ejemplo usa ConcurrentHashMap por simplicidad. En la práctica, escribirías en PostgreSQL, Elasticsearch o Redis. Lo importante es que la projection tiene un groupId fijo, así que Kafka recuerda el offset y solo procesa eventos nuevos después de cada reinicio.

La projection debe ser idempotente. Kafka garantiza at-least-once, así que un evento puede ser entregado de nuevo después de un rebalance. Con una base de datos, puedes guardar el offset procesado más recientemente por partición e ignorar registros más antiguos.

7.1. Replay para reconstruir el read model

Supongamos que la projection tiene un bug que hace que total se calcule incorrectamente. Arreglas el código, eliminas los datos del read model, paras la aplicación y reseteas el offset del grupo al principio:

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

Cuando reinicies, la projection lee todo el historial y crea el read model correcto. Para añadir un reporte completamente nuevo, solo necesitas escribir un nuevo listener con un groupId nuevo. Ese reporte automáticamente tiene todos los datos desde el primer día. Este es el mayor beneficio de guardar eventos en lugar de estado.

8. Errores comunes y cuándo no usar event sourcing

Cambiar el schema de eventos. Los eventos antiguos existen para siempre. Solo añade campos nuevos con valores por defecto, y no elimines ni cambies el significado de campos antiguos. Cuando necesites cambios importantes, crea un nuevo tipo de evento (ItemAddedV2) o escribe un ‘upcaster’ que convierta eventos antiguos al leerlos. Para sistemas grandes, usa Schema Registry con Avro o Protobuf.

Meter estado en los eventos. Un evento tipo OrderUpdated { todo el pedido } es solo CRUD disfrazado de event sourcing. Los eventos deben describir la intención de negocio, como ItemAdded u OrderCancelled.

Replay muy lento. Un aggregate con miles de eventos cargará lentamente. La solución es snapshot: periódicamente guarda el estado junto con version en un topic compactado, cuando cargas lees el snapshot y solo replays los eventos posteriores.

Llamar side effects en apply. Si apply envía un email, cada replay volverá a enviar el email. apply solo debe cambiar el estado en memoria.

Datos personales y derecho al olvido. El log inmutable entra en conflicto con los requisitos de eliminación de datos de usuario. La técnica comúnmente usada es crypto-shredding: encripta datos personales con una clave única para cada cliente, cuando necesites eliminar destruye la clave.

No deberías usar event sourcing cuando:

  • El dominio es principalmente CRUD simple y nadie necesita historial.
  • El equipo no está familiarizado con la eventual consistency entre el lado de escritura y los read models.
  • Necesitas consultas consistentes inmediatamente después de escribir en múltiples aggregates.

Nota: Kafka es un excelente event log, pero no es un event store completo. Le faltan dos cosas: lectura rápida por aggregate y escritura condicional según versión. Muchos sistemas usan una base de datos o un event store dedicado como fuente de verdad, luego usan Kafka (normalmente vía outbox pattern) para distribuir eventos a otras projections y servicios.

9. Conclusión

En este artículo, has guardado pedidos como una secuencia inmutable de eventos en Kafka en lugar de un registro que se sobrescribe. Has modelado eventos con sealed interface y record, reconstruido el estado con Order.replay y verificado reglas de negocio antes de append. También has construido read models con projection y los has reconstruido reseteando el offset.

Hay tres puntos para recordar:

  • Usa orderId como clave para mantener el orden de eventos de cada pedido.
  • Desactiva la retención por tiempo y no uses compaction para el topic de eventos.
  • Ten clara conciencia de las limitaciones de concurrencia de Kafka antes de considerarlo como la única fuente de verdad.

Cuando se aplica en el lugar correcto, event sourcing te proporciona historial completo y la capacidad de crear nuevos read models sin migración.

kafkaevent-sourcingspring-bootcqrsspring-kafka
App del autorHoanTien.proCashback on Shopee, TikTok Shop and Lazada (Vietnam)iOS · Android · Web

Más sobre este tema