Mastering Event Sourcing & CQRS with Spring Boot & Apache Kafka

Introduction

In the realm of modern application development, traditional CRUD (Create, Read, Update, Delete) architectures often fall short when faced with demands for high auditability, scalability, complex business processes, and the ability to reconstruct historical states. As systems grow in complexity and distributed nature, developers seek patterns that offer greater flexibility, resilience, and performance.

This is where Event Sourcing (ES) and Command Query Responsibility Segregation (CQRS) shine. These architectural patterns, often used together, provide a powerful foundation for building robust, scalable, and auditable microservices. When combined with Spring Boot for rapid application development and Apache Kafka for a high-throughput, fault-tolerant event bus, you get an incredibly potent stack.

This comprehensive guide will deep dive into understanding, implementing, and leveraging Event Sourcing and CQRS with Spring Boot and Apache Kafka, providing practical examples and best practices to empower your next-generation applications.

Prerequisites

To get the most out of this guide, a basic understanding of the following is recommended:

Java 17+
Spring Boot (basics of controllers, services, repositories)
Maven or Gradle
Apache Kafka (core concepts like topics, producers, consumers)
Docker (for running Kafka and PostgreSQL locally)
Domain-Driven Design (DDD) concepts (Aggregates, Entities, Value Objects)

Understanding Event Sourcing

Event Sourcing is an architectural pattern where the state of an application is stored as a sequence of immutable events, rather than just the current state. Instead of updating a record in a database, you record an event that describes what happened. For example, instead of updating an Account balance, you record an AccountCreditedEvent or AccountDebitedEvent.

How Event Sourcing Works:

Events: Represent facts that have occurred in the past. They are immutable and carry all necessary information about the change.
Aggregate Root: A cluster of domain objects that can be treated as a single unit. It processes commands, applies business rules, and emits events.
Event Store: A specialized database that durably stores the sequence of events. It's the single source of truth for the application's state.

Why Event Sourcing?

Full Audit Trail: Every change is recorded as an event, providing a complete, immutable history.
Time Travel Debugging: Reconstruct the state of your application at any point in time.
Easier Evolution: New features can often be built by replaying existing events and projecting them into new read models.
Improved Debugging: Understand exactly how an aggregate reached its current state.
Foundation for CQRS: The event stream naturally feeds into separate read models.

Understanding CQRS (Command Query Responsibility Segregation)

CQRS is an architectural pattern that separates the concerns of reading data (queries) from writing data (commands). In traditional systems, a single data model and API handle both, leading to complexities when optimizing for different access patterns.

How CQRS Works:

Commands: Represent an intent to change the state of the system (e.g., DepositMoneyCommand). They are imperative and should only modify the system.
Queries: Represent a request for data (e.g., GetAccountBalanceQuery). They are declarative and should not modify the system state.
Command Model (Write Model): Optimized for writes. Handles commands, applies business logic, and emits events. Often built on an Event Store in an ES/CQRS setup.
Query Model (Read Model): Optimized for reads. A denormalized, often simpler data structure built by consuming events from the write model. It can be tailored for specific query needs.

Why CQRS?

Independent Scalability: Read and write models can be scaled and optimized independently.
Performance: Read models can be denormalized and stored in purpose-built databases (e.g., NoSQL for fast reads) without impacting write performance.
Simpler Models: Each model can be simpler and more focused on its specific responsibility.
Flexibility: Allows for different data storage technologies for reads and writes.
Separation of Concerns: Clear distinction between state-changing operations and data retrieval.

The Power of ES + CQRS Together

Event Sourcing and CQRS are highly complementary. Event Sourcing provides the immutable, auditable source of truth (the write model), while CQRS leverages this event stream to build highly optimized, denormalized read models.

When a command is processed, it generates events, which are then stored in the event store. These same events are then published to an event bus (like Apache Kafka). Event consumers (projectors or denormalizers) listen to these events, transform them, and update the read models, which are then queried by the application's read-side services.

This combination offers unparalleled flexibility: you can create multiple read models from the same event stream, each optimized for different query patterns or user interfaces, without affecting the core business logic or the integrity of the write model.

Introducing Apache Kafka as the Event Bus

Apache Kafka is a distributed streaming platform that provides high-throughput, fault-tolerant, and durable messaging. It's an ideal choice for the event bus in an ES/CQRS architecture due to its key features:

