I’ve been thinking a lot about how we build systems that not only handle transactions but also remember every change that ever happened. That’s why I want to share my approach to implementing event sourcing with Spring Boot and Apache Kafka. It’s a pattern that has transformed how I think about data consistency and system reliability.

Event sourcing fundamentally changes how we store application state. Instead of keeping only the current state, we persist every state change as an immutable event. This creates a complete history of all changes, providing audit capabilities and enabling powerful features like temporal queries and system replay.

What if you could rebuild your entire application state from scratch just by replaying events?

Let me show you how to set up the foundation. First, we need to define our domain events. These are the building blocks of our event-sourced system.

public abstract class DomainEvent {
    private final String eventId;
    private final String aggregateId;
    private final LocalDateTime occurredOn;
    private final long version;

    protected DomainEvent(String aggregateId, long version) {
        this.eventId = UUID.randomUUID().toString();
        this.aggregateId = aggregateId;
        this.occurredOn = LocalDateTime.now();
        this.version = version;
    }
}

The aggregate root is where the magic happens. It’s responsible for applying events and maintaining consistency.

public abstract class AggregateRoot {
    protected String id;
    protected long version;
    private final List<DomainEvent> uncommittedEvents = new ArrayList<>();
    
    protected void applyEvent(DomainEvent event, boolean isNew) {
        applyChange(event);
        if (isNew) {
            uncommittedEvents.add(event);
        }
        this.version = event.getVersion();
    }
}

Now, let’s look at a concrete example. Here’s how we might implement order events:

public class OrderCreatedEvent extends DomainEvent {
    private final String customerId;
    private final LocalDateTime createdAt;
    
    public OrderCreatedEvent(String aggregateId, long version, String customerId) {
        super(aggregateId, version);
        this.customerId = customerId;
        this.createdAt = LocalDateTime.now();
    }
}

Integrating with Kafka gives us reliable event streaming. The configuration is straightforward but powerful.

@Configuration
public class KafkaConfig {
    
    @Bean
    public ProducerFactory<String, DomainEvent> producerFactory() {
        Map<String, Object> config = new HashMap<>();
        config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return new DefaultKafkaProducerFactory<>(config);
    }
}

Have you ever considered what happens when you need to rebuild a projection from events?

The event store is where we persist our events. Here’s a simple repository implementation:

@Repository
public interface EventStoreRepository extends JpaRepository<EventEntity, Long> {
    List<EventEntity> findByAggregateIdOrderByVersionAsc(String aggregateId);
}

Testing is crucial in event-sourced systems. We need to verify that our aggregates behave correctly when events are applied.

@Test
void should_create_order_when_applying_created_event() {
    Order order = new Order();
    OrderCreatedEvent event = new OrderCreatedEvent("order-123", 1, "customer-456");
    
    order.applyEvent(event, false);
    
    assertEquals("order-123", order.getId());
    assertEquals("customer-456", order.getCustomerId());
}

What challenges might you face when implementing event versioning and schema evolution?

Error handling requires careful consideration. We need to ensure events are processed exactly once and in order.

@KafkaListener(topics = "orders-events")
public void handleEvent(ConsumerRecord<String, DomainEvent> record) {
    try {
        DomainEvent event = record.value();
        eventProcessor.process(event);
    } catch (Exception e) {
        log.error("Failed to process event: {}", record.key(), e);
        // Implement retry or dead letter queue logic
    }
}

The beauty of this approach is how it enables CQRS. We can build multiple read models optimized for different queries without affecting the write side.

Snapshotting helps optimize performance for aggregates with long event histories.

@Entity
@Table(name = "order_snapshots")
public class OrderSnapshot {
    @Id
    private String aggregateId;
    private long version;
    private String customerId;
    private String status;
    private LocalDateTime snapshotDate;
}

Implementing event sourcing requires shifting our mindset from state-based to event-based thinking. But the benefits in terms of auditability, reproducibility, and system resilience are well worth the effort.

I’d love to hear about your experiences with event sourcing. What challenges have you faced? What successes have you had? Share your thoughts in the comments below, and if you found this useful, please like and share with others who might benefit from this approach.