Event Sourcing with Kafka: Save Events Instead of State

1. Overview

Most applications only save the current state. When an order is cancelled, the row in the orders table changes to CANCELLED. No one remembers what items the order had before and whether it was paid.

Event sourcing does the opposite: we only save the sequence of events that occurred, and the state is a result calculated from that sequence. In this article, you’ll use Kafka as an event log for the Order aggregate with Spring Boot 3.x. You’ll rebuild state by replaying events, build a read model using projections, and understand the practical limitations of this approach.

2. Saving State vs. Saving Events

With the traditional approach, an order is a single record:

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

With event sourcing, the same order is a sequence of events that are only appended, never modified or deleted:

OrderCreated   (A1, customer=C7)
ItemAdded      (A1, sku=COFFEE, price=150000)
ItemAdded      (A1, sku=MUG,    price=100000)
OrderCancelled (A1, reason="Customer changed mind")

The state CANCELLED, total=250000 is calculated by “folding” (or reducing) the events in order. This way of storing brings three benefits:

Natural audit trail: the complete history is the source data itself.
New read models without migration: to add a report, you replay the log into a new projection.
Debug across time: you can rebuild the state at any point in time.

Kafka is well-suited for this log role. A topic is an append-only, ordered sequence within each partition. If you use orderId as the key, all events for the same order will land in the same partition and maintain the correct order.

3. Project Setup

3.1. Dependencies

<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 manages the version of spring-kafka, so you don’t need to declare <version>. spring-boot-starter-json brings Jackson and the jsr310 module for serializing Instant.

3.2. Configuration

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

Key points to note:

acks: all and enable.idempotence: true ensure events are durably written and producer retries don’t create duplicate records.
We disable Spring’s type headers. The event type will be in the JSON itself via a type attribute, so consumers written in other languages can still read it.
auto-offset-reset: earliest allows a new consumer group to read the entire log from the beginning. This is essential for replay.

3.3. Topic Preserves Events Forever

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

By default, Kafka deletes data after 7 days. With an event store, that means losing the source data. retention.ms=-1 disables time-based deletion.

Note: Don’t use cleanup.policy=compact for event topics. Compaction only keeps the last record for each key, meaning it deletes the entire history of an order. Compaction is only appropriate for snapshot topics.

4. Define Events

Events are immutable data that describe what has happened and are named in the past tense. Java 21’s sealed interface combined with record is perfect for this:

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

When you send an ItemAdded, the payload on Kafka will look like:

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

The names in @JsonSubTypes are a long-term contract. You can rename the Java class, but you must never rename the string "ItemAdded", otherwise old events won’t deserialize.

5. Aggregate: Build State from Events

Order has two kinds of methods:

Command (addItem, pay, cancel): check business rules then return an event. Commands don’t modify state themselves.
apply: receives an event and updates state. This method must not throw business exceptions, because the event has already happened.

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, "Can only add items to a new order");
        return new OrderEvent.ItemAdded(id, sku, price, Instant.now());
    }

    public OrderEvent pay() {
        requireStatus(Status.NEW, "Order is not awaiting payment");
        if (total.signum() == 0) {
            throw new IllegalStateException("Cannot pay for an empty order");
        }
        return new OrderEvent.Paid(id, total, Instant.now());
    }

    public OrderEvent cancel(String reason) {
        if (status == Status.CANCELLED) {
            throw new IllegalStateException("Order is already cancelled");
        }
        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; }
}

Since OrderEvent is sealed, the compiler fully checks the switch in apply. If you add a new event type and forget to handle it, compilation fails. This is a valuable safety net.

5.1. Testing the Aggregate

The aggregate doesn’t depend on Kafka, so tests run very fast:

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", "Customer changed mind", 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", "Out of stock", now)));

        assertThatThrownBy(order::pay)
                .isInstanceOf(IllegalStateException.class)
                .hasMessageContaining("awaiting payment");
    }
}

The first test shows that four events are folded into a consistent state, with version equal to the number of events applied. The second test shows that business rules are validated based on state rebuilt from history.

6. Write Events to Kafka

6.1. Event Store