Durability: Events are persisted on disk, ensuring no data loss.
High Throughput: Capable of handling millions of events per second.
Scalability: Easily scales horizontally to accommodate growing event volumes.
Distributed Nature: Designed for distributed systems, offering high availability.
Replayability: Consumers can re-read events from any point in time, crucial for rebuilding read models or debugging.
Ordered Delivery: Events within a partition are delivered in order, which is vital for maintaining event sequence.

In our setup, Kafka will serve as the conduit for events flowing from the write model to the read model. When an aggregate emits an event, it will be saved to the event store and then published to a Kafka topic. Consumers on the read side will subscribe to this topic to update their respective read models.

Architectural Overview with Spring Boot & Kafka

Let's visualize the architecture:

Loading Chart...

Components:

Client: Initiates commands and queries.
Command Gateway: Entry point for commands. Dispatches commands to CommandHandlers.
Command Handler: Receives a command, loads the aggregate from the EventStore by replaying events, applies the command to the aggregate, saves new events to the EventStore, and publishes them to Kafka.
Aggregate: Encapsulates business logic and state. Applies events to itself.
Event Store: Persists all events as the single source of truth (e.g., PostgreSQL table).
Kafka: The event bus.
Event Processor (Projector/Denormalizer): Subscribes to Kafka topics, consumes events, and updates the ReadModelDB.
Read Model DB: Denormalized database optimized for queries (e.g., another PostgreSQL table, MongoDB).
Query Service: Exposes REST APIs to retrieve data from the ReadModelDB.

Implementing the Command Side (Write Model) with Spring Boot

Let's set up a simple Account service where we can deposit money.

Dependencies (build.gradle):

dependencies {
    implementation 'org.springframework.boot:spring-boot-starter-web'
    implementation 'org.springframework.boot:spring-boot-starter-data-jpa'
    implementation 'org.springframework.kafka:spring-kafka'
    runtimeOnly 'org.postgresql:postgresql'
    compileOnly 'org.projectlombok:lombok'
    annotationProcessor 'org.projectlombok:lombok'
    testImplementation 'org.springframework.boot:spring-boot-starter-test'
    testImplementation 'org.springframework.kafka:spring-kafka-test'
}

1. Events

Events are immutable facts. We'll use a simple Event interface and concrete implementations.

// src/main/java/com/example/eventsourcingcqrs/events/Event.java
package com.example.eventsourcingcqrs.events;

import java.time.LocalDateTime;
import java.util.UUID;

public interface Event {
    UUID getId();
    UUID getAggregateId();
    LocalDateTime getTimestamp();
    int getVersion();
    String getType();
}

// src/main/java/com/example/eventsourcingcqrs/events/AccountCreatedEvent.java
package com.example.eventsourcingcqrs.events;

import lombok.Value;
import java.time.LocalDateTime;
import java.util.UUID;

@Value
public class AccountCreatedEvent implements Event {
    UUID id = UUID.randomUUID();
    UUID aggregateId;
    LocalDateTime timestamp = LocalDateTime.now();
    int version;
    String owner;
    double initialBalance;

    @Override
    public String getType() {
        return "AccountCreatedEvent";
    }
}

// src/main/java/com/example/eventsourcingcqrs/events/AccountCreditedEvent.java
package com.example.eventsourcingcqrs.events;

import lombok.Value;
import java.time.LocalDateTime;
import java.util.UUID;

@Value
public class AccountCreditedEvent implements Event {
    UUID id = UUID.randomUUID();
    UUID aggregateId;
    LocalDateTime timestamp = LocalDateTime.now();
    int version;
    double amount;
    String transactionId;

    @Override
    public String getType() {
        return "AccountCreditedEvent";
    }
}

2. Commands

Commands represent an intent to change the system state.

// src/main/java/com/example/eventsourcingcqrs/commands/CreateAccountCommand.java
package com.example.eventsourcingcqrs.commands;

import lombok.Value;
import java.util.UUID;

@Value
public class CreateAccountCommand {
    UUID accountId;
    String owner;
    double initialBalance;
}

// src/main/java/com/example/eventsourcingcqrs/commands/DepositMoneyCommand.java
package com.example.eventsourcingcqrs.commands;

import lombok.Value;
import java.util.UUID;

@Value
public class DepositMoneyCommand {
    UUID accountId;
    double amount;
    UUID transactionId;
}

3. Aggregate Root

Our Account aggregate will handle commands and emit events.

// src/main/java/com/example/eventsourcingcqrs/domain/AccountAggregate.java
package com.example.eventsourcingcqrs.domain;

import com.example.eventsourcingcqrs.events.AccountCreatedEvent;
import com.example.eventsourcingcqrs.events.AccountCreditedEvent;
import com.example.eventsourcingcqrs.events.Event;
import lombok.Getter;
import lombok.NoArgsConstructor;
import lombok.Setter;

import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

@Getter
@Setter
@NoArgsConstructor
public class AccountAggregate {
    private UUID id;
    private String owner;
    private double balance;
    private int version = -1; // -1 indicates a new aggregate
    private final List<Event> uncommittedEvents = new ArrayList<>();

    public AccountAggregate(UUID id, String owner, double initialBalance) {
        AccountCreatedEvent event = new AccountCreatedEvent(id, 0, owner, initialBalance);
        apply(event);
        uncommittedEvents.add(event);
    }

    // Apply method to reconstruct state from events
    public void apply(Event event) {
        if (event instanceof AccountCreatedEvent) {
            apply((AccountCreatedEvent) event);
        } else if (event instanceof AccountCreditedEvent) {
            apply((AccountCreditedEvent) event);
        }
        this.version = event.getVersion();
    }

    private void apply(AccountCreatedEvent event) {
        this.id = event.getAggregateId();
        this.owner = event.getOwner();
        this.balance = event.getInitialBalance();
        this.version = event.getVersion();
    }

    private void apply(AccountCreditedEvent event) {
        this.balance += event.getAmount();
        this.version = event.getVersion();
    }

    // Command handlers
    public void createAccount(UUID accountId, String owner, double initialBalance) {
        if (this.id != null) {
            throw new IllegalStateException("Account already exists");
        }
        AccountCreatedEvent event = new AccountCreatedEvent(accountId, this.version + 1, owner, initialBalance);
        apply(event);
        uncommittedEvents.add(event);
    }

    public void depositMoney(UUID transactionId, double amount) {
        if (this.id == null) {
            throw new IllegalStateException("Account does not exist");
        }
        if (amount <= 0) {
            throw new IllegalArgumentException("Deposit amount must be positive");
        }
        AccountCreditedEvent event = new AccountCreditedEvent(this.id, this.version + 1, amount, transactionId.toString());
        apply(event);
        uncommittedEvents.add(event);
    }

    public static AccountAggregate loadFromHistory(UUID aggregateId, List<Event> history) {
        AccountAggregate aggregate = new AccountAggregate();
        for (Event event : history) {
            aggregate.apply(event);
        }
        return aggregate;
    }
}

4. Event Store (Persistence)

We'll store events in a PostgreSQL table. Each row represents an event.

// src/main/java/com/example/eventsourcingcqrs/eventstore/StoredEvent.java
package com.example.eventsourcingcqrs.eventstore;

import jakarta.persistence.Column;
import jakarta.persistence.Entity;
import jakarta.persistence.Id;
import jakarta.persistence.Lob;
import jakarta.persistence.Table;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.time.LocalDateTime;
import java.util.UUID;

@Entity
@Table(name = "event_store")
@Data
@NoArgsConstructor
@AllArgsConstructor
public class StoredEvent {
    @Id
    private UUID eventId;
    private UUID aggregateId;
    private LocalDateTime timestamp;
    private String eventType;
    private int version;
    @Lob // For larger JSON payloads
    @Column(columnDefinition = "TEXT")
    private String eventData;
}

// src/main/java/com/example/eventsourcingcqrs/eventstore/EventStoreRepository.java
package com.example.eventsourcingcqrs.eventstore;

import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository;

import java.util.List;
import java.util.UUID;

@Repository
public interface EventStoreRepository extends JpaRepository<StoredEvent, UUID> {
    List<StoredEvent> findByAggregateIdOrderByVersionAsc(UUID aggregateId);
}

5. Event Publisher (Kafka Producer)

Publishes events to Kafka after saving them to the event store.

// src/main/java/com/example/eventsourcingcqrs/kafka/EventProducer.java
package com.example.eventsourcingcqrs.kafka;

import com.example.eventsourcingcqrs.events.Event;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
@RequiredArgsConstructor
@Slf4j
public class EventProducer {
    private final KafkaTemplate<String, String> kafkaTemplate;
    private final ObjectMapper objectMapper;