To handle commands, you need the history of one order. Kafka doesn’t support queries like “give me all records with key A1”, so we keep a copy in memory. This copy is loaded by reading the entire topic every time the application starts:

@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("Failed to write event " + 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);
    }
}

Here’s what each part does:

send(TOPIC, event.orderId(), event) uses orderId as the key. This ensures all events for one order land in the same partition and maintain the correct order.
.get(...) waits for broker confirmation. A command only succeeds once the event is in the log.
The listener uses a random groupId. Each startup is a new group, which combined with earliest means the entire history is read back into the map.

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("Order not found: " + orderId);
        }
        return Order.replay(history);
    }
}

The command handling flow is always the same: load history → replay → call command → append event. There are no UPDATE statements at all.

Note: The in-memory map is updated asynchronously via the listener. If you call addItem right after createOrder, the map might not have processed Created yet. Also, Kafka has no optimistic concurrency like “only write if the current version is 4”. Two instances handling the same order can write conflicting events. In production, you should route all commands for an orderId to a single consumer (command topic keyed by orderId), or use Kafka Streams with a state store. Another option is to use a dedicated event store or database as the write source, then publish to Kafka.

7. Projections: Build a Read Model

The event log is excellent for writing, but inconvenient for queries like “list all paid orders by customer C7”. Projections solve this. A projection is a consumer that reads events and maintains a model optimized for reading. This is the “Q” in 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();
    }
}

This example uses ConcurrentHashMap for brevity. In practice, you’d write to PostgreSQL, Elasticsearch, or Redis. The important point is that the projection has a fixed groupId, so Kafka remembers the offset and only processes new events after each restart.

A projection must be idempotent. Kafka guarantees at-least-once delivery, so an event can be delivered again after rebalancing. With a database, you can store the latest processed offset per partition and skip older records.

7.1. Replay to Rebuild the Read Model

Suppose the projection has a bug that miscalculates total. You fix the code, clear the read model data, stop the application, and reset the offset for the group to the beginning:

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

On startup, the projection reads the entire history and creates the correct read model. To add a completely new report, just write a new listener with a new groupId. That report automatically has all data from day one. This is the biggest advantage of saving events instead of state.

8. Common Mistakes and When Not to Use This

Changing event schema. Old events exist forever. Only add new fields with default values, and never delete or change the meaning of old fields. For major changes, create a new event type (ItemAddedV2) or write an “upcaster” to convert old events on read. For large systems, use Schema Registry with Avro or Protobuf.

Putting state into events. An event like OrderUpdated { entire order } is just CRUD in event sourcing clothing. Events describe business intent, like ItemAdded or OrderCancelled.

Replay taking too long. An aggregate with thousands of events loads slowly. The solution is snapshots: periodically save state with version to a compacted topic, then on load read the snapshot and only replay events after it.

Calling side effects in apply. If apply sends an email, each replay sends it again. apply should only change in-memory state.

Personal data and the right to deletion. An immutable log conflicts with requests to delete user data. A common technique is crypto-shredding: encrypt personal data with a key specific to each customer, then destroy the key to “delete.”

You should not use event sourcing when:

Your domain is mainly simple CRUD with no one needing history.
Your team isn’t comfortable with eventual consistency between writes and read models.
You need consistent queries across multiple aggregates right after a write.

Note: Kafka is an excellent event log, but not a complete event store. It lacks two things: fast reads per aggregate and conditional writes based on version. Many systems use a database or dedicated event store as the source of truth, then use Kafka (often via the outbox pattern) to distribute events to projections and other services.

9. Conclusion

In this article, you saved orders as an immutable sequence of events on Kafka instead of a single record being overwritten. You modeled events with sealed interface and record, rebuilt state with Order.replay, and validated business rules before appending. You also built a read model with projections and rebuilt it by resetting the offset.

Three points to remember:

Use orderId as the key to maintain event order for each order.
Disable time-based retention and don’t use compaction for event topics.
Understand Kafka’s concurrency limitations before treating it as the single source of truth.

When applied correctly, event sourcing gives you a complete history and the ability to create new read models without migration.