    public EventProducer() {
        this.kafkaTemplate = null; // Will be injected by Spring
        this.objectMapper = new ObjectMapper();
        this.objectMapper.registerModule(new JavaTimeModule());
        this.objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
    }

    public void publish(String topic, Event event) {
        try {
            String eventPayload = objectMapper.writeValueAsString(event);
            kafkaTemplate.send(topic, event.getAggregateId().toString(), eventPayload);
            log.info("Published event {} to topic {}: {}", event.getType(), topic, eventPayload);
        } catch (JsonProcessingException e) {
            log.error("Error serializing event to JSON: {}", e.getMessage());
            throw new RuntimeException("Failed to serialize event", e);
        }
    }
}

6. Command Handler Service

Orchestrates loading aggregates, applying commands, saving events, and publishing.

// src/main/java/com/example/eventsourcingcqrs/command/AccountCommandService.java
package com.example.eventsourcingcqrs.command;

import com.example.eventsourcingcqrs.commands.CreateAccountCommand;
import com.example.eventsourcingcqrs.commands.DepositMoneyCommand;
import com.example.eventsourcingcqrs.domain.AccountAggregate;
import com.example.eventsourcingcqrs.events.Event;
import com.example.eventsourcingcqrs.eventstore.EventStoreRepository;
import com.example.eventsourcingcqrs.eventstore.StoredEvent;
import com.example.eventsourcingcqrs.kafka.EventProducer;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;
import java.util.UUID;
import java.util.stream.Collectors;

@Service
@RequiredArgsConstructor
@Slf4j
public class AccountCommandService {
    private final EventStoreRepository eventStoreRepository;
    private final EventProducer eventProducer;
    private final ObjectMapper objectMapper;

    public AccountCommandService(EventStoreRepository eventStoreRepository, EventProducer eventProducer) {
        this.eventStoreRepository = eventStoreRepository;
        this.eventProducer = eventProducer;
        this.objectMapper = new ObjectMapper();
        this.objectMapper.registerModule(new JavaTimeModule());
        this.objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
    }

    @Transactional
    public UUID handle(CreateAccountCommand command) {
        List<Event> history = loadEventsForAggregate(command.getAccountId());
        AccountAggregate aggregate;
        if (history.isEmpty()) {
            aggregate = new AccountAggregate(); // Create new aggregate if no history
        }
        // Reconstruct aggregate state from history
        aggregate = AccountAggregate.loadFromHistory(command.getAccountId(), history);

        aggregate.createAccount(command.getAccountId(), command.getOwner(), command.getInitialBalance());

        saveAndPublishEvents(command.getAccountId(), aggregate.getUncommittedEvents());
        return aggregate.getId();
    }

    @Transactional
    public void handle(DepositMoneyCommand command) {
        List<Event> history = loadEventsForAggregate(command.getAccountId());
        if (history.isEmpty()) {
            throw new IllegalStateException("Account not found: " + command.getAccountId());
        }
        AccountAggregate aggregate = AccountAggregate.loadFromHistory(command.getAccountId(), history);

        aggregate.depositMoney(command.getTransactionId(), command.getAmount());

        saveAndPublishEvents(command.getAccountId(), aggregate.getUncommittedEvents());
    }

    private List<Event> loadEventsForAggregate(UUID aggregateId) {
        return eventStoreRepository.findByAggregateIdOrderByVersionAsc(aggregateId).stream()
                .map(this::deserializeEvent)
                .collect(Collectors.toList());
    }

    private Event deserializeEvent(StoredEvent storedEvent) {
        try {
            // Dynamically determine event type and deserialize
            String eventType = storedEvent.getEventType();
            Class<? extends Event> eventClass = (Class<? extends Event>) Class.forName("com.example.eventsourcingcqrs.events." + eventType);
            return objectMapper.readValue(storedEvent.getEventData(), eventClass);
        } catch (Exception e) {
            log.error("Error deserializing event {}: {}", storedEvent.getEventId(), e.getMessage());
            throw new RuntimeException("Failed to deserialize event", e);
        }
    }

    private void saveAndPublishEvents(UUID aggregateId, List<Event> events) {
        events.forEach(event -> {
            try {
                // Save to event store
                StoredEvent storedEvent = new StoredEvent(
                        event.getId(),
                        event.getAggregateId(),
                        event.getTimestamp(),
                        event.getType(),
                        event.getVersion(),
                        objectMapper.writeValueAsString(event)
                );
                eventStoreRepository.save(storedEvent);

                // Publish to Kafka
                eventProducer.publish("account-events", event);
            } catch (JsonProcessingException e) {
                log.error("Error saving/publishing event {}: {}", event.getId(), e.getMessage());
                throw new RuntimeException("Failed to save or publish event", e);
            }
        });
        events.clear(); // Clear uncommitted events after saving and publishing
    }
}

7. Command Controller

Exposes REST endpoints for commands.

// src/main/java/com/example/eventsourcingcqrs/command/AccountCommandController.java
package com.example.eventsourcingcqrs.command;

import com.example.eventsourcingcqrs.commands.CreateAccountCommand;
import com.example.eventsourcingcqrs.commands.DepositMoneyCommand;
import lombok.RequiredArgsConstructor;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.*;

import java.util.UUID;

@RestController
@RequestMapping("/api/accounts")
@RequiredArgsConstructor
public class AccountCommandController {
    private final AccountCommandService accountCommandService;

    @PostMapping
    public ResponseEntity<UUID> createAccount(@RequestBody CreateAccountCommand command) {
        UUID accountId = accountCommandService.handle(command);
        return new ResponseEntity<>(accountId, HttpStatus.CREATED);
    }

    @PostMapping("/{accountId}/deposit")
    public ResponseEntity<Void> depositMoney(@PathVariable UUID accountId, @RequestBody DepositMoneyCommand command) {
        if (!accountId.equals(command.getAccountId())) {
            return new ResponseEntity<>(HttpStatus.BAD_REQUEST);
        }
        accountCommandService.handle(command);
        return new ResponseEntity<>(HttpStatus.OK);
    }
}

Implementing the Query Side (Read Model) with Spring Boot

Now, let's build the read model, which will consume events from Kafka and update a denormalized view.

1. Read Model Entity and Repository

This entity represents the denormalized state for queries.

// src/main/java/com/example/eventsourcingcqrs/readmodel/AccountReadModel.java
package com.example.eventsourcingcqrs.readmodel;

import jakarta.persistence.Entity;
import jakarta.persistence.Id;
import jakarta.persistence.Table;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

import java.util.UUID;

@Entity
@Table(name = "account_read_model")
@Data
@NoArgsConstructor
@AllArgsConstructor
public class AccountReadModel {
    @Id
    private UUID accountId;
    private String owner;
    private double balance;
    private int version; // To handle out-of-order events or concurrency
}

// src/main/java/com/example/eventsourcingcqrs/readmodel/AccountReadModelRepository.java
package com.example.eventsourcingcqrs.readmodel;

import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.stereotype.Repository;

import java.util.Optional;
import java.util.UUID;

@Repository
public interface AccountReadModelRepository extends JpaRepository<AccountReadModel, UUID> {
    Optional<AccountReadModel> findByAccountId(UUID accountId);
}

2. Event Consumer (Projector)

This component listens to Kafka events and updates the AccountReadModel.

// src/main/java/com/example/eventsourcingcqrs/readmodel/AccountProjector.java
package com.example.eventsourcingcqrs.readmodel;

import com.example.eventsourcingcqrs.events.AccountCreatedEvent;
import com.example.eventsourcingcqrs.events.AccountCreditedEvent;
import com.example.eventsourcingcqrs.events.Event;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

@Component
@RequiredArgsConstructor
@Slf4j
public class AccountProjector {
    private final AccountReadModelRepository accountReadModelRepository;
    private final ObjectMapper objectMapper;

    public AccountProjector(AccountReadModelRepository accountReadModelRepository) {
        this.accountReadModelRepository = accountReadModelRepository;
        this.objectMapper = new ObjectMapper();
        this.objectMapper.registerModule(new JavaTimeModule());
        this.objectMapper.disable(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS);
    }

    @KafkaListener(topics = "account-events", groupId = "account-read-model-group")
    @Transactional
    public void handleAccountEvent(String eventPayload) {
        try {
            // A more robust solution would involve a custom deserializer or event type header
            // For simplicity, we'll try to deserialize to common event types
            Event event;
            if (eventPayload.contains("AccountCreatedEvent")) {
                event = objectMapper.readValue(eventPayload, AccountCreatedEvent.class);
            } else if (eventPayload.contains("AccountCreditedEvent")) {
                event = objectMapper.readValue(eventPayload, AccountCreditedEvent.class);
            } else {
                log.warn("Unknown event type received: {}", eventPayload);
                return;
            }

            log.info("Processing event: {}", event.getType());

            // Idempotency check: only process if event version is greater than current read model version
            AccountReadModel existingReadModel = accountReadModelRepository.findByAccountId(event.getAggregateId()).orElse(null);
            if (existingReadModel != null && event.getVersion() <= existingReadModel.getVersion()) {
                log.warn("Skipping older or duplicate event {}: aggregateId={}, version={}", event.getType(), event.getAggregateId(), event.getVersion());
                return;
            }

            if (event instanceof AccountCreatedEvent createdEvent) {
                AccountReadModel newReadModel = new AccountReadModel(
                        createdEvent.getAggregateId(),
                        createdEvent.getOwner(),
                        createdEvent.getInitialBalance(),
                        createdEvent.getVersion()
                );
                accountReadModelRepository.save(newReadModel);
            } else if (event instanceof AccountCreditedEvent creditedEvent) {
                AccountReadModel readModel = accountReadModelRepository.findByAccountId(creditedEvent.getAggregateId())
                        .orElseThrow(() -> new IllegalStateException("Read model for account not found: " + creditedEvent.getAggregateId()));
                readModel.setBalance(readModel.getBalance() + creditedEvent.getAmount());
                readModel.setVersion(creditedEvent.getVersion());
                accountReadModelRepository.save(readModel);
            }
        } catch (Exception e) {
            log.error("Error processing Kafka event: {}", e.getMessage(), e);
            // In a real application, consider dead-letter queues or retry mechanisms
            throw new RuntimeException("Failed to process event", e);
        }
    }
}

3. Query Service

Exposes read-only API endpoints.

// src/main/java/com/example/eventsourcingcqrs/readmodel/AccountQueryService.java
package com.example.eventsourcingcqrs.readmodel;

import lombok.RequiredArgsConstructor;
import org.springframework.stereotype.Service;

import java.util.Optional;
import java.util.UUID;

@Service
@RequiredArgsConstructor
public class AccountQueryService {
    private final AccountReadModelRepository accountReadModelRepository;

    public Optional<AccountReadModel> getAccountById(UUID accountId) {
        return accountReadModelRepository.findByAccountId(accountId);
    }
}

4. Query Controller

REST API for querying account data.

// src/main/java/com/example/eventsourcingcqrs/readmodel/AccountQueryController.java
package com.example.eventsourcingcqrs.readmodel;

import lombok.RequiredArgsConstructor;
import org.springframework.http.HttpStatus;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.PathVariable;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

import java.util.UUID;

@RestController
@RequestMapping("/api/query/accounts")
@RequiredArgsConstructor
public class AccountQueryController {
    private final AccountQueryService accountQueryService;

    @GetMapping("/{accountId}")
    public ResponseEntity<AccountReadModel> getAccount(@PathVariable UUID accountId) {
        return accountQueryService.getAccountById(accountId)
                .map(account -> new ResponseEntity<>(account, HttpStatus.OK))
                .orElse(new ResponseEntity<>(HttpStatus.NOT_FOUND));
    }
}

Integrating Spring Boot with Kafka

To make the Spring Boot services communicate with Kafka, you need to configure application.yml (or application.properties).

# src/main/resources/application.yml
spring:
  application:
    name: event-sourcing-cqrs-app
  kafka:
    bootstrap-servers: localhost:9092 # Or your Kafka cluster address
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
    consumer:
      group-id: account-read-model-group
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      auto-offset-reset: earliest # Start reading from the beginning if no offset is committed
  datasource:
    url: jdbc:postgresql://localhost:5432/eventstore_db
    username: user
    password: password
    driver-class-name: org.postgresql.Driver
  jpa:
    hibernate:
      ddl-auto: update # In production, use 'none' and manage schema migrations manually
    show-sql: true
    properties:
      hibernate:
        format_sql: true

Also, ensure you have Kafka and PostgreSQL running, e.g., via Docker:

# docker-compose.yml
version: '3.8'
services:
  zookeeper:
    image: 'bitnami/zookeeper:latest'
    ports:
      - '2181:2181'
    environment:
      - ALLOW_ANONYMOUS_LOGIN=yes
  kafka:
    image: 'bitnami/kafka:latest'
    ports:
      - '9092:9092'
    environment:
      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_CFG_LISTENERS=PLAINTEXT://:9092
      - KAFKA_CFG_ADVERTISED_LISTENERS=PLAINTEXT://localhost:9092
      - ALLOW_PLAINTEXT_LISTENER=yes
    depends_on:
      - zookeeper
  postgres:
    image: 'postgres:15'
    environment:
      POSTGRES_DB: eventstore_db
      POSTGRES_USER: user
      POSTGRES_PASSWORD: password
    ports:
      - '5432:5432'
    volumes:
      - pg_data:/var/lib/postgresql/data

volumes:
  pg_data:

Run with docker-compose up -d.

Best Practices for ES/CQRS with Spring Boot & Kafka

Event Immutability and Versioning: Events are facts and should never change. For evolving event schemas, use event versioning (e.g., AccountCreatedEventV1, AccountCreatedEventV2) and migration strategies in your projectors.
Idempotency in Event Consumers: Event consumers must be idempotent. If an event is processed multiple times (due to retries or network issues), it should not lead to incorrect state changes. Our AccountProjector includes a basic version check for this.
Error Handling (Dead-Letter Queues): Implement robust error handling for Kafka consumers. Events that fail processing should be moved to a Dead-Letter Queue (DLQ) for later inspection and reprocessing, preventing consumer group blockage.
Transaction Management: Ensure that saving events to the event store and publishing to Kafka are part of a single, atomic transaction (e.g., using Spring's @Transactional for JPA and an outbox pattern for Kafka publishing).
Snapshotting for Large Aggregates: For aggregates with a very long history of events, replaying all events to reconstruct state can be slow. Implement snapshotting to periodically save the aggregate's current state, reducing replay time.
Consistency Models: Embrace eventual consistency for the read model. Understand that there will be a small delay between the write model update and the read model reflecting that update. Design your UI/UX accordingly.
Monitoring and Observability: Instrument your services with logging, metrics (e.g., Micrometer, Prometheus), and tracing (e.g., Sleuth, Zipkin) to monitor event flow, processing times, and potential bottlenecks.
Bounded Contexts: Apply Domain-Driven Design principles. Each microservice should ideally correspond to a bounded context, with its own aggregates, events, and read models.

Common Pitfalls to Avoid

Over-engineering: Event Sourcing and CQRS add complexity. Don't apply them to every service or every part of your application. Use them where the benefits (auditability, scalability, complex domain) outweigh the overhead.
Tight Coupling Between Read/Write Models: The read model should be completely decoupled from the write model's internal structure. Changes in the write model should only affect the read model via events.
Ignoring Eventual Consistency: Failing to design for eventual consistency can lead to confusing user experiences or incorrect assumptions about data freshness.
Complex Event Schemas Without Versioning: Events are your public contract. Treat them with care. Without proper versioning, evolving your domain can become a nightmare.
Performance Bottlenecks in Event Processing: If your projectors can't keep up with the incoming event stream, your read models will fall behind. Monitor consumer lag and scale your consumers as needed.
Not Handling Out-of-Order Events: While Kafka guarantees order within a partition, multiple partitions or consumer retries can lead to events being processed out of order. Implement version checks or other mechanisms to handle this (as shown in our projector).

Real-world Use Cases

Event Sourcing and CQRS are particularly well-suited for domains requiring high auditability, complex business logic, or extreme scalability:

E-commerce: Order processing, inventory management, shopping cart history.
Financial Systems: Transaction processing, ledger systems, fraud detection, compliance audit trails.
IoT Platforms: Processing vast streams of sensor data, device state management.
Gaming: Player actions, game state changes, leaderboards.
Supply Chain Management: Tracking goods, status updates, logistics events.

Conclusion

Event Sourcing and CQRS, when implemented thoughtfully with Spring Boot and Apache Kafka, offer a powerful paradigm for building modern, scalable, and resilient microservices. They provide unparalleled auditability, flexibility in evolving domain models, and the ability to optimize read and write concerns independently.

While these patterns introduce complexity, the benefits for specific problem domains are immense. By understanding the core concepts, adhering to best practices, and leveraging robust tools like Spring Boot and Kafka, you can architect systems that are not only performant and scalable but also deeply insightful into their own history and evolution.

Start experimenting with these patterns in a controlled environment, perhaps with a less critical bounded context, to gain experience before applying them to core business functionalities. The journey into event-driven architectures is rewarding and will undoubtedly equip you with advanced tools for tackling the challenges of distributed systems